diff --git a/workbench/network/campus-net/README.md b/workbench/network/campus-net/README.md index bc768e6..eb57985 100644 --- a/workbench/network/campus-net/README.md +++ b/workbench/network/campus-net/README.md @@ -81,7 +81,7 @@ if (-not (Test-Path -LiteralPath config.json)) { - `username`、`password`:校园网账号和密码。程序直接读取本地配置,不会再次询问密码或读取密码环境变量。 - `carrier`:运营商显示名。程序会从本次会话的服务列表中找到“中国电信”对应的动态值再提交。 -程序只使用 `interface_index` 指定的 IPv4 接口,不按接口名称或类型猜测校园网,也不会在失败时改用其他网络接口。校园网可以位于 WLAN、以太网或其他 IPv4 接口;若所选接口未连接、无可用路由或返回其他 captive 门户,CLI 与 GUI 会停止并提示重新选择实际承载校园网的接口,不会提交账号密码。所选接口已经可以访问互联网时,程序会跳过登录,并提醒校园网位于其他接口时需要改选接口。 +程序始终先使用 `interface_index` 指定的 IPv4 接口,不按接口名称或类型猜测校园网,也不会自动改用其他网络接口。校园网可以位于 WLAN、以太网或其他 IPv4 接口;若交互式 CLI 无法确认所选接口的网络状态,会列出当前 Windows IPv4 接口,由用户显式选择一个接口后重新探测。该选择仅用于本次运行,不修改 `config.json`;重定向输入输出、后台任务和 GUI 不会等待终端输入,仍会安全停止。GUI 可在界面中重新选择接口再连接。任何改选都发生在提交账号密码之前。所选接口已经可以访问互联网时,程序会跳过登录,并提醒校园网位于其他接口时需要改选接口。 以下协议参数不再属于用户配置,由代码统一维护:User-Agent、连通性探测指纹、门户入口路径、验证码模式和次数、登录后确认间隔与超时。这样升级门户适配时只改实现,不要求用户同步一组内部常量。 @@ -136,6 +136,12 @@ uv sync --group dev & .\.venv\Scripts\python.exe -m PyInstaller --clean --noconfirm gui.spec ``` +控制台版构建后可运行不读取配置、不访问网络的入口冒烟检查: + +```powershell +& .\dist\Auto-Connect-CampusNet.exe --help +``` + 子项目在 `pyproject.toml` 中声明自己的 uv workspace 边界,因此 `uv sync` 不会继续向上扫描仓库根仅用于 Ruff 的 `pyproject.toml`。同步后直接使用该环境的 Python,也不会让后续每条测试或构建命令重新触发项目发现,从而避免与项目构建无关的“缺少 `[project]`”警告。 `build.spec` 继续生成控制台版 `dist\Auto-Connect-CampusNet.exe`,`gui.spec` 生成无控制台窗口版 `dist\Auto-Connect-CampusNet-GUI.exe`。两个 spec 都不会把 `config.json` 打进 EXE,发布物也不应携带用户配置或备份。 diff --git a/workbench/network/campus-net/campus_net/application.py b/workbench/network/campus-net/campus_net/application.py index 89dd43a..f8460e4 100644 --- a/workbench/network/campus-net/campus_net/application.py +++ b/workbench/network/campus-net/campus_net/application.py @@ -21,7 +21,7 @@ ) from .interactive import CaptchaPromptError, prompt_captcha from .legacy import run_legacy -from .runner import run_captive_http +from .runner import InterfaceSelector, run_captive_http from .sso import SsoProtocolError StatusReporter = Callable[[str], None] @@ -40,6 +40,7 @@ async def execute_config( probe_only: bool = False, captcha_provider: CaptchaProvider = prompt_captcha, status_callback: StatusReporter = print, + interface_selector: InterfaceSelector | None = None, ) -> int: adapter = get_adapter(cfg) if adapter == ADAPTER_CAPTIVE_SSO_HTTP: @@ -49,6 +50,7 @@ async def execute_config( probe_only=probe_only, captcha_provider=captcha_provider, status_callback=status_callback, + interface_selector=interface_selector, ) if adapter == ADAPTER_LEGACY_EPORTAL: if probe_only: diff --git a/workbench/network/campus-net/campus_net/gui.py b/workbench/network/campus-net/campus_net/gui.py index b10c641..49fa9bc 100644 --- a/workbench/network/campus-net/campus_net/gui.py +++ b/workbench/network/campus-net/campus_net/gui.py @@ -33,12 +33,11 @@ FailedEvent, FinishedEvent, LogEvent, - NetworkInterface, OperationController, build_config_from_form, form_values_from_config, - list_windows_ipv4_interfaces, ) +from .interfaces import NetworkInterface, list_windows_ipv4_interfaces from .window_state import ( ScreenRect, WindowState, diff --git a/workbench/network/campus-net/campus_net/gui_core.py b/workbench/network/campus-net/campus_net/gui_core.py index 310e77f..069ac80 100644 --- a/workbench/network/campus-net/campus_net/gui_core.py +++ b/workbench/network/campus-net/campus_net/gui_core.py @@ -2,14 +2,9 @@ import asyncio import copy -import json -import os -import subprocess -import sys import threading from collections.abc import Awaitable, Callable, Mapping from dataclasses import dataclass, field -from pathlib import Path from queue import Empty, Queue from typing import Any @@ -59,23 +54,6 @@ class FailedEvent: GuiEvent = LogEvent | CaptchaRequested | TerminalEvent -@dataclass(frozen=True, slots=True) -class NetworkInterface: - index: int - alias: str - state: str - metric: int - - @property - def display_name(self) -> str: - state = { - "connected": "已连接", - "disconnected": "未连接", - "authenticating": "正在认证", - }.get(self.state.casefold(), self.state or "未知状态") - return f"{self.index} · {self.alias} · {state}" - - @dataclass(slots=True) class _PendingCaptcha: loop: asyncio.AbstractEventLoop @@ -352,98 +330,6 @@ def form_values_from_config(config: EditableConfig) -> tuple[int, dict[str, str] raise TypeError("GUI 只接受正式 version=1 或 version=2 配置") -def list_windows_ipv4_interfaces() -> list[NetworkInterface]: - if sys.platform != "win32": - raise OSError("网卡列表当前仅支持 Windows") - system_root = os.environ.get("SystemRoot") - if not system_root: - raise OSError("Windows 环境缺少 SystemRoot") - powershell_path = ( - Path(system_root) / "System32" / "WindowsPowerShell" / "v1.0" / "powershell.exe" - ) - if not powershell_path.is_file(): - raise OSError(f"找不到系统 PowerShell:{powershell_path}") - command = ( - "[Console]::OutputEncoding = [Text.UTF8Encoding]::new(); " - "Get-NetIPInterface -AddressFamily IPv4 | " - "Select-Object InterfaceIndex,InterfaceAlias," - "@{Name='ConnectionState';Expression={$_.ConnectionState.ToString()}}," - "InterfaceMetric | " - "ConvertTo-Json -Compress" - ) - creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) - result = subprocess.run( - [str(powershell_path), "-NoProfile", "-NonInteractive", "-Command", command], - check=True, - capture_output=True, - text=True, - encoding="utf-8", - timeout=10, - creationflags=creation_flags, - ) - return parse_windows_ipv4_interfaces(result.stdout) - - -def parse_windows_ipv4_interfaces(payload: str) -> list[NetworkInterface]: - try: - decoded = json.loads(payload) - except json.JSONDecodeError as error: - raise ValueError("无法解析 Windows IPv4 接口列表") from error - if isinstance(decoded, dict): - items = [decoded] - elif isinstance(decoded, list): - items = decoded - else: - raise ValueError("Windows IPv4 接口列表格式无效") - - interfaces: list[NetworkInterface] = [] - for item in items: - if not isinstance(item, dict): - continue - index = item.get("InterfaceIndex") - alias = item.get("InterfaceAlias") - state = _normalize_connection_state(item.get("ConnectionState")) - metric = item.get("InterfaceMetric") - if ( - isinstance(index, bool) - or not isinstance(index, int) - or not isinstance(alias, str) - or not alias.strip() - ): - continue - interfaces.append( - NetworkInterface( - index=index, - alias=alias.strip(), - state=state, - metric=metric if isinstance(metric, int) and not isinstance(metric, bool) else 0, - ) - ) - if not interfaces: - raise ValueError("没有找到可用的 Windows IPv4 接口") - return sorted( - interfaces, - key=lambda item: ( - item.state.casefold() != "connected", - item.metric, - item.index, - ), - ) - - -def _normalize_connection_state(value: object) -> str: - if isinstance(value, str): - return value.strip() - if isinstance(value, int) and not isinstance(value, bool): - return { - 0: "Disconnected", - 1: "Connected", - 2: "Disconnected", - 3: "Authenticating", - }.get(value, f"状态 {value}") - return "" - - def _set_future_result(future: asyncio.Future[str], value: str) -> None: if not future.done(): future.set_result(value) diff --git a/workbench/network/campus-net/campus_net/interfaces.py b/workbench/network/campus-net/campus_net/interfaces.py new file mode 100644 index 0000000..ff14404 --- /dev/null +++ b/workbench/network/campus-net/campus_net/interfaces.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from dataclasses import dataclass +from pathlib import Path + + +@dataclass(frozen=True, slots=True) +class NetworkInterface: + index: int + alias: str + state: str + metric: int + + @property + def display_name(self) -> str: + state = { + "connected": "已连接", + "disconnected": "未连接", + "authenticating": "正在认证", + }.get(self.state.casefold(), self.state or "未知状态") + return f"{self.index} · {self.alias} · {state}" + + +def list_windows_ipv4_interfaces() -> list[NetworkInterface]: + if sys.platform != "win32": + raise OSError("网卡列表当前仅支持 Windows") + system_root = os.environ.get("SystemRoot") + if not system_root: + raise OSError("Windows 环境缺少 SystemRoot") + powershell_path = ( + Path(system_root) / "System32" / "WindowsPowerShell" / "v1.0" / "powershell.exe" + ) + if not powershell_path.is_file(): + raise OSError(f"找不到系统 PowerShell:{powershell_path}") + command = ( + "[Console]::OutputEncoding = [Text.UTF8Encoding]::new(); " + "Get-NetIPInterface -AddressFamily IPv4 | " + "Select-Object InterfaceIndex,InterfaceAlias," + "@{Name='ConnectionState';Expression={$_.ConnectionState.ToString()}}," + "InterfaceMetric | " + "ConvertTo-Json -Compress" + ) + creation_flags = getattr(subprocess, "CREATE_NO_WINDOW", 0) + result = subprocess.run( + [str(powershell_path), "-NoProfile", "-NonInteractive", "-Command", command], + check=True, + capture_output=True, + text=True, + encoding="utf-8", + timeout=10, + creationflags=creation_flags, + ) + return parse_windows_ipv4_interfaces(result.stdout) + + +def parse_windows_ipv4_interfaces(payload: str) -> list[NetworkInterface]: + try: + decoded = json.loads(payload) + except json.JSONDecodeError as error: + raise ValueError("无法解析 Windows IPv4 接口列表") from error + if isinstance(decoded, dict): + items = [decoded] + elif isinstance(decoded, list): + items = decoded + else: + raise ValueError("Windows IPv4 接口列表格式无效") + + interfaces: list[NetworkInterface] = [] + for item in items: + if not isinstance(item, dict): + continue + index = item.get("InterfaceIndex") + alias = item.get("InterfaceAlias") + state = _normalize_connection_state(item.get("ConnectionState")) + metric = item.get("InterfaceMetric") + if ( + isinstance(index, bool) + or not isinstance(index, int) + or not isinstance(alias, str) + or not alias.strip() + ): + continue + interfaces.append( + NetworkInterface( + index=index, + alias=alias.strip(), + state=state, + metric=metric if isinstance(metric, int) and not isinstance(metric, bool) else 0, + ) + ) + if not interfaces: + raise ValueError("没有找到可用的 Windows IPv4 接口") + return sorted( + interfaces, + key=lambda item: ( + item.state.casefold() != "connected", + item.metric, + item.index, + ), + ) + + +def _normalize_connection_state(value: object) -> str: + if isinstance(value, str): + return value.strip() + if isinstance(value, int) and not isinstance(value, bool): + return { + 0: "Disconnected", + 1: "Connected", + 2: "Disconnected", + 3: "Authenticating", + }.get(value, f"状态 {value}") + return "" diff --git a/workbench/network/campus-net/campus_net/runner.py b/workbench/network/campus-net/campus_net/runner.py index c4372eb..92c7a59 100644 --- a/workbench/network/campus-net/campus_net/runner.py +++ b/workbench/network/campus-net/campus_net/runner.py @@ -1,6 +1,7 @@ from __future__ import annotations from collections.abc import Callable +from dataclasses import replace import aiohttp @@ -14,6 +15,8 @@ from .config import CaptiveHttpConfig from .interactive import prompt_captcha +InterfaceSelector = Callable[[int], int | None] + async def run_captive_http( config: CaptiveHttpConfig, @@ -21,7 +24,58 @@ async def run_captive_http( probe_only: bool = False, captcha_provider: CaptchaProvider = prompt_captcha, status_callback: Callable[[str], None] = print, + interface_selector: InterfaceSelector | None = None, ) -> int: + active_config = config + while True: + exit_code, retry_reason = await _run_captive_http_attempt( + active_config, + probe_only=probe_only, + captcha_provider=captcha_provider, + status_callback=status_callback, + ) + if retry_reason is None: + return exit_code + + reason = retry_reason.rstrip("。") or "探测结果无法识别" + if interface_selector is None: + status_callback( + f"无法确认 IPv4 接口 {active_config.interface_index} 的网络状态:{reason}。" + "请确认该接口已接入校园网,或重新选择实际承载校园网的 IPv4 接口;" + "已停止,未改用其他网络接口。" + ) + return exit_code + + status_callback( + f"无法确认 IPv4 接口 {active_config.interface_index} 的网络状态:{reason}。" + "可从当前 Windows IPv4 接口列表中显式选择一个接口重新探测;" + "程序不会自动选择,也不会修改 config.json。" + ) + selected_index = interface_selector(active_config.interface_index) + if selected_index is None: + status_callback("已取消接口选择;已停止,未改用其他网络接口。") + return exit_code + if ( + isinstance(selected_index, bool) + or not isinstance(selected_index, int) + or not 1 <= selected_index <= 0xFFFFFFFF + ): + raise ValueError("选择的 interface_index 不是有效的 Windows 接口索引") + + active_config = replace(active_config, interface_index=selected_index) + status_callback( + f"已显式选择 IPv4 接口 {selected_index},仅本次运行使用;" + "config.json 未修改,正在重新探测。" + ) + + +async def _run_captive_http_attempt( + config: CaptiveHttpConfig, + *, + probe_only: bool, + captcha_provider: CaptchaProvider, + status_callback: Callable[[str], None], +) -> tuple[int, str | None]: connector = create_bound_connector(config.interface_index) cookie_jar = aiohttp.CookieJar(unsafe=True) timeout = aiohttp.ClientTimeout(total=15, connect=8) @@ -37,25 +91,19 @@ async def run_captive_http( timeout=timeout, ) as session: client = AggregatePortalClient(session, config) - status_callback("正在通过配置的 IPv4 接口探测校园网状态…") + status_callback(f"正在通过 IPv4 接口 {config.interface_index} 探测校园网状态…") initialization = await client.initialize() if initialization.probe.state is NetworkState.ONLINE: status_callback( - f"配置的 IPv4 接口 {config.interface_index} 已通过独立连通性探测,无需登录;" + f"IPv4 接口 {config.interface_index} 已通过独立连通性探测,无需登录;" "如果校园网实际位于另一接口,请重新选择对应的 IPv4 接口。" ) - return 0 + return 0, None if initialization.probe.state is NetworkState.UNKNOWN: - reason = initialization.probe.reason.rstrip("。") or "探测结果无法识别" - status_callback( - f"无法确认配置的 IPv4 接口 {config.interface_index} 的网络状态:{reason}。" - "请确认该接口已接入校园网,或重新选择实际承载校园网的 IPv4 接口;" - "已停止,未改用其他网络接口。" - ) - return 2 + return 2, initialization.probe.reason if probe_only: status_callback("已识别当前接口的 captive 门户入口;未提交任何认证信息。") - return 0 + return 0, None if initialization.context is None: raise PortalProtocolError("captive 初始化没有生成门户会话") @@ -106,7 +154,7 @@ async def run_captive_http( status_callback("门户在线检查响应不完整;没有重放请求,改用只读状态核对。") if await client.verify_online(context.session_id): status_callback("校园网登录完成,门户状态与独立连通性探测均已确认在线。") - return 0 + return 0, None status_callback("门户未能与独立连通性探测同时确认在线。") - return 4 + return 4, None diff --git a/workbench/network/campus-net/docs/new-portal-adaptation.md b/workbench/network/campus-net/docs/new-portal-adaptation.md index 7d08ea9..dd4e997 100644 --- a/workbench/network/campus-net/docs/new-portal-adaptation.md +++ b/workbench/network/campus-net/docs/new-portal-adaptation.md @@ -53,7 +53,7 @@ IPPROTO_IP / IP_UNICAST_IF / network-byte-order(interface_index) 这样 Wi-Fi、有线网卡或其他 IPv4 接口都使用同一套逻辑。它减少普通默认路由造成的选择歧义;程序不会修改系统路由、sing-box 或 v2rayN 配置。`IP_UNICAST_IF` 只能要求 socket 选定接口,不能证明最终物理出口。内核级 TUN/WFP 仍可能透明截获流量,甚至返回可被分类为 `ONLINE` 或 `CAPTIVE` 的合法响应;需要证明物理出口时仍应使用抓包或系统路由诊断。 -运行时不会把任何接口名称或类型固定为校园网。校园网可以位于 WLAN、以太网或其他 IPv4 接口,配置必须指向实际承载校园网的接口索引。配置接口未连接、没有可用路由或无法完成探测时,流程保持 `UNKNOWN` 并以退出码 `2` 停止,提示中不暴露 `aiohttp`、主机、SSL 或本地化 socket 异常细节,也不会创建无绑定连接器或回退到其他网络接口。识别门户后的 HTTP 超时或连接错误同样由 CLI 与 GUI 共用的异常分类器转换为通用接口提示,不直接显示底层传输异常。配置接口已通过独立在线指纹时只说明该接口已经联网;如果校园网位于另一接口,用户仍需改选对应的 IPv4 接口。 +运行时不会把任何接口名称或类型固定为校园网。校园网可以位于 WLAN、以太网或其他 IPv4 接口,配置应指向实际承载校园网的接口索引。配置接口未连接、没有可用路由或无法完成探测时,流程保持 `UNKNOWN`,且不会创建无绑定连接器或自动回退到其他网络接口。交互式 CLI 会在第一次探测会话关闭后列出 Windows IPv4 接口;只有用户显式选择后,才以 `dataclasses.replace()` 创建仅用于当前进程的运行时配置并建立新的绑定会话,磁盘配置和凭据不被改写。用户取消、接口枚举失败或终端非交互时以退出码 `2` 停止。提示中不暴露 `aiohttp`、主机、SSL 或本地化 socket 异常细节。识别门户后的 HTTP 超时或连接错误同样由 CLI 与 GUI 共用的异常分类器转换为通用接口提示,不直接显示底层传输异常。接口已通过独立在线指纹时只说明该接口已经联网;如果校园网位于另一接口,用户仍需改选对应的 IPv4 接口。 连接器同时固定 `AF_INET` 并启用 `force_close=True`:当前流程只创建 IPv4 socket,也不复用持久连接。 @@ -329,19 +329,20 @@ User-Agent、探测指纹、入口路径、验证码策略常量、确认间隔 | 文件 | 职责 | | --- | --- | -| [`main.py`](../main.py) | CLI 配置查找、参数解析和顶层退出码处理 | +| [`main.py`](../main.py) | CLI 配置查找、参数解析、交互式接口选择和顶层退出码处理 | | [`gui_main.py`](../gui_main.py) | GUI 参数解析和独立可执行文件入口 | | [`campus_net/application.py`](../campus_net/application.py) | CLI 与 GUI 共用的协议分派和错误分类 | | [`campus_net/config.py`](../campus_net/config.py) | `version` 分派、配置校验、历史配置兼容和内部默认值 | | [`campus_net/config_paths.py`](../campus_net/config_paths.py) | 源码与打包态共用的配置候选路径和 `dist` 父项目目录回退 | | [`campus_net/config_store.py`](../campus_net/config_store.py) | 正式配置编辑、并发修订检查、校验备份和原子保存 | | [`campus_net/captive.py`](../campus_net/captive.py) | Windows 接口绑定、captive URL 提取、在线指纹分类 | +| [`campus_net/interfaces.py`](../campus_net/interfaces.py) | CLI 与 GUI 共用的 Windows IPv4 接口枚举、状态归一化和排序 | | [`campus_net/aggregate.py`](../campus_net/aggregate.py) | 聚合门户工作流、验证码策略、运营商选择和在线确认 | | [`campus_net/sso.py`](../campus_net/sso.py) | SSO 页面解析、AES 表单、验证码 URL 和登录结果判定 | | [`campus_net/interactive.py`](../campus_net/interactive.py) | CLI 原生验证码窗口和无 GUI 时的临时文件回退 | -| [`campus_net/runner.py`](../campus_net/runner.py) | 串联新版流程并处理不确定 POST 的只读状态核对 | +| [`campus_net/runner.py`](../campus_net/runner.py) | 串联新版流程、运行时接口重试和不确定 POST 的只读状态核对 | | [`campus_net/legacy.py`](../campus_net/legacy.py) | 保留旧版 ePortal 登录流程 | -| [`campus_net/gui_core.py`](../campus_net/gui_core.py) | GUI 后台任务、事件队列、验证码桥接、接口枚举和表单校验 | +| [`campus_net/gui_core.py`](../campus_net/gui_core.py) | GUI 后台任务、事件队列、验证码桥接和表单校验 | | [`campus_net/gui.py`](../campus_net/gui.py) | `ttkbootstrap` 主线程窗口、配置编辑和用户交互 | | [`campus_net/window_state.py`](../campus_net/window_state.py) | 窗口位置、尺寸、最大化状态的无敏感信息持久化与多显示器校正 | @@ -349,7 +350,7 @@ User-Agent、探测指纹、入口路径、验证码策略常量、确认间隔 图形界面和 CLI 共用 `application.execute_config()`,不会维护第二套校园网协议实现。Tk/`ttkbootstrap` 控件只在主线程创建和更新;连接任务在单独线程中运行自己的 asyncio 事件循环,通过队列上报日志、结果和验证码图片。验证码答案由请求 ID 关联回当前异步任务,关闭验证码窗口会取消当前连接,而不会提交空答案。取消发生在请求提交附近时,界面会明确提示服务端可能已经处理该请求,不把本地取消等同于服务端回滚。 -IPv4 接口枚举固定调用 `%SystemRoot%\System32\WindowsPowerShell\v1.0\powershell.exe`,避免当前目录中的同名程序参与 Windows 可执行文件搜索。PowerShell 枚举值在 JSON 中可能是 `0/1` 数字,也可能是状态字符串;解析层统一归一化,再在界面显示为“已连接”“未连接”或“正在认证”。枚举进程超时、失败或后台线程无法启动时都会恢复刷新按钮,用户仍可直接填写接口索引。 +CLI 与 GUI 共用的 IPv4 接口枚举固定调用 `%SystemRoot%\System32\WindowsPowerShell\v1.0\powershell.exe`,避免当前目录中的同名程序参与 Windows 可执行文件搜索。PowerShell 枚举值在 JSON 中可能是 `0/1` 数字,也可能是状态字符串;解析层统一归一化,再显示为“已连接”“未连接”或“正在认证”。GUI 枚举进程超时、失败或后台线程无法启动时都会恢复刷新按钮,用户仍可直接填写接口索引。CLI 只有在标准输入和标准输出都连接真实终端时才提供选择;管道、重定向、计划任务或 EOF 不会阻塞等待输入。 窗口关闭时把普通窗口的 `x/y`、宽高和最大化状态原子写入 `%LOCALAPPDATA%\Cedarflake\CampusNet\window-state.json`。该独立 UI 状态不含配置或登录数据,读取失败会静默回退默认布局。恢复时按 Windows 各显示器工作区检查负坐标和可见范围;原显示器不存在时回到主屏居中,尺寸同时受最小窗口和当前工作区约束。 @@ -386,7 +387,7 @@ PyInstaller 保留两个独立入口:`build.spec` 生成控制台版 `Auto-Con 单元测试按协议边界拆分: - `test_captive.py`:真实脚本形态、入口参数、origin/path、歧义响应、在线指纹、Windows socket 绑定和探测异常脱敏; -- `test_runner.py`:配置接口不可达时的退出码、CLI/GUI 共用提示、只使用绑定连接器和禁止其他网络回退; +- `test_runner.py`:配置接口不可达时的退出码、显式运行时改选、配置不变、只使用绑定连接器和禁止自动回退; - `test_sso.py`:动态字段、AES/PKCS#7、验证码策略、风险类型、成功回调和错误码; - `test_aggregate.py`:完整会话顺序、CSRF 头、验证码刷新、非幂等请求不重放、运营商和双重在线判定; - `test_captive_config.py`:`version=2` 六字段配置和早期嵌套配置兼容; @@ -394,6 +395,7 @@ PyInstaller 保留两个独立入口:`build.spec` 生成控制台版 `Auto-Con - `test_config_store.py`:正式配置解析、修订冲突、强制备份、完整性校验和原子保存; - `test_config_paths.py`:源码和打包态候选顺序、`dist` 父目录、显式路径与符号链接边界; - `test_gui_core.py`:无 Tk 的表单、后台任务、取消、验证码桥接和接口列表测试; +- `test_main.py`:CLI 终端判定、通用接口列表、输入校验、取消和非交互边界; - `test_window_state.py`:窗口状态原子保存、损坏回退、负坐标多显示器和离屏校正; - `test_application.py`:CLI/GUI 共用协议分派、错误分类和后续网络异常脱敏。 diff --git a/workbench/network/campus-net/main.py b/workbench/network/campus-net/main.py index 9624f4a..69be519 100644 --- a/workbench/network/campus-net/main.py +++ b/workbench/network/campus-net/main.py @@ -3,11 +3,18 @@ import argparse import asyncio import json +import subprocess import sys +from collections.abc import Callable from typing import Any from campus_net.application import classify_run_error, execute_config from campus_net.config_paths import resolve_config_path +from campus_net.interfaces import NetworkInterface, list_windows_ipv4_interfaces + +InputReader = Callable[[str], str] +OutputWriter = Callable[[str], None] +InterfaceLoader = Callable[[], list[NetworkInterface]] def load_config() -> dict[str, Any]: @@ -34,9 +41,60 @@ def parse_args() -> argparse.Namespace: return parser.parse_args() +def is_interactive_terminal() -> bool: + for stream in (sys.stdin, sys.stdout): + try: + if stream is None or not stream.isatty(): + return False + except (AttributeError, OSError): + return False + return True + + +def prompt_interface_selection( + current_index: int, + *, + interface_loader: InterfaceLoader = list_windows_ipv4_interfaces, + input_reader: InputReader = input, + output_writer: OutputWriter = print, +) -> int | None: + try: + interfaces = interface_loader() + except (OSError, ValueError, RuntimeError, subprocess.SubprocessError) as error: + output_writer(f"无法读取 Windows IPv4 接口列表:{error}") + return None + + output_writer("当前 Windows IPv4 接口(已连接接口优先):") + for position, interface in enumerate(interfaces, start=1): + current_marker = "(当前配置)" if interface.index == current_index else "" + output_writer(f" [{position}] {interface.display_name}{current_marker}") + + while True: + try: + answer = input_reader("请选择接口序号(直接回车或输入 0 取消):").strip() + except (EOFError, KeyboardInterrupt): + output_writer("接口选择已取消。") + return None + if answer.casefold() in {"", "0", "q", "quit"}: + return None + try: + position = int(answer) + except ValueError: + output_writer(f"请输入 1 到 {len(interfaces)} 之间的序号,或输入 0 取消。") + continue + if 1 <= position <= len(interfaces): + return interfaces[position - 1].index + output_writer(f"请输入 1 到 {len(interfaces)} 之间的序号,或输入 0 取消。") + + async def async_main(*, probe_only: bool = False) -> int: cfg = load_config() - return await execute_config(cfg, probe_only=probe_only) + interface_selector = prompt_interface_selection if is_interactive_terminal() else None + return await execute_config( + cfg, + probe_only=probe_only, + interface_selector=interface_selector, + ) def main() -> int: diff --git a/workbench/network/campus-net/tests/test_application.py b/workbench/network/campus-net/tests/test_application.py index ebd796c..f391d92 100644 --- a/workbench/network/campus-net/tests/test_application.py +++ b/workbench/network/campus-net/tests/test_application.py @@ -71,6 +71,7 @@ async def test_dispatches_version_2_to_captive_runner(self, run_captive_http): run_captive_http.await_args.kwargs["status_callback"], status_callback, ) + self.assertIsNone(run_captive_http.await_args.kwargs["interface_selector"]) @patch("campus_net.application.run_legacy", new_callable=AsyncMock) async def test_dispatches_version_1_to_legacy_runner(self, run_legacy): diff --git a/workbench/network/campus-net/tests/test_gui_core.py b/workbench/network/campus-net/tests/test_gui_core.py index 546dd87..0a55a40 100644 --- a/workbench/network/campus-net/tests/test_gui_core.py +++ b/workbench/network/campus-net/tests/test_gui_core.py @@ -28,13 +28,12 @@ FinishedEvent, GuiEvent, LogEvent, - NetworkInterface, OperationController, _PendingCaptcha, build_config_from_form, form_values_from_config, - parse_windows_ipv4_interfaces, ) +from campus_net.interfaces import NetworkInterface, parse_windows_ipv4_interfaces def captive_values() -> dict[str, str]: diff --git a/workbench/network/campus-net/tests/test_main.py b/workbench/network/campus-net/tests/test_main.py new file mode 100644 index 0000000..5f25a67 --- /dev/null +++ b/workbench/network/campus-net/tests/test_main.py @@ -0,0 +1,156 @@ +import unittest +from unittest.mock import AsyncMock, Mock, patch + +import main +from campus_net.interfaces import NetworkInterface + + +class _TerminalStream: + def __init__(self, is_terminal: bool, *, raises: bool = False) -> None: + self.is_terminal = is_terminal + self.raises = raises + + def isatty(self) -> bool: + if self.raises: + raise OSError("stream unavailable") + return self.is_terminal + + +class TestInteractiveTerminal(unittest.TestCase): + def test_requires_both_input_and_output_terminals(self): + for stdin_terminal, stdout_terminal, expected in ( + (True, True, True), + (False, True, False), + (True, False, False), + ): + with ( + self.subTest( + stdin_terminal=stdin_terminal, + stdout_terminal=stdout_terminal, + ), + patch.object(main.sys, "stdin", _TerminalStream(stdin_terminal)), + patch.object(main.sys, "stdout", _TerminalStream(stdout_terminal)), + ): + self.assertIs(main.is_interactive_terminal(), expected) + + def test_handles_missing_or_unavailable_streams(self): + with patch.object(main.sys, "stdin", None): + self.assertFalse(main.is_interactive_terminal()) + with ( + patch.object(main.sys, "stdin", _TerminalStream(True, raises=True)), + patch.object(main.sys, "stdout", _TerminalStream(True)), + ): + self.assertFalse(main.is_interactive_terminal()) + + +class TestInterfaceSelectionPrompt(unittest.TestCase): + def test_lists_generic_interfaces_and_accepts_explicit_selection(self): + interfaces = [ + NetworkInterface(31, "WLAN", "Connected", 25), + NetworkInterface(8, "以太网", "Connected", 35), + NetworkInterface(44, "USB 网络共享", "Disconnected", 5), + ] + answers = iter(("not-a-number", "9", "2")) + output: list[str] = [] + + selected_index = main.prompt_interface_selection( + 24, + interface_loader=lambda: interfaces, + input_reader=lambda _prompt: next(answers), + output_writer=output.append, + ) + + self.assertEqual(selected_index, 8) + self.assertIn("31 · WLAN · 已连接", "\n".join(output)) + self.assertIn("8 · 以太网 · 已连接", "\n".join(output)) + self.assertIn("44 · USB 网络共享 · 未连接", "\n".join(output)) + self.assertEqual(sum("请输入 1 到 3" in line for line in output), 2) + + def test_marks_current_interface_but_never_auto_selects_it(self): + output: list[str] = [] + + selected_index = main.prompt_interface_selection( + 24, + interface_loader=lambda: [ + NetworkInterface(24, "WLAN", "Connected", 25), + ], + input_reader=lambda _prompt: "", + output_writer=output.append, + ) + + self.assertIsNone(selected_index) + self.assertIn("(当前配置)", "\n".join(output)) + + def test_eof_and_keyboard_interrupt_cancel_cleanly(self): + interfaces = [NetworkInterface(31, "WLAN", "Connected", 25)] + for error in (EOFError(), KeyboardInterrupt()): + with self.subTest(error_type=type(error).__name__): + output: list[str] = [] + + def raise_input(_prompt: str) -> str: + raise error + + selected_index = main.prompt_interface_selection( + 24, + interface_loader=lambda: interfaces, + input_reader=raise_input, + output_writer=output.append, + ) + + self.assertIsNone(selected_index) + self.assertEqual(output[-1], "接口选择已取消。") + + def test_enumeration_failure_preserves_safe_exit(self): + output: list[str] = [] + + def fail_loader() -> list[NetworkInterface]: + raise OSError("PowerShell unavailable") + + selected_index = main.prompt_interface_selection( + 24, + interface_loader=fail_loader, + input_reader=Mock(side_effect=AssertionError("must not prompt")), + output_writer=output.append, + ) + + self.assertIsNone(selected_index) + self.assertEqual( + output, + ["无法读取 Windows IPv4 接口列表:PowerShell unavailable"], + ) + + +class TestAsyncMain(unittest.IsolatedAsyncioTestCase): + @patch("main.execute_config", new_callable=AsyncMock) + @patch("main.load_config") + @patch("main.is_interactive_terminal") + async def test_enables_selector_only_for_interactive_terminal( + self, + is_interactive_terminal, + load_config, + execute_config, + ): + config = {"version": 2} + load_config.return_value = config + execute_config.return_value = 2 + + for is_interactive, expected_selector in ( + (True, main.prompt_interface_selection), + (False, None), + ): + with self.subTest(is_interactive=is_interactive): + is_interactive_terminal.return_value = is_interactive + execute_config.reset_mock() + + exit_code = await main.async_main(probe_only=True) + + self.assertEqual(exit_code, 2) + execute_config.assert_awaited_once_with( + config, + probe_only=True, + interface_selector=expected_selector, + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/workbench/network/campus-net/tests/test_runner.py b/workbench/network/campus-net/tests/test_runner.py index 22f6eaf..bf3eabc 100644 --- a/workbench/network/campus-net/tests/test_runner.py +++ b/workbench/network/campus-net/tests/test_runner.py @@ -1,5 +1,5 @@ import unittest -from unittest.mock import AsyncMock, patch, sentinel +from unittest.mock import AsyncMock, Mock, call, patch, sentinel from campus_net.aggregate import InitializationResult from campus_net.captive import NetworkState, ProbeResult @@ -39,6 +39,19 @@ async def __aexit__(self, _error_type, _error, _traceback): return False +class TrackingSessionContext: + def __init__(self, session): + self.session = session + self.exited = False + + async def __aenter__(self): + return self.session + + async def __aexit__(self, _error_type, _error, _traceback): + self.exited = True + return False + + class TestCaptiveRunner(unittest.IsolatedAsyncioTestCase): async def test_online_selected_interface_does_not_claim_campus_identity(self): config = http_config() @@ -73,8 +86,8 @@ async def test_online_selected_interface_does_not_claim_campus_identity(self): self.assertEqual( status_messages, [ - "正在通过配置的 IPv4 接口探测校园网状态…", - "配置的 IPv4 接口 24 已通过独立连通性探测,无需登录;" + "正在通过 IPv4 接口 24 探测校园网状态…", + "IPv4 接口 24 已通过独立连通性探测,无需登录;" "如果校园网实际位于另一接口,请重新选择对应的 IPv4 接口。", ], ) @@ -125,8 +138,8 @@ async def test_unreachable_interface_stops_without_falling_back_to_other_network self.assertEqual( status_messages, [ - "正在通过配置的 IPv4 接口探测校园网状态…", - "无法确认配置的 IPv4 接口 24 的网络状态:配置接口未连接,或该接口当前没有可用路由。" + "正在通过 IPv4 接口 24 探测校园网状态…", + "无法确认 IPv4 接口 24 的网络状态:配置接口未连接,或该接口当前没有可用路由。" "请确认该接口已接入校园网,或重新选择实际承载校园网的 IPv4 接口;" "已停止,未改用其他网络接口。", ], @@ -145,6 +158,161 @@ async def test_unreachable_interface_stops_without_falling_back_to_other_network client.trigger_online_check.assert_not_awaited() client.verify_online.assert_not_awaited() + async def test_closes_old_session_before_retrying_with_fresh_network_state(self): + config = http_config() + first_context = TrackingSessionContext(sentinel.first_session) + second_context = TrackingSessionContext(sentinel.second_session) + first_client = Mock() + first_client.initialize = AsyncMock( + return_value=InitializationResult( + probe=ProbeResult( + NetworkState.UNKNOWN, + reason="配置接口未连接,或该接口当前没有可用路由", + ) + ) + ) + second_client = Mock() + second_client.initialize = AsyncMock( + return_value=InitializationResult( + probe=ProbeResult(NetworkState.ONLINE), + ) + ) + + def select_interface(current_index: int) -> int: + self.assertEqual(current_index, 24) + self.assertTrue(first_context.exited) + self.assertFalse(second_context.exited) + return 31 + + with ( + patch( + "campus_net.runner.create_bound_connector", + side_effect=(sentinel.first_connector, sentinel.second_connector), + ) as create_connector, + patch( + "campus_net.runner.aiohttp.CookieJar", + side_effect=(sentinel.first_cookie_jar, sentinel.second_cookie_jar), + ) as cookie_jar_type, + patch( + "campus_net.runner.aiohttp.ClientSession", + side_effect=(first_context, second_context), + ) as session_type, + patch( + "campus_net.runner.AggregatePortalClient", + side_effect=(first_client, second_client), + ) as client_type, + ): + exit_code = await run_captive_http( + config, + interface_selector=select_interface, + status_callback=lambda _message: None, + ) + + self.assertEqual(exit_code, 0) + self.assertTrue(first_context.exited) + self.assertTrue(second_context.exited) + self.assertEqual(config.interface_index, 24) + create_connector.assert_has_calls([call(24), call(31)]) + self.assertEqual(cookie_jar_type.call_count, 2) + self.assertEqual(session_type.call_count, 2) + self.assertIs(client_type.call_args_list[0].args[0], sentinel.first_session) + self.assertEqual(client_type.call_args_list[0].args[1].interface_index, 24) + self.assertIs(client_type.call_args_list[1].args[0], sentinel.second_session) + self.assertEqual(client_type.call_args_list[1].args[1].interface_index, 31) + + @patch("campus_net.runner._run_captive_http_attempt", new_callable=AsyncMock) + async def test_explicit_selection_retries_with_runtime_only_interface( + self, + run_attempt, + ): + config = http_config() + run_attempt.side_effect = [ + (2, "配置接口未连接,或该接口当前没有可用路由"), + (0, None), + ] + interface_selector = Mock(return_value=31) + status_messages: list[str] = [] + + exit_code = await run_captive_http( + config, + probe_only=True, + interface_selector=interface_selector, + status_callback=status_messages.append, + ) + + self.assertEqual(exit_code, 0) + self.assertEqual(config.interface_index, 24) + self.assertEqual(run_attempt.await_count, 2) + first_config = run_attempt.await_args_list[0].args[0] + second_config = run_attempt.await_args_list[1].args[0] + self.assertIs(first_config, config) + self.assertEqual(second_config.interface_index, 31) + self.assertEqual(second_config.username, config.username) + self.assertEqual(second_config.password, config.password) + for attempt_call in run_attempt.await_args_list: + self.assertTrue(attempt_call.kwargs["probe_only"]) + interface_selector.assert_called_once_with(24) + self.assertEqual( + status_messages, + [ + "无法确认 IPv4 接口 24 的网络状态:配置接口未连接,或该接口当前没有可用路由。" + "可从当前 Windows IPv4 接口列表中显式选择一个接口重新探测;" + "程序不会自动选择,也不会修改 config.json。", + "已显式选择 IPv4 接口 31,仅本次运行使用;config.json 未修改,正在重新探测。", + ], + ) + + @patch("campus_net.runner._run_captive_http_attempt", new_callable=AsyncMock) + async def test_cancelled_selection_preserves_safe_exit(self, run_attempt): + run_attempt.return_value = (2, "探测结果无法识别") + interface_selector = Mock(return_value=None) + status_messages: list[str] = [] + + exit_code = await run_captive_http( + http_config(), + interface_selector=interface_selector, + status_callback=status_messages.append, + ) + + self.assertEqual(exit_code, 2) + run_attempt.assert_awaited_once() + interface_selector.assert_called_once_with(24) + self.assertEqual(status_messages[-1], "已取消接口选择;已停止,未改用其他网络接口。") + + @patch("campus_net.runner._run_captive_http_attempt", new_callable=AsyncMock) + async def test_second_unknown_prompts_again_until_user_cancels(self, run_attempt): + run_attempt.side_effect = [ + (2, "配置接口未连接,或该接口当前没有可用路由"), + (2, "响应不匹配 captive 或精确在线指纹"), + ] + interface_selector = Mock(side_effect=(31, None)) + + exit_code = await run_captive_http( + http_config(), + interface_selector=interface_selector, + status_callback=lambda _message: None, + ) + + self.assertEqual(exit_code, 2) + self.assertEqual( + [attempt.args[0].interface_index for attempt in run_attempt.await_args_list], + [24, 31], + ) + self.assertEqual(interface_selector.call_args_list, [call(24), call(31)]) + + @patch("campus_net.runner._run_captive_http_attempt", new_callable=AsyncMock) + async def test_rejects_invalid_interface_selection(self, run_attempt): + run_attempt.return_value = (2, "探测结果无法识别") + + for selected_index in (True, 0, 0x1_0000_0000, "24"): + with self.subTest(selected_index=selected_index): + with self.assertRaisesRegex(ValueError, "interface_index"): + await run_captive_http( + http_config(), + interface_selector=Mock(return_value=selected_index), + status_callback=lambda _message: None, + ) + if __name__ == "__main__": unittest.main()