commit e053d31a2fe3013580964182937fe968b9245d45 Author: roy1994 Date: Sun Aug 23 05:11:20 2026 +0800 feat: add ICT gateway login and heartbeat CLI diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..ec27f1c --- /dev/null +++ b/.gitignore @@ -0,0 +1,6 @@ +__pycache__/ +.pytest_cache/ +.ruff_cache/ +*.egg-info/ +build/ +dist/ diff --git a/README.md b/README.md new file mode 100644 index 0000000..d04da36 --- /dev/null +++ b/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`。 +- 工具每次登录只发送一次认证请求，不会重试密码，避免触发账号锁定。 diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..e69dd2a --- /dev/null +++ b/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"] diff --git a/run_ictlogin.py b/run_ictlogin.py new file mode 100644 index 0000000..e9bbd74 --- /dev/null +++ b/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() diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..fca07b2 --- /dev/null +++ b/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"]}, +) diff --git a/src/ictlogin/__init__.py b/src/ictlogin/__init__.py new file mode 100644 index 0000000..3e0e7e3 --- /dev/null +++ b/src/ictlogin/__init__.py @@ -0,0 +1,4 @@ +"""ICT SRun gateway command-line client.""" + +__all__ = ["__version__"] +__version__ = "0.2.0" diff --git a/src/ictlogin/__main__.py b/src/ictlogin/__main__.py new file mode 100644 index 0000000..2f05ddc --- /dev/null +++ b/src/ictlogin/__main__.py @@ -0,0 +1,5 @@ +from .cli import main + + +if __name__ == "__main__": + main() diff --git a/src/ictlogin/cli.py b/src/ictlogin/cli.py new file mode 100644 index 0000000..aea57c8 --- /dev/null +++ b/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"] diff --git a/src/ictlogin/client.py b/src/ictlogin/client.py new file mode 100644 index 0000000..011244a --- /dev/null +++ b/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", +] diff --git a/src/ictlogin/protocol.py b/src/ictlogin/protocol.py new file mode 100644 index 0000000..e27d410 --- /dev/null +++ b/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", +] diff --git a/tests/test_heartbeat.py b/tests/test_heartbeat.py new file mode 100644 index 0000000..4deb56a --- /dev/null +++ b/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 diff --git a/tests/test_protocol.py b/tests/test_protocol.py new file mode 100644 index 0000000..5b8358b --- /dev/null +++ b/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" + )