feat: add ICT gateway login and heartbeat CLI

.gitignore

@@ -0,0 +1,6 @@
+__pycache__/
+.pytest_cache/
+.ruff_cache/
+*.egg-info/
+build/
+dist/

README.md

@@ -0,0 +1,71 @@
+# ICT 网关命令行登录工具
+
+这是针对 `https://gw.ict.ac.cn`（SRun Portal）的服务器命令行客户端。它复刻网页端的挑战码、HMAC-MD5、XXTEA、定制 Base64 和 SHA-1 校验流程，不依赖第三方 Python 包，也不会保存密码。
+
+## 安装
+
+在服务器上进入本目录后执行：
+
+```bash
+python3 -m pip install --user --no-deps --no-build-isolation .
+```
+
+若 `ictlogin` 不在当前 shell 的 `PATH` 中，先执行 `export PATH="$HOME/.local/bin:$PATH"`，或直接使用 `~/.local/bin/ictlogin`。
+
+如果不想安装，也可以直接运行 `python3 run_ictlogin.py ...`。
+
+## 使用
+
+先检查当前 IP：
+
+```bash
+ictlogin status
+```
+
+交互式登录（账号和密码不会出现在 shell 历史或进程列表中）：
+
+```bash
+ictlogin login
+```
+
+也可以只把账号放在参数中，密码仍会安全地提示输入：
+
+```bash
+ictlogin login --username YOUR_ACCOUNT
+```
+
+注销和 JSON 状态输出：
+
+```bash
+ictlogin logout
+ictlogin status --json
+```
+
+保持服务器会话活跃：
+
+```bash
+ictlogin heartbeat --interval 60
+```
+
+该命令会周期性请求网关的在线接口并持续运行，按 `Ctrl-C` 停止；先用 `--once` 做单次检查。若希望掉线后自动重新认证，必须显式开启并提供账号密码：
+
+```bash
+ictlogin heartbeat --auto-login --username YOUR_ACCOUNT --password-env ICT_PASSWORD
+```
+
+每次连续掉线期间只尝试一次自动认证，恢复在线后才会重置，避免错误密码造成反复锁定。
+
+脚本场景可以使用标准输入或显式环境变量读取密码，避免使用 `--password` 这种会暴露在进程列表中的参数：
+
+```bash
+printf '%s\n' "$ICT_PASSWORD" | ictlogin login --username YOUR_ACCOUNT --password-stdin
+ictlogin login --username YOUR_ACCOUNT --password-env ICT_PASSWORD
+```
+
+若网关要求图片验证码，CLI 会将 PNG 保存到 `~/.cache/ictlogin/`（目录和文件权限为当前用户私有），然后提示输入验证码。可用 `--ip`、`--ac-id` 和 `--base-url` 覆盖自动探测的连接参数。
+
+## 安全说明
+
+- 密码只在内存中用于生成认证请求，程序不会写入配置文件、日志或命令行参数。
+- 默认校验 HTTPS 证书；只有在明确确认网关证书问题时才使用 `--insecure`。
+- 工具每次登录只发送一次认证请求，不会重试密码，避免触发账号锁定。

pyproject.toml

@@ -0,0 +1,22 @@
+[build-system]
+requires = ["setuptools>=61"]
+build-backend = "setuptools.build_meta"
+
+[project]
+name = "ictlogin"
+version = "0.2.0"
+description = "Command-line client for the ICT SRun network gateway"
+readme = "README.md"
+requires-python = ">=3.8"
+license = {text = "MIT"}
+authors = [{name = "ICT login CLI contributors"}]
+dependencies = []
+
+[project.scripts]
+ictlogin = "ictlogin.cli:main"
+
+[tool.setuptools]
+package-dir = {"" = "src"}
+
+[tool.setuptools.packages.find]
+where = ["src"]

run_ictlogin.py

@@ -0,0 +1,17 @@
+#!/usr/bin/env python3
+"""Run the ICT login CLI directly from a source checkout."""
+
+from pathlib import Path
+import sys
+
+
+ROOT = Path(__file__).resolve().parent
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from ictlogin.cli import main  # noqa: E402
+
+
+if __name__ == "__main__":
+    main()

setup.py

@@ -0,0 +1,12 @@
+from setuptools import find_packages, setup
+
+
+setup(
+    name="ictlogin",
+    version="0.2.0",
+    description="Command-line client for the ICT SRun network gateway",
+    packages=find_packages("src"),
+    package_dir={"": "src"},
+    python_requires=">=3.8",
+    entry_points={"console_scripts": ["ictlogin=ictlogin.cli:main"]},
+)

src/ictlogin/__init__.py

@@ -0,0 +1,4 @@
+"""ICT SRun gateway command-line client."""
+
+__all__ = ["__version__"]
+__version__ = "0.2.0"

src/ictlogin/__main__.py

@@ -0,0 +1,5 @@
+from .cli import main
+
+
+if __name__ == "__main__":
+    main()

src/ictlogin/cli.py

@@ -0,0 +1,442 @@
+"""Argument parsing and interactive flows for the ICT login CLI."""
+
+from __future__ import annotations
+
+import argparse
+import getpass
+import json
+import math
+import os
+import sys
+import time
+from datetime import datetime
+from typing import Optional
+
+from . import __version__
+from .client import GatewayClient, GatewayError
+
+
+def _normalize_domain(value: Optional[str]) -> str:
+    if not value:
+        return ""
+    value = value.strip()
+    if not value:
+        return ""
+    return value if value.startswith("@") else "@" + value
+
+
+def _add_connection_options(command: argparse.ArgumentParser) -> None:
+    """Allow connection flags both before and after the subcommand."""
+
+    command.add_argument(
+        "--base-url",
+        default=argparse.SUPPRESS,
+        help="网关地址",
+    )
+    command.add_argument("--ac-id", default=argparse.SUPPRESS, help="认证区域 ID")
+    command.add_argument("--ip", default=argparse.SUPPRESS, help="网关看到的服务器 IP")
+    command.add_argument(
+        "--timeout", type=float, default=argparse.SUPPRESS, help="单次请求超时秒数"
+    )
+    command.add_argument(
+        "--insecure",
+        "--no-verify-tls",
+        action="store_true",
+        default=argparse.SUPPRESS,
+        help="跳过 TLS 证书校验",
+    )
+    command.add_argument(
+        "--verbose",
+        action="store_true",
+        default=argparse.SUPPRESS,
+        help="输出请求路径诊断信息",
+    )
+
+
+def _build_parser() -> argparse.ArgumentParser:
+    parser = argparse.ArgumentParser(
+        prog="ictlogin",
+        description="中科院计算所网关（SRun Portal）命令行登录工具",
+    )
+    parser.add_argument("--version", action="version", version=__version__)
+    parser.add_argument(
+        "--base-url",
+        default=os.environ.get("ICTLOGIN_BASE_URL", "https://gw.ict.ac.cn"),
+        help="网关地址（默认 https://gw.ict.ac.cn）",
+    )
+    parser.add_argument(
+        "--ac-id",
+        default=os.environ.get("ICTLOGIN_AC_ID", "1"),
+        help="认证区域 ID（默认 1）",
+    )
+    parser.add_argument(
+        "--ip",
+        default=os.environ.get("ICTLOGIN_IP"),
+        help="网关看到的服务器 IP；默认自动探测",
+    )
+    parser.add_argument(
+        "--timeout", type=float, default=15.0, help="单次请求超时秒数（默认 15）"
+    )
+    parser.add_argument(
+        "--insecure",
+        "--no-verify-tls",
+        action="store_true",
+        help="跳过 TLS 证书校验（仅在网关证书异常时使用）",
+    )
+    parser.add_argument("--verbose", action="store_true", help="输出请求路径诊断信息")
+
+    commands = parser.add_subparsers(dest="command", required=True)
+
+    status = commands.add_parser("status", help="查询当前 IP 的在线状态")
+    _add_connection_options(status)
+    status.add_argument("--json", action="store_true", help="以 JSON 输出完整响应")
+
+    login = commands.add_parser("login", help="交互式登录网关")
+    _add_connection_options(login)
+    login.add_argument("-u", "--username", help="账号；省略时交互输入")
+    login.add_argument("--domain", help="账号域，例如 @cmcc")
+    login.add_argument(
+        "--password-stdin",
+        action="store_true",
+        help="从标准输入读取一行密码（适合脚本，不要把密码放在命令行）",
+    )
+    login.add_argument(
+        "--password-env",
+        metavar="NAME",
+        help="从指定环境变量读取密码",
+    )
+    login.add_argument("--captcha", help="已知验证码；需要验证码时省略则交互提示")
+
+    logout = commands.add_parser("logout", help="注销当前 IP")
+    _add_connection_options(logout)
+    logout.add_argument("-u", "--username", help="账号；在线时省略可自动获取")
+    logout.add_argument("--domain", help="账号域，例如 @cmcc")
+
+    heartbeat = commands.add_parser(
+        "heartbeat",
+        aliases=["keepalive"],
+        help="周期性探测在线状态并保持会话",
+    )
+    _add_connection_options(heartbeat)
+    heartbeat.add_argument("-u", "--username", help="自动重登时使用的账号")
+    heartbeat.add_argument("--domain", help="账号域，例如 @cmcc")
+    heartbeat.add_argument(
+        "--interval",
+        type=float,
+        default=60.0,
+        help="心跳间隔秒数（默认 60）",
+    )
+    heartbeat.add_argument(
+        "--once",
+        action="store_true",
+        help="只执行一次探测后退出",
+    )
+    heartbeat.add_argument(
+        "--auto-login",
+        action="store_true",
+        help="掉线时自动重新认证（需提供密码）",
+    )
+    heartbeat.add_argument(
+        "--password-stdin",
+        action="store_true",
+        help="自动重登时从标准输入读取一行密码",
+    )
+    heartbeat.add_argument(
+        "--password-env",
+        metavar="NAME",
+        help="自动重登时从指定环境变量读取密码",
+    )
+    heartbeat.add_argument("--captcha", help="自动重登时使用的验证码")
+
+    return parser
+
+
+def _client(args: argparse.Namespace) -> GatewayClient:
+    return GatewayClient(
+        args.base_url,
+        ac_id=args.ac_id,
+        timeout=args.timeout,
+        insecure=args.insecure,
+        verbose=args.verbose,
+    )
+
+
+def _resolve_ip(client: GatewayClient, configured: Optional[str]) -> str:
+    if configured:
+        return configured.strip()
+    ip = client.discover_ip()
+    print(f"检测到服务器 IP：{ip}", file=sys.stderr)
+    return ip
+
+
+def _read_password(args: argparse.Namespace) -> str:
+    if args.password_stdin and args.password_env:
+        raise GatewayError("--password-stdin 与 --password-env 不能同时使用")
+    if args.password_stdin:
+        value = sys.stdin.readline().rstrip("\r\n")
+    elif args.password_env:
+        value = os.environ.get(args.password_env, "")
+        if not value:
+            raise GatewayError(f"环境变量 {args.password_env} 未设置或为空")
+    else:
+        value = getpass.getpass("密码：")
+    if not value:
+        raise GatewayError("密码不能为空")
+    return value
+
+
+def _read_username(value: Optional[str]) -> str:
+    username = (value or os.environ.get("ICTLOGIN_USERNAME") or "").strip()
+    if not username:
+        username = input("账号：").strip()
+    if not username:
+        raise GatewayError("账号不能为空")
+    return username
+
+
+def _status_command(args: argparse.Namespace) -> int:
+    client = _client(args)
+    status = client.status(args.ip)
+    if args.json:
+        print(json.dumps(status.payload, ensure_ascii=False, sort_keys=True))
+        return 0
+    if not status.online:
+        print("离线")
+        return 0
+    payload = status.payload
+    print("在线")
+    for key, label in (
+        ("user_name", "账号"),
+        ("client_ip", "客户端 IP"),
+        ("online_ip", "在线 IP"),
+        ("products_name", "套餐"),
+        ("sum_bytes", "已用流量"),
+        ("sum_seconds", "已用时长（秒）"),
+    ):
+        value = payload.get(key)
+        if value not in (None, ""):
+            print(f"{label}：{value}")
+    return 0
+
+
+def _captcha_for_login(
+    client: GatewayClient, username: str, ip: str, supplied: Optional[str]
+):
+    if not client.check_captcha(username, ip):
+        return "", supplied or ""
+    captcha_id, image = client.fetch_captcha()
+    path = client.save_captcha(image)
+    print(f"需要验证码，图片已保存到：{path}", file=sys.stderr)
+    value = supplied or input("验证码：").strip()
+    if not value:
+        raise GatewayError("验证码不能为空")
+    return captcha_id, value
+
+
+def _login_command(args: argparse.Namespace) -> int:
+    client = _client(args)
+    username = _read_username(args.username)
+    domain = _normalize_domain(args.domain)
+    ip = _resolve_ip(client, args.ip)
+    password = _read_password(args)
+    captcha_id, captcha_value = _captcha_for_login(
+        client, username + domain, ip, args.captcha
+    )
+    client.authenticate(
+        username=username,
+        domain=domain,
+        password=password,
+        ip=ip,
+        captcha_id=captcha_id,
+        captcha_value=captcha_value,
+    )
+    verified = client.wait_for_status(ip, online=True)
+    if verified.online:
+        print(f"登录成功：{username + domain}（{ip}）")
+    else:
+        print(
+            "网关已接受登录请求，但暂未查询到在线状态；请稍后运行 ictlogin status",
+            file=sys.stderr,
+        )
+    return 0
+
+
+def _logout_command(args: argparse.Namespace) -> int:
+    client = _client(args)
+    ip = _resolve_ip(client, args.ip)
+    current = client.status(ip)
+    if not current.online:
+        print("当前 IP 已离线")
+        return 0
+    payload = current.payload
+    inferred = not bool(args.username)
+    username = (args.username or str(payload.get("user_name") or "")).strip()
+    if not username:
+        raise GatewayError("在线状态没有返回账号，请用 --username 指定")
+    domain = _normalize_domain(args.domain)
+    if inferred and not domain:
+        domain_value = str(payload.get("domain") or "").strip()
+        domain = _normalize_domain(domain_value)
+    username += domain
+    client.logout(username=username, ip=ip)
+    verified = client.wait_for_status(ip, online=False)
+    if verified.online:
+        print("注销请求已发送，但暂未确认离线", file=sys.stderr)
+        return 1
+    print(f"注销成功：{username}（{ip}）")
+    return 0
+
+
+def _heartbeat_stamp() -> str:
+    return datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S%z")
+
+
+def _print_heartbeat(result, ip: str) -> None:
+    status = result.status
+    if status.online:
+        keepalive = (
+            f" keepalive_time={result.keepalive_time}"
+            if result.keepalive_time is not None
+            else ""
+        )
+        print(
+            f"[{_heartbeat_stamp()}] 心跳正常：在线 IP={ip} "
+            f"RTT={result.latency_seconds * 1000:.0f}ms{keepalive}",
+            flush=True,
+        )
+    else:
+        print(f"[{_heartbeat_stamp()}] 心跳响应：离线 IP={ip}", flush=True)
+
+
+def _heartbeat_identity(args: argparse.Namespace, result) -> tuple[str, str]:
+    username = (args.username or "").strip()
+    domain = _normalize_domain(args.domain)
+    if not username and result is not None and result.status.online:
+        payload = result.status.payload
+        username = str(payload.get("user_name") or "").strip()
+        if not domain:
+            domain = _normalize_domain(str(payload.get("domain") or ""))
+    return username, domain
+
+
+def _attempt_relogin(
+    client: GatewayClient,
+    *,
+    username: str,
+    domain: str,
+    password: str,
+    ip: str,
+    captcha: Optional[str],
+) -> bool:
+    captcha_id, captcha_value = _captcha_for_login(
+        client, username + domain, ip, captcha
+    )
+    client.authenticate(
+        username=username,
+        domain=domain,
+        password=password,
+        ip=ip,
+        captcha_id=captcha_id,
+        captcha_value=captcha_value,
+    )
+    return client.wait_for_status(ip, online=True).online
+
+
+def _heartbeat_command(args: argparse.Namespace) -> int:
+    if not math.isfinite(args.interval) or args.interval <= 0:
+        raise GatewayError("--interval 必须是大于 0 的有限秒数")
+
+    client = _client(args)
+    ip = _resolve_ip(client, args.ip)
+    first = None
+    try:
+        first = client.heartbeat(ip)
+    except GatewayError as exc:
+        if args.once:
+            raise
+        print(f"[{_heartbeat_stamp()}] 心跳失败：{exc}", file=sys.stderr, flush=True)
+
+    username, domain = _heartbeat_identity(args, first)
+    password: Optional[str] = None
+    if args.auto_login:
+        if not username:
+            username = _read_username(None)
+        password = _read_password(args)
+
+    relogin_attempted = False
+    while True:
+        if first is not None:
+            result = first
+            first = None
+        else:
+            try:
+                result = client.heartbeat(ip)
+            except GatewayError as exc:
+                print(
+                    f"[{_heartbeat_stamp()}] 心跳失败：{exc}",
+                    file=sys.stderr,
+                    flush=True,
+                )
+                if args.once:
+                    return 1
+                time.sleep(args.interval)
+                continue
+
+        _print_heartbeat(result, ip)
+        relogin_online = False
+        if result.status.online:
+            relogin_attempted = False
+        elif args.auto_login:
+            if not relogin_attempted:
+                relogin_attempted = True
+                print("检测到掉线，尝试自动重新认证（本次掉线只尝试一次）", flush=True)
+                try:
+                    if not _attempt_relogin(
+                        client,
+                        username=username,
+                        domain=domain,
+                        password=password or "",
+                        ip=ip,
+                        captcha=args.captcha,
+                    ):
+                        print("自动重新认证后仍未上线", file=sys.stderr, flush=True)
+                    else:
+                        print("自动重新认证成功", flush=True)
+                        relogin_attempted = False
+                        relogin_online = True
+                except (GatewayError, EOFError) as exc:
+                    print(f"自动重新认证失败：{exc}", file=sys.stderr, flush=True)
+        else:
+            print("当前为离线状态；未启用自动重新认证", file=sys.stderr, flush=True)
+
+        if args.once:
+            return 0 if result.status.online or relogin_online else 1
+        try:
+            time.sleep(args.interval)
+        except KeyboardInterrupt:
+            print("心跳已停止", file=sys.stderr)
+            return 0
+
+
+def main(argv: Optional[list[str]] = None) -> None:
+    parser = _build_parser()
+    args = parser.parse_args(argv)
+    try:
+        if args.command == "status":
+            code = _status_command(args)
+        elif args.command == "login":
+            code = _login_command(args)
+        elif args.command == "logout":
+            code = _logout_command(args)
+        elif args.command in ("heartbeat", "keepalive"):
+            code = _heartbeat_command(args)
+        else:  # pragma: no cover - argparse enforces this
+            parser.error("需要指定命令")
+            return
+    except (GatewayError, ValueError, EOFError, KeyboardInterrupt) as exc:
+        print(f"错误：{exc}", file=sys.stderr)
+        code = 130 if isinstance(exc, KeyboardInterrupt) else 1
+    raise SystemExit(code)
+
+
+__all__ = ["main"]

src/ictlogin/client.py

@@ -0,0 +1,420 @@
+"""HTTP client for the ICT SRun Portal endpoints."""
+
+from __future__ import annotations
+
+import base64
+import json
+import os
+import random
+import re
+import socket
+import ssl
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping, Optional
+
+from .protocol import is_success, login_parameters
+
+
+ERROR_MESSAGES = {
+    "auth_info_error": "账号或密码错误",
+    "login_error": "登录失败",
+    "ip_already_online_error": "本机 IP 已经使用其他账号登录",
+    "not_online_error": "当前设备未在线",
+    "missing_required_parameters_error": "缺少必要参数",
+    "challenge_expire_error": "登录挑战已过期，请重试",
+    "vcode_error": "验证码错误",
+    "speed_limit_error": "认证请求过于频繁，请稍后再试",
+    "account_locked": "账号已被锁定",
+    "user_must_modify_password": "账号需要先修改密码",
+    "no_response_data_error": "认证服务没有返回数据",
+    "memory_db_error": "认证服务暂时无响应",
+    "E2531": "账号或密码错误",
+    "E2533": "密码错误次数过多，请稍后再试",
+    "E2553": "密码错误",
+    "E2620": "已超过允许的在线设备数",
+}
+
+
+class GatewayError(RuntimeError):
+    """An expected gateway or transport failure."""
+
+    def __init__(
+        self,
+        message: str,
+        *,
+        code: str = "",
+        payload: Optional[Mapping[str, Any]] = None,
+        status: Optional[int] = None,
+    ) -> None:
+        super().__init__(message)
+        self.code = code
+        self.payload = dict(payload or {})
+        self.status = status
+
+
+@dataclass(frozen=True)
+class Status:
+    online: bool
+    payload: Mapping[str, Any]
+
+
+@dataclass(frozen=True)
+class Heartbeat:
+    """Result of an active online probe."""
+
+    status: Status
+    latency_seconds: float
+    keepalive_time: Optional[int]
+
+
+def _error_code(payload: Mapping[str, Any]) -> str:
+    value = payload.get("error") or payload.get("res") or payload.get("code") or ""
+    return str(value)
+
+
+def error_message(payload: Mapping[str, Any]) -> str:
+    code = _error_code(payload)
+    detail = str(payload.get("error_msg") or payload.get("message") or "").strip()
+    if code in ERROR_MESSAGES:
+        message = ERROR_MESSAGES[code]
+    elif code:
+        message = code
+    else:
+        message = "网关返回了无法识别的响应"
+    return f"{message}（{detail}）" if detail and detail != code else message
+
+
+class GatewayClient:
+    """Small dependency-free client for the SRun JSON/JSONP API."""
+
+    def __init__(
+        self,
+        base_url: str = "https://gw.ict.ac.cn",
+        *,
+        ac_id: str = "1",
+        timeout: float = 15.0,
+        insecure: bool = False,
+        verbose: bool = False,
+    ) -> None:
+        parsed = urllib.parse.urlsplit(base_url)
+        if parsed.scheme not in ("http", "https") or not parsed.netloc:
+            raise ValueError("base URL 必须是 http(s) URL")
+        self.base_url = base_url.rstrip("/")
+        self.ac_id = str(ac_id)
+        self.timeout = timeout
+        self.verbose = verbose
+        context = (
+            ssl._create_unverified_context()
+            if insecure
+            else ssl.create_default_context()
+        )
+        # The gateway is on the server's private network.  Do not accidentally
+        # route these requests through a stale HTTP(S)_PROXY environment value.
+        self._opener = urllib.request.build_opener(
+            urllib.request.ProxyHandler({}),
+            urllib.request.HTTPCookieProcessor(),
+            urllib.request.HTTPSHandler(context=context),
+        )
+
+    def _url(self, path: str, params: Optional[Mapping[str, Any]] = None) -> str:
+        if not path.startswith("/"):
+            path = "/" + path
+        url = self.base_url + path
+        if params:
+            filtered = [
+                (key, value) for key, value in params.items() if value is not None
+            ]
+            if filtered:
+                url += "?" + urllib.parse.urlencode(filtered)
+        return url
+
+    def _request(
+        self,
+        path: str,
+        params: Optional[Mapping[str, Any]] = None,
+        *,
+        jsonp: bool = False,
+    ) -> Any:
+        callback = ""
+        request_params = dict(params or {})
+        if jsonp:
+            callback = "ictlogin_cb_" + f"{random.getrandbits(48):012x}"
+            request_params["callback"] = callback
+        url = self._url(path, request_params)
+        if self.verbose:
+            print(f"请求 {path}", file=os.sys.stderr)
+        request = urllib.request.Request(
+            url,
+            headers={
+                "Accept": "application/json, text/plain, */*",
+                "User-Agent": "ictlogin/0.1 (+https://gw.ict.ac.cn)",
+                "Cache-Control": "no-cache",
+            },
+            method="GET",
+        )
+        try:
+            with self._opener.open(request, timeout=self.timeout) as response:
+                body = response.read(4 * 1024 * 1024).decode("utf-8", "replace")
+        except urllib.error.HTTPError as exc:
+            body = exc.read(4 * 1024 * 1024).decode("utf-8", "replace")
+            payload = self._decode(body, callback)
+            if isinstance(payload, Mapping):
+                raise GatewayError(
+                    error_message(payload),
+                    code=_error_code(payload),
+                    payload=payload,
+                    status=exc.code,
+                ) from exc
+            raise GatewayError(
+                f"网关 HTTP 错误（{exc.code}）",
+                status=exc.code,
+            ) from exc
+        except (urllib.error.URLError, TimeoutError, socket.timeout) as exc:
+            raise GatewayError(
+                f"无法连接网关：{exc.reason if hasattr(exc, 'reason') else exc}"
+            ) from exc
+
+        payload = self._decode(body, callback)
+        if isinstance(payload, Mapping):
+            return payload
+        if payload == "ok":
+            return {"error": "ok", "res": "ok"}
+        raise GatewayError("网关返回了无法解析的响应")
+
+    @staticmethod
+    def _decode(body: str, callback: str = "") -> Any:
+        text = body.strip()
+        if callback:
+            prefix = callback + "("
+            if text.startswith(prefix):
+                text = text[len(prefix) :]
+                if text.endswith(");"):
+                    text = text[:-2]
+                elif text.endswith(")"):
+                    text = text[:-1]
+        if not text:
+            return {}
+        try:
+            return json.loads(text)
+        except json.JSONDecodeError:
+            # Be tolerant of a JSONP callback name generated by a proxy or an
+            # older SRun build that adds whitespace around the expression.
+            match = re.match(r"^[A-Za-z_$][\w$]*\s*\((.*)\)\s*;?\s*$", text, re.S)
+            if match:
+                try:
+                    return json.loads(match.group(1))
+                except json.JSONDecodeError:
+                    pass
+            return text
+
+    def _ensure_ok(self, payload: Mapping[str, Any]) -> Mapping[str, Any]:
+        if not is_success(payload):
+            raise GatewayError(
+                error_message(payload),
+                code=_error_code(payload),
+                payload=payload,
+            )
+        return payload
+
+    def status(self, ip: Optional[str] = None) -> Status:
+        params = {"ip": ip} if ip else {}
+        payload = self._request("/cgi-bin/rad_user_info", params, jsonp=True)
+        if not isinstance(payload, Mapping):
+            raise GatewayError("状态接口返回了无法解析的响应")
+        code = _error_code(payload)
+        if code == "not_online_error":
+            return Status(False, payload)
+        if is_success(payload):
+            return Status(True, payload)
+        raise GatewayError(error_message(payload), code=code, payload=payload)
+
+    def heartbeat(self, ip: str) -> Heartbeat:
+        """Probe the authenticated IP and expose the gateway keepalive marker.
+
+        SRun exposes its online/keepalive state through ``rad_user_info``;
+        keeping this request periodic is the least invasive way to keep a
+        headless session active without sending arbitrary external traffic.
+        """
+
+        started = time.monotonic()
+        status = self.status(ip)
+        elapsed = time.monotonic() - started
+        raw_keepalive = status.payload.get("keepalive_time")
+        try:
+            keepalive_time = int(raw_keepalive) if raw_keepalive is not None else None
+        except (TypeError, ValueError):
+            keepalive_time = None
+        return Heartbeat(status, elapsed, keepalive_time)
+
+    def discover_ip(self) -> str:
+        """Ask the gateway which client IP it sees, without local heuristics."""
+
+        status = self.status()
+        payload = status.payload
+        for key in ("client_ip", "online_ip", "ip"):
+            value = str(payload.get(key) or "").strip()
+            if value:
+                return value
+        # A status response should normally always contain client_ip.  Keep a
+        # socket fallback for older deployments that omit it.
+        parsed = urllib.parse.urlsplit(self.base_url)
+        host = parsed.hostname or "gw.ict.ac.cn"
+        family = socket.AF_INET6 if ":" in host else socket.AF_INET
+        sock = socket.socket(family, socket.SOCK_DGRAM)
+        try:
+            sock.connect(
+                (host, parsed.port or (443 if parsed.scheme == "https" else 80))
+            )
+            value = sock.getsockname()[0]
+        except OSError as exc:
+            raise GatewayError(
+                f"无法确定服务器在网关侧的 IP，请使用 --ip 指定（{exc}）"
+            ) from exc
+        finally:
+            sock.close()
+        if not value:
+            raise GatewayError("无法确定服务器在网关侧的 IP，请使用 --ip 指定")
+        return value
+
+    def check_captcha(self, username: str, ip: str) -> bool:
+        try:
+            payload = self._request(
+                "/v2/srun_portal_captcha_image_info",
+                {"user_name": username, "ip": ip},
+            )
+        except GatewayError as exc:
+            if exc.status == 404:
+                return False
+            raise
+        if not isinstance(payload, Mapping):
+            return False
+        return str(payload.get("data", "0")) == "1"
+
+    def fetch_captcha(self) -> tuple[str, bytes]:
+        payload = self._request(
+            "/v2/srun_portal_captcha_image",
+            {"t": str(int(time.time()))},
+        )
+        if not isinstance(payload, Mapping) or payload.get("code") != 0:
+            raise GatewayError(
+                error_message(payload if isinstance(payload, Mapping) else {}),
+                payload=payload if isinstance(payload, Mapping) else {},
+            )
+        data = payload.get("data")
+        if not isinstance(data, Mapping):
+            raise GatewayError("网关没有返回验证码图片")
+        captcha_id = str(data.get("captcha_id") or "")
+        image = str(data.get("captcha_img") or "")
+        if not captcha_id or "," not in image:
+            raise GatewayError("网关返回的验证码图片无效")
+        try:
+            image_bytes = base64.b64decode(image.split(",", 1)[1], validate=True)
+        except (ValueError, base64.binascii.Error) as exc:
+            raise GatewayError("验证码图片解码失败") from exc
+        return captcha_id, image_bytes
+
+    @staticmethod
+    def save_captcha(image: bytes) -> Path:
+        cache_root = (
+            Path(os.environ.get("XDG_CACHE_HOME", str(Path.home() / ".cache")))
+            / "ictlogin"
+        )
+        cache_root.mkdir(mode=0o700, parents=True, exist_ok=True)
+        os.chmod(cache_root, 0o700)
+        path = (
+            cache_root / f"captcha-{int(time.time())}-{random.getrandbits(24):06x}.png"
+        )
+        with path.open("wb") as handle:
+            handle.write(image)
+        os.chmod(path, 0o600)
+        return path
+
+    def challenge(self, username: str, ip: str) -> str:
+        payload = self._request(
+            "/cgi-bin/get_challenge",
+            {"username": username, "ip": ip},
+            jsonp=True,
+        )
+        if not isinstance(payload, Mapping):
+            raise GatewayError("挑战接口返回了无法解析的响应")
+        self._ensure_ok(payload)
+        token = str(payload.get("challenge") or "")
+        if not token:
+            raise GatewayError("网关没有返回挑战码", payload=payload)
+        return token
+
+    def authenticate(
+        self,
+        *,
+        username: str,
+        password: str,
+        ip: str,
+        captcha_id: str = "",
+        captcha_value: str = "",
+        domain: str = "",
+        nas_ip: str = "",
+        ap_id: str = "",
+        ap_ip: str = "",
+        mac: str = "",
+    ) -> Mapping[str, Any]:
+        full_username = username + domain
+        token = self.challenge(full_username, ip)
+        params = login_parameters(
+            username=full_username,
+            password=password,
+            ip=ip,
+            ac_id=self.ac_id,
+            token=token,
+            captcha_id=captcha_id,
+            captcha_value=captcha_value,
+            nas_ip=nas_ip,
+            ap_id=ap_id,
+            ap_ip=ap_ip,
+            mac=mac,
+        )
+        payload = self._request("/cgi-bin/srun_portal", params, jsonp=True)
+        if not isinstance(payload, Mapping):
+            raise GatewayError("登录接口返回了无法解析的响应")
+        self._ensure_ok(payload)
+        return payload
+
+    def wait_for_status(self, ip: str, online: bool, timeout: float = 5.0) -> Status:
+        deadline = time.monotonic() + max(0.0, timeout)
+        latest: Optional[Status] = None
+        while True:
+            latest = self.status(ip)
+            if latest.online == online:
+                return latest
+            if time.monotonic() >= deadline:
+                return latest
+            time.sleep(0.5)
+
+    def logout(self, *, username: str, ip: str) -> Mapping[str, Any]:
+        payload = self._request(
+            "/cgi-bin/srun_portal",
+            {
+                "action": "logout",
+                "username": username,
+                "ip": ip,
+                "ac_id": self.ac_id,
+            },
+            jsonp=True,
+        )
+        if not isinstance(payload, Mapping):
+            raise GatewayError("注销接口返回了无法解析的响应")
+        self._ensure_ok(payload)
+        return payload
+
+
+__all__ = [
+    "GatewayClient",
+    "GatewayError",
+    "Heartbeat",
+    "Status",
+    "error_message",
+]

src/ictlogin/protocol.py

@@ -0,0 +1,233 @@
+"""SRun Portal's browser-side login encoding.
+
+The gateway uses the same protocol as the JavaScript Portal client:
+HMAC-MD5 for the password, a little-endian XXTEA variant for ``info``,
+the gateway's custom Base64 alphabet, and a SHA-1 request checksum.
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+from typing import Any, Mapping
+
+
+CUSTOM_ALPHABET = "LVoJPiCN2R8G90yg+hmFHuacZ1OWMnrsSTXkYpUq/3dlbfKwv6xztjI7DeBE45QA"
+STANDARD_ALPHABET = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/"
+_U32 = 0xFFFFFFFF
+_DELTA = 0x9E3779B9
+
+
+def hmac_md5(password: str, token: str) -> str:
+    """Return the hex HMAC-MD5 used by ``md5(password, token)`` in Portal.js."""
+
+    return hmac.new(
+        token.encode("utf-8"), password.encode("utf-8"), hashlib.md5
+    ).hexdigest()
+
+
+def _js_code_units(value: str) -> list[int]:
+    """Return JavaScript UTF-16 code units for a Python string."""
+
+    raw = value.encode("utf-16-le", "surrogatepass")
+    return [raw[i] | (raw[i + 1] << 8) for i in range(0, len(raw), 2)]
+
+
+def _to_words(value: str, include_length: bool) -> list[int]:
+    """Pack four JavaScript characters into each little-endian 32-bit word."""
+
+    chars = _js_code_units(value)
+    words: list[int] = []
+    for offset in range(0, len(chars), 4):
+        word = 0
+        for shift, code in enumerate(chars[offset : offset + 4]):
+            # This intentionally mirrors JS bitwise coercion.  SRun's
+            # implementation packs character values, rather than UTF-8 bytes.
+            word |= code << (shift * 8)
+        words.append(word & _U32)
+    if include_length:
+        words.append(len(chars))
+    return words
+
+
+def _from_words(words: list[int]) -> bytes:
+    output = bytearray()
+    for word in words:
+        word &= _U32
+        output.extend(
+            (
+                word & 0xFF,
+                (word >> 8) & 0xFF,
+                (word >> 16) & 0xFF,
+                (word >> 24) & 0xFF,
+            )
+        )
+    return bytes(output)
+
+
+def _xxtea_encode(value: str, key: str) -> bytes:
+    """Encode ``value`` with the XXTEA routine embedded in Portal.js."""
+
+    if value == "":
+        return b""
+
+    words = _to_words(value, include_length=True)
+    key_words = _to_words(key, include_length=False)
+    if len(key_words) < 4:
+        key_words.extend([0] * (4 - len(key_words)))
+
+    n = len(words) - 1
+    z = words[n]
+    y = words[0]
+    rounds = 6 + 52 // (n + 1)
+    total = 0
+
+    while rounds > 0:
+        rounds -= 1
+        total = (total + _DELTA) & _U32
+        e = (total >> 2) & 3
+
+        for p in range(n):
+            y = words[p + 1]
+            mixed = (z >> 5) ^ ((y << 2) & _U32)
+            mixed = (mixed + ((y >> 3) ^ ((z << 4) & _U32) ^ (total ^ y))) & _U32
+            mixed = (mixed + (key_words[(p & 3) ^ e] ^ z)) & _U32
+            z = words[p] = (words[p] + mixed) & _U32
+
+        y = words[0]
+        mixed = (z >> 5) ^ ((y << 2) & _U32)
+        mixed = (mixed + ((y >> 3) ^ ((z << 4) & _U32) ^ (total ^ y))) & _U32
+        # JavaScript's ``for (p = 0; p < n; p++)`` leaves p == n here.
+        mixed = (mixed + (key_words[(n & 3) ^ e] ^ z)) & _U32
+        z = words[n] = (words[n] + mixed) & _U32
+
+    return _from_words(words)
+
+
+def custom_base64(data: bytes) -> str:
+    """Encode bytes using SRun's 64-character alphabet."""
+
+    standard = base64.b64encode(data).decode("ascii")
+    translation = str.maketrans(STANDARD_ALPHABET, CUSTOM_ALPHABET)
+    return standard.translate(translation)
+
+
+def encode_info(
+    username: str,
+    password: str,
+    ip: str,
+    ac_id: str,
+    token: str,
+) -> str:
+    """Build the ``{SRBX1}`` value sent to the authentication endpoint."""
+
+    info = json.dumps(
+        {
+            "username": username,
+            "password": password,
+            "ip": ip,
+            "acid": str(ac_id),
+            "enc_ver": "srun_bx1",
+        },
+        ensure_ascii=False,
+        separators=(",", ":"),
+    )
+    return "{SRBX1}" + custom_base64(_xxtea_encode(info, token))
+
+
+def checksum(
+    token: str,
+    username: str,
+    hmd5: str,
+    ac_id: str,
+    ip: str,
+    info: str,
+    n: str = "200",
+    auth_type: str = "1",
+) -> str:
+    """Return the SHA-1 checksum for a login request."""
+
+    material = "".join(
+        (
+            token,
+            username,
+            token,
+            hmd5,
+            token,
+            str(ac_id),
+            token,
+            ip,
+            token,
+            str(n),
+            token,
+            str(auth_type),
+            token,
+            info,
+        )
+    )
+    return hashlib.sha1(material.encode("utf-8")).hexdigest()
+
+
+def login_parameters(
+    *,
+    username: str,
+    password: str,
+    ip: str,
+    ac_id: str,
+    token: str,
+    captcha_id: str = "",
+    captcha_value: str = "",
+    os_name: str = "Linux",
+    platform_name: str = "Linux",
+    nas_ip: str = "",
+    ap_id: str = "",
+    ap_ip: str = "",
+    mac: str = "",
+) -> dict[str, str]:
+    """Create all query parameters for ``action=login``."""
+
+    n = "200"
+    auth_type = "1"
+    hmd5 = hmac_md5(password, token)
+    info = encode_info(username, password, ip, ac_id, token)
+    return {
+        "action": "login",
+        "username": username,
+        "password": "{MD5}" + hmd5,
+        "os": os_name,
+        "name": platform_name,
+        "nas_ip": nas_ip,
+        "double_stack": "0",
+        "chksum": checksum(token, username, hmd5, ac_id, ip, info, n, auth_type),
+        "info": info,
+        "ac_id": str(ac_id),
+        "ip": ip,
+        "n": n,
+        "type": auth_type,
+        "captchaId": captcha_id,
+        "captchaVal": captcha_value,
+        "ap_id": ap_id,
+        "ap_ip": ap_ip,
+        "mac": mac,
+    }
+
+
+def is_success(payload: Mapping[str, Any]) -> bool:
+    """Recognize successful responses from different SRun server versions."""
+
+    if payload.get("error") == "ok" or payload.get("res") == "ok":
+        return True
+    return payload.get("code") == 0 and not payload.get("error")
+
+
+__all__ = [
+    "CUSTOM_ALPHABET",
+    "checksum",
+    "custom_base64",
+    "encode_info",
+    "hmac_md5",
+    "is_success",
+    "login_parameters",
+]

tests/test_heartbeat.py

@@ -0,0 +1,27 @@
+from ictlogin.cli import _build_parser
+from ictlogin.client import GatewayClient, Status
+
+
+class _FakeClient(GatewayClient):
+    def __init__(self):
+        pass
+
+    def status(self, ip=None):
+        return Status(True, {"keepalive_time": "12345"})
+
+
+def test_heartbeat_extracts_keepalive_marker():
+    result = _FakeClient().heartbeat("10.0.0.1")
+    assert result.status.online
+    assert result.keepalive_time == 12345
+    assert result.latency_seconds >= 0
+
+
+def test_heartbeat_parser_supports_once_and_auto_login():
+    args = _build_parser().parse_args(
+        ["heartbeat", "--once", "--interval", "10", "--auto-login"]
+    )
+    assert args.command == "heartbeat"
+    assert args.once is True
+    assert args.interval == 10
+    assert args.auto_login is True

tests/test_protocol.py

@@ -0,0 +1,56 @@
+import hashlib
+import hmac
+
+from ictlogin.protocol import (
+    checksum,
+    custom_base64,
+    encode_info,
+    hmac_md5,
+    login_parameters,
+)
+
+
+def test_hmac_md5_matches_portal_md5_helper():
+    assert (
+        hmac_md5("password", "token")
+        == hmac.new(b"token", b"password", hashlib.md5).hexdigest()
+    )
+
+
+def test_custom_base64_alphabet():
+    assert custom_base64(b"abc") == "ZaRk"
+
+
+def test_login_parameters_are_self_consistent():
+    params = login_parameters(
+        username="alice",
+        password="secret",
+        ip="10.208.16.60",
+        ac_id="1",
+        token="0123456789abcdef0123456789abcdef",
+    )
+    expected = checksum(
+        params["password"][5:] and "0123456789abcdef0123456789abcdef",
+        "alice",
+        params["password"][5:],
+        "1",
+        "10.208.16.60",
+        params["info"],
+    )
+    assert params["chksum"] == expected
+    assert params["info"].startswith("{SRBX1}")
+    assert params["password"].startswith("{MD5}")
+
+
+def test_known_info_is_deterministic():
+    token = "0123456789abcdef0123456789abcdef"
+    first = encode_info("alice", "secret", "10.208.16.60", "1", token)
+    second = encode_info("alice", "secret", "10.208.16.60", "1", token)
+    assert first == second
+
+
+def test_browser_compatible_info_vector():
+    token = "0123456789abcdef0123456789abcdef"
+    assert encode_info("alice", "secret", "10.208.16.60", "1", token) == (
+        "{SRBX1}0o4inQiaHwW7y5JpHN8VD90N6xFNpZTtn0AXMsitCTUCwNKGNypce0RXMvQhFZ4Eir85FjJW6Rhk/lLkIoqATc8mywXQOVTQFhF3BZCTwMP+Jp4PD2Rob52dxvouN2Gs"
+    )