From 6fb29266274f4e47e1ef06594fb5feb4d455f532 Mon Sep 17 00:00:00 2001 From: RoyirinRosmontis Date: Wed, 16 Sep 2026 23:20:16 +0800 Subject: [PATCH] =?UTF-8?q?=E6=B8=85=E7=90=86=EF=BC=9A=E7=A7=BB=E9=99=A4?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E8=BE=85=E5=8A=A9=E8=84=9A=E6=9C=AC=E4=B8=8E?= =?UTF-8?q?=E6=B5=8B=E8=AF=95=E5=B7=A5=E5=85=B7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .gitignore | 4 - lite/MODULE_SPLIT_PLAN.md | 148 ---- lite/build_pyinstaller.bat | 12 - lite/claude.bat | 12 - lite/others/nodes_tool/nodes_tool.py | 126 --- .../update_pac_domains/update_pac_domains.py | 143 ---- .../update_pac_ip_cn/update_pac_ip_cn.py | 86 -- lite/test.bat | 4 - lite/test/PyQt5/QtCore.py | 10 - lite/test/PyQt5/QtGui.py | 6 - lite/test/PyQt5/QtWidgets.py | 28 - lite/test/PyQt5/__init__.py | 2 - lite/test/check_node_menu.py | 127 --- lite/test/nodes.json | 21 - lite/test/probe_nodes.py | 151 ---- lite/test/test_core_modules.py | 114 --- lite/test/test_pac_routing.py | 739 ------------------ 17 files changed, 1733 deletions(-) delete mode 100644 lite/MODULE_SPLIT_PLAN.md delete mode 100644 lite/build_pyinstaller.bat delete mode 100644 lite/claude.bat delete mode 100644 lite/others/nodes_tool/nodes_tool.py delete mode 100644 lite/others/update_pac_domains/update_pac_domains.py delete mode 100644 lite/others/update_pac_ip_cn/update_pac_ip_cn.py delete mode 100644 lite/test.bat delete mode 100644 lite/test/PyQt5/QtCore.py delete mode 100644 lite/test/PyQt5/QtGui.py delete mode 100644 lite/test/PyQt5/QtWidgets.py delete mode 100644 lite/test/PyQt5/__init__.py delete mode 100644 lite/test/check_node_menu.py delete mode 100644 lite/test/nodes.json delete mode 100644 lite/test/probe_nodes.py delete mode 100644 lite/test/test_core_modules.py delete mode 100644 lite/test/test_pac_routing.py diff --git a/.gitignore b/.gitignore index 6def320..0f4c7f8 100644 --- a/.gitignore +++ b/.gitignore @@ -7,7 +7,3 @@ lite/.claude/ lite/backups/ lite/build/ lite/dist/ - -# Runtime PAC downloads generated by the UI smoke test. -lite/test/pac_direct_domains.json -lite/test/pac_domains.json diff --git a/lite/MODULE_SPLIT_PLAN.md b/lite/MODULE_SPLIT_PLAN.md deleted file mode 100644 index 06db1d5..0000000 --- a/lite/MODULE_SPLIT_PLAN.md +++ /dev/null @@ -1,148 +0,0 @@ -# RRPP Lite 模块拆分计划 - -## 目标与边界 - -将当前 `main.py`（约 3,480 行）渐进拆成职责清晰、可独立测试的模块，同时保持以下行为不变： - -- PAC、全局代理、智能分流和本地 TLS 转发的现有接口与优先级； -- 节点订阅、DPAPI 加密存储、登录及本地运行数据位置； -- 源码运行、PyInstaller 单文件版和 Nuitka 版的启动方式； -- 现有 `lite/test/test_pac_routing.py` 的全部回归用例。 - -本轮拆分期间**不提交、不推送到主仓库、不打包发布**。每个阶段均只在本地工作区验证；出现回归时优先回退到拆分前备份。 - -## 开始前的保护措施 - -1. 记录当前 Git 状态与当前 HEAD，不触碰既有未跟踪的构建产物、备份和用户文件。 -2. 在 `lite/backups/` 创建带时间戳的源码备份，例如： - `proxy_lite.py.bak_YYYYMMDD_HHMMSS_before_module_split`。 -3. 首次拆分前运行一次完整基线： - - ```powershell - wsl.exe -e bash -lc "cd '/mnt/d/Royirin Rosmontis Private Proxy' && python3 -m py_compile lite/main.py lite/test/test_pac_routing.py && python3 lite/test/test_pac_routing.py" - ``` - -4. 每完成一个阶段，先做编译检查，再跑完整 PAC 回归；不通过则在进入下一阶段前修复或回退。 - -## 当前实施进度（2026-09-16） - -- [x] 已创建拆分前备份：`lite/backups/proxy_lite.py.bak_20260916_220240_before_module_split`。 -- [x] 已完成一次拆分前基线与多次拆分后全量 PAC 回归。 -- [x] UI 主题、控件和三个对话框已迁入 `rrpp_ui/theme.py`、`rrpp_ui/widgets.py`、 - `rrpp_ui/dialogs.py`；通过显式服务对象调用业务层，不反向导入入口文件。 -- [x] 已迁入核心包的无状态功能：CIDR/Trie 与 PAC 规则解析、PAC 拉取和原子写入、 - `settings.json` 原子存储、DPAPI 节点/凭据存储、HTTP 转发报文解析与认证注入、 - DNS/私网判断与 TTL 缓存、流量统计、节点认证请求协议。 -- [ ] 节点同步服务、带缓存的完整路由决策、转发器 socket 生命周期、系统代理注册表控制 - 和主窗口仍在入口文件，待后续阶段继续迁移。 -- [x] 已获得用户授权，在完成构建与验证后执行 `git commit`、`git push` 和 Release 发布。 - -## 目标目录 - -```text -lite/ -├─ main.py # 轻量入口、组装依赖、main -├─ rrpp_core/ -│ ├─ __init__.py -│ ├─ runtime.py # 路径、常量、日志与共享运行时上下文 -│ ├─ config_store.py # 普通配置、节点、DPAPI 凭据 -│ ├─ pac_data.py # PAC 在线数据下载、校验、缓存失效 -│ ├─ routing.py # 域名/IP 分流、白名单、CIDR/Trie -│ ├─ forwarder.py # 本地 TLS 转发、HTTP CONNECT、认证注入 -│ └─ system_proxy.py # PAC 服务、Windows 系统代理状态 -├─ rrpp_ui/ -│ ├─ __init__.py -│ ├─ theme.py # UI 色彩与样式常量 -│ ├─ widgets.py # NodeLabel、开关、状态点等可复用控件 -│ └─ dialogs.py # 登录、白名单、设置对话框 -└─ test/ - └─ test_pac_routing.py -``` - -目录是目标形态，而不是一次性迁移要求。入口已统一为 `main.py`；测试和构建脚本均以该文件为准。 - -## 分阶段实施 - -### 阶段 0：基线与备份 - -- 创建源码备份并保存当前测试输出。 -- 核验当前核心测试：PAC 规则优先级、IPv4/IPv6、私网、CONNECT 解析、认证注入、启动连通性校验及其取消流程。 -- 产物：可随时恢复的单文件版本；无业务代码变更。 - -### 阶段 1：抽取无状态 UI 组件 - -- 将 `NodeLabel`、`ToggleSwitch`、`StatusDot` 及图标辅助逻辑移到 `rrpp_ui/widgets.py`。 -- 主窗口继续从原入口导入/调用，保持控件信号、样式和构造参数不变。 -- 风险低：不涉及代理线程、注册表或路由决策。 -- 验收：`py_compile`、PAC 全量测试、一次 Windows GUI 手工启动。 - -### 阶段 2：建立共享运行时与配置存储 - -- 抽取路径常量、配置读写、节点读写、DPAPI 加解密、认证保存和订阅节点同步至 `rrpp_core/runtime.py`、`config_store.py`。 -- 明确依赖方向：模块不得反向导入 UI；日志通过注入函数或统一运行时上下文提供。 -- 保持 `%LOCALAPPDATA%\\RRPP`、配置文件名、加密格式和节点选择恢复逻辑完全兼容。 -- 风险中等：涉及用户已有数据；只做机械迁移，不改变存储格式。 -- 验收：现有节点可读取；登录/同步至少做一次人工冒烟；全部自动测试通过。 - -### 阶段 3：抽取 PAC 数据与路由引擎 - -- 将 PAC 数据下载/校验/写入和缓存失效移至 `rrpp_core/pac_data.py`。 -- 将白名单、强制代理、明确直连域名、CIDR、IPv6、DNS 解析、Trie 匹配和直连决策移至 `rrpp_core/routing.py`。 -- 使用显式的 `RoutingContext` 或等价配置对象替代跨模块全局变量；规则优先级必须锁定为： - - ```text - 用户白名单 > 强制代理 > 明确直连域名 > LAN/私网 > GeoIP 智能分流 - ``` - -- 风险最高的核心阶段；禁止顺便修改规则语义或远程 PAC 格式。 -- 验收：扩充并运行分流回归测试，覆盖 Google/OpenAI/Anthropic、国内直连、IPv6、无 DNS 结果与缓存刷新。 - -### 阶段 4：抽取转发器与系统代理控制 - -- 将流量统计、本地 TLS 监听、CONNECT/绝对 URL 解析、认证注入与连接回退移至 `rrpp_core/forwarder.py`。 -- 将 PAC HTTP 服务、注册表备份/恢复、全局/PAC 开关和陈旧状态清理移至 `rrpp_core/system_proxy.py`。 -- 以明确的 start/stop 生命周期对象管理线程与端口，避免模块级状态互相覆盖。 -- 风险中高：涉及注册表和后台线程；本阶段不改变任何系统代理策略。 -- 验收：本地端口监听、PAC 开关、全局开关、取消启动、退出清理和重复启停的人工测试；完整回归测试。 - -### 阶段 5：抽取对话框与收敛入口 - -- 将 `LoginDialog`、`WhitelistDialog`、`SettingsDialog` 移至 `rrpp_ui/dialogs.py`，`MainWindow` 可保留在入口或迁到 `rrpp_ui/main_window.py`。 -- `main.py` 收敛为：初始化 DPI/单实例、创建运行时、装配 UI、调用 `main()`。 -- 保留必要的兼容导出，待构建脚本和测试完全迁移后再移除。 -- 验收：源码 GUI 手工走查、全量自动测试、PyInstaller/Nuitka 各做一次仅本地的构建与启动冒烟。 - -### 阶段 6：清理与交付前评审 - -- 删除已确认无引用的兼容层，补齐模块文档和最小单元测试。 -- 检查循环导入、隐式全局状态、未关闭线程、路径在冻结环境下的差异。 -- 对比拆分前后的 PAC 输出和系统代理状态变更。 -- 仅在用户确认后，另行决定提交、推送、打包和发布；它们不属于本次拆分的自动后续动作。 - -## 依赖与架构约束 - -```text -main.py → rrpp_ui → rrpp_core - │ │ - └───────────┘（仅通过显式上下文/接口协作） -``` - -- `rrpp_core` 不依赖 PyQt；这样 PAC 与转发逻辑可在无 GUI 环境测试。 -- UI 不直接读写注册表、节点文件或 socket；统一通过核心服务对象操作。 -- 禁止新的模块级可变全局状态，已有状态在迁移时集中到上下文或生命周期对象。 -- 配置、加密和 PAC 文件的文件名、字段、URL 与优先级，在本次重构内视为兼容性契约。 - -## 验收清单 - -- [x] 源码 `py_compile` 通过。 -- [x] `test_pac_routing.py` 全部通过。 -- [ ] PAC 启动/取消、全局模式、退出恢复系统代理均可用。 -- [ ] 既有节点和已加密凭据可读取，订阅同步不回退格式。 -- [ ] `chatgpt.com`、`auth.openai.com`、`claude.ai`、Google CDN 仍命中强制代理；国内显式直连规则仍生效。 -- [x] PyInstaller 与 Nuitka 均已完成本地构建；Release 附件将使用版本化文件名和 SHA-256 校验。 -- [x] 发布暂存仅包含本轮源码、测试、构建脚本、图标和模板；构建产物、账号数据和备份文件已排除。 -- [ ] 待提交、推送并上传 Release 附件。 - -## 回滚策略 - -任一阶段出现无法在当前阶段定位的回归时：停止继续迁移，将该阶段涉及文件恢复到阶段开始前的本地备份或 Git 基线；保留失败现场的测试输出以便定位。不得通过覆盖用户的运行配置、订阅或构建目录来“修复”回归。 diff --git a/lite/build_pyinstaller.bat b/lite/build_pyinstaller.bat deleted file mode 100644 index c2f6137..0000000 --- a/lite/build_pyinstaller.bat +++ /dev/null @@ -1,12 +0,0 @@ -@echo off -cd /d "%~dp0" -call ..\.venv\Scripts\python.exe -m PyInstaller RRPP_Lite.spec -if errorlevel 1 ( - echo. - echo Build failed. - pause - exit /b 1 -) -echo. -echo Build complete. Output: dist\Royirin Rosmontis Private Proxy Lite.exe -pause diff --git a/lite/claude.bat b/lite/claude.bat deleted file mode 100644 index 1439940..0000000 --- a/lite/claude.bat +++ /dev/null @@ -1,12 +0,0 @@ -@echo off -setlocal -rem Convert current Windows dir to a WSL path (lowercase drive + forward slashes) -set "WSLDIR=%CD:\=/%" -set "WSLDIR=%WSLDIR::=%" -set "DRIVE=%WSLDIR:~0,1%" -for %%C in (a b c d e f g h i j k l m n o p q r s t u v w x y z) do if /i "%%C"=="%DRIVE%" set "WSLDIR=%%C%WSLDIR:~1%" -set "WSLDIR=/mnt/%WSLDIR%" -rem -i is essential: in a non-interactive shell PS1 is empty, so ~/.bashrc's -rem interactive guard returns early and the model-mapping env vars are lost -wsl -e bash -ic "cd '%WSLDIR%' && /root/.local/bin/claude --continue" -pause diff --git a/lite/others/nodes_tool/nodes_tool.py b/lite/others/nodes_tool/nodes_tool.py deleted file mode 100644 index 47ac9ce..0000000 --- a/lite/others/nodes_tool/nodes_tool.py +++ /dev/null @@ -1,126 +0,0 @@ -# -*- coding: utf-8 -*- -"""管理用户数据目录里的 nodes.json（Windows DPAPI 加密存储）。 - -用法（在 Windows 上运行）: - python nodes_tool.py encrypt <明文nodes.json> # 加密导入到 %LOCALAPPDATA%\\RRPP - python nodes_tool.py decrypt # 打印解密后的内容 - python nodes_tool.py path # 显示存储路径 - -注意：DPAPI 加密后只有「本机 + 本用户」能解密；文件拷到别的机器/账户无效。 -""" -import ctypes -import json -import os -import sys -from pathlib import Path - - -def user_data_dir(): - if os.name == "nt" and os.environ.get("LOCALAPPDATA"): - return Path(os.environ["LOCALAPPDATA"]) / "RRPP" - return Path.home() / "RRPP" - - -class _DATA_BLOB(ctypes.Structure): - _fields_ = [("cbData", ctypes.c_uint32), - ("pbData", ctypes.POINTER(ctypes.c_char))] - - -def dpapi_protect(data: bytes) -> bytes: - crypt32 = ctypes.WinDLL("crypt32") - kernel32 = ctypes.WinDLL("kernel32") - crypt32.CryptProtectData.argtypes = [ctypes.POINTER(_DATA_BLOB), ctypes.c_wchar_p, - ctypes.POINTER(_DATA_BLOB), ctypes.c_void_p, - ctypes.c_void_p, ctypes.c_uint32, - ctypes.POINTER(_DATA_BLOB)] - crypt32.CryptProtectData.restype = ctypes.c_bool - kernel32.LocalFree.argtypes = [ctypes.c_void_p] - - buf = ctypes.create_string_buffer(data, len(data)) - blob_in = _DATA_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char))) - blob_out = _DATA_BLOB() - if not crypt32.CryptProtectData(ctypes.byref(blob_in), "RRPP nodes.json", None, - None, None, 0, ctypes.byref(blob_out)): - raise OSError("CryptProtectData 调用失败") - try: - return ctypes.string_at(blob_out.pbData, blob_out.cbData) - finally: - kernel32.LocalFree(blob_out.pbData) - - -def dpapi_unprotect(data: bytes) -> bytes: - crypt32 = ctypes.WinDLL("crypt32") - kernel32 = ctypes.WinDLL("kernel32") - crypt32.CryptUnprotectData.argtypes = [ctypes.POINTER(_DATA_BLOB), ctypes.POINTER(ctypes.c_wchar_p), - ctypes.POINTER(_DATA_BLOB), ctypes.c_void_p, - ctypes.c_void_p, ctypes.c_uint32, - ctypes.POINTER(_DATA_BLOB)] - crypt32.CryptUnprotectData.restype = ctypes.c_bool - kernel32.LocalFree.argtypes = [ctypes.c_void_p] - - buf = ctypes.create_string_buffer(data, len(data)) - blob_in = _DATA_BLOB(len(data), ctypes.cast(buf, ctypes.POINTER(ctypes.c_char))) - descr = ctypes.c_wchar_p() - blob_out = _DATA_BLOB() - if not crypt32.CryptUnprotectData(ctypes.byref(blob_in), ctypes.byref(descr), None, - None, None, 0, ctypes.byref(blob_out)): - raise OSError("CryptUnprotectData 调用失败") - try: - return ctypes.string_at(blob_out.pbData, blob_out.cbData) - finally: - kernel32.LocalFree(blob_out.pbData) - if descr.value: - kernel32.LocalFree(descr) - - -def cmd_encrypt(src: str): - plain = Path(src) - if not plain.is_file(): - print(f"未找到 {plain}") - sys.exit(1) - data = json.loads(plain.read_text("utf-8")) - if not (isinstance(data, list) and data and all(isinstance(n, dict) for n in data)): - print("nodes.json 应为非空节点数组") - sys.exit(1) - out_dir = user_data_dir() - out_dir.mkdir(parents=True, exist_ok=True) - out = out_dir / "nodes.json" - text = json.dumps(data, ensure_ascii=False, indent=2) + "\n" - blob = dpapi_protect(text.encode("utf-8")) if os.name == "nt" else text.encode("utf-8") - out.write_bytes(blob) - print(f"已{'加密' if os.name == 'nt' else '写入(非Windows不加密)'}: {out}（{len(data)} 个节点）") - - -def cmd_decrypt(): - f = user_data_dir() / "nodes.json" - if not f.is_file(): - print(f"未找到 {f}") - sys.exit(1) - blob = f.read_bytes() - if os.name == "nt": - try: - print(dpapi_unprotect(blob).decode("utf-8"), end="") - return - except Exception as e: - print(f"DPAPI 解密失败: {e}") - print(blob.decode("utf-8"), end="") - - -def main(): - if len(sys.argv) < 2: - print(__doc__) - sys.exit(1) - cmd = sys.argv[1] - if cmd == "encrypt" and len(sys.argv) == 3: - cmd_encrypt(sys.argv[2]) - elif cmd == "decrypt": - cmd_decrypt() - elif cmd == "path": - print(user_data_dir()) - else: - print(__doc__) - sys.exit(1) - - -if __name__ == "__main__": - main() diff --git a/lite/others/update_pac_domains/update_pac_domains.py b/lite/others/update_pac_domains/update_pac_domains.py deleted file mode 100644 index 4310e32..0000000 --- a/lite/others/update_pac_domains/update_pac_domains.py +++ /dev/null @@ -1,143 +0,0 @@ -# -*- coding: utf-8 -*- -"""更新 pac_domains.json：从 v2fly/domain-list-community (geosite) 拉取境外服务 -分类，与现有手工条目合并，生成新的强制走隧道域名后缀表。 - -用法: - python update_pac_domains.py # 拉取并更新（先自动备份） - python update_pac_domains.py --dry-run # 只统计，不写文件 - -geosite 数据格式（data/<分类>，每行一条规则）: - include:xxx 引用其他分类（递归解析） - example.com 匹配域名及其子域（与我们的后缀匹配一致） - +example.com 同上（additive） - .example.com 仅精确匹配（不匹配子域）-> 跳过 - -example.com 排除 -> 跳过 - /regex/ 正则 -> 跳过 - example.com @cn 属性标签 -> 去掉标签 - 无点的单标签（如 chrome）-> 无实际主机名会命中，跳过 -""" -import argparse -import json -import sys -import urllib.request -from datetime import datetime -from pathlib import Path - -BASE_URL = "https://raw.githubusercontent.com/v2fly/domain-list-community/master/data/" -# 需要强制走隧道的境外服务分类（geosite 分类名），可按需增删 -CATEGORIES = ["google", "apple", "telegram", "openai", "twitter", "facebook", - "github", "netflix", "discord", "spotify", "whatsapp", "paypal"] - -SCRIPT_DIR = Path(__file__).resolve().parent -LITE_DIR = SCRIPT_DIR.parent.parent -PAC_FILE = LITE_DIR / "pac_domains.json" -BACKUP_DIR = LITE_DIR / "backups" -# 本地 pac_domains.json 已不再随项目保留：缺失时从 CDN 拉当前版本作合并基底 -CDN_URL = "https://cdn.rrrosmontis.icu/RRPP/PAC/pac_domains.json" - - -def fetch(cat, tries=2): - for i in range(tries): - try: - with urllib.request.urlopen(BASE_URL + cat, timeout=20) as r: - return r.read().decode("utf-8", "replace").splitlines() - except Exception: - if i == tries - 1: - raise - - -def resolve(cat, seen=None): - """递归展开 include:，返回原始行列表；循环引用用 seen 防死循环。""" - seen = seen or set() - if cat in seen: - return [] - seen.add(cat) - lines = [] - try: - for raw in fetch(cat): - line = raw.strip() - if line.startswith("include:"): - lines += resolve(line[len("include:"):].strip(), seen) - else: - lines.append(line) - except Exception as e: - print(f" [warn] 拉取 {cat} 失败: {e}") - return lines - - -def parse_domains(lines): - """过滤 geosite 行，只保留可映射为后缀匹配的纯域名。""" - out = set() - for line in lines: - s = line.strip() - if not s or s.startswith("#") or s.startswith("/"): - continue - if s.startswith("-") or s.startswith("."): - continue - s = s.lstrip("+").split("@")[0].strip() - if "." not in s: - continue - out.add(s.lower()) - return out - - -def main(): - parser = argparse.ArgumentParser(description="更新 pac_domains.json") - parser.add_argument("--dry-run", action="store_true", help="只统计，不写文件") - args = parser.parse_args() - - if PAC_FILE.is_file(): - cur = set(json.loads(PAC_FILE.read_text("utf-8"))["domains"]) - else: - # 项目目录已不保留该文件：从 CDN 拉当前版本作合并基底（保留其中的手工条目） - print(f"本地未找到 {PAC_FILE}，从 CDN 拉取当前版本作为合并基底") - try: - with urllib.request.urlopen(CDN_URL, timeout=20) as r: - cur = set(json.loads(r.read().decode("utf-8"))["domains"]) - except Exception as e: - print(f"拉取失败: {e}") - sys.exit(1) - - print(f"拉取 geosite 分类: {', '.join(CATEGORIES)}") - merged = set() - for cat in CATEGORIES: - doms = parse_domains(resolve(cat)) - merged |= doms - print(f" {cat:<12} -> {len(doms):>5} 条") - - new = merged - cur - kept = cur & merged - only_cur = cur - merged - print(f"\n合并后总数: {len(merged | cur)}（geosite {len(merged)} + 现有 {len(cur)}）") - print(f"新增 {len(new)}，与现有重叠 {len(kept)}，现有手工条目 {len(only_cur)} 条全部保留") - - if args.dry_run: - print("--dry-run: 未写文件") - return - - final = sorted(merged | cur) - now = datetime.now().strftime("%Y-%m-%d %H:%M") - comment = ( - f"强制走代理隧道的域名后缀（IP 智能分流的补充规则）：命中这些域名的连接即使" - f"解析到国内 IP 也强制走隧道。匹配规则：host == 条目 或 host 以 \".条目\" 结尾。" - f"由 others/update_pac_domains/update_pac_domains.py 从 v2fly geosite " - f"({', '.join(CATEGORIES)}) 生成于 {now}，手工条目已保留；" - f"客户端按文件内 version/date 从 CDN 在线更新到 %LOCALAPPDATA%\\RRPP。" - ) - out = {"_comment": comment, - "version": datetime.now().strftime("%Y%m%d%H%M"), - "date": datetime.now().strftime("%Y-%m-%d"), - "domains": final} - - BACKUP_DIR.mkdir(exist_ok=True) - bak = BACKUP_DIR / f"pac_domains.json.bak_{datetime.now().strftime('%Y%m%d_%H%M%S')}" - bak.write_bytes(PAC_FILE.read_bytes()) - print(f"已备份: {bak}") - - PAC_FILE.write_text(json.dumps(out, ensure_ascii=False, indent=2) + "\n", "utf-8") - size = PAC_FILE.stat().st_size / 1024 - print(f"已写入: {PAC_FILE}（{len(final)} 条，{size:.0f} KB）") - - -if __name__ == "__main__": - main() diff --git a/lite/others/update_pac_ip_cn/update_pac_ip_cn.py b/lite/others/update_pac_ip_cn/update_pac_ip_cn.py deleted file mode 100644 index d5f8a5a..0000000 --- a/lite/others/update_pac_ip_cn/update_pac_ip_cn.py +++ /dev/null @@ -1,86 +0,0 @@ -# -*- coding: utf-8 -*- -"""更新 pac_ip_cn.json：从 17mon/china_ip_list 拉取中国 IPv4 段，写入带 -version/date 戳的本地 PAC 数据文件（客户端按 pac_manifest.json 比对在线更新）。 - -用法: - python update_pac_ip_cn.py # 拉取并更新（先自动备份） - python update_pac_ip_cn.py --dry-run # 只统计，不写文件 -""" -import argparse -import ipaddress -import json -import sys -import urllib.request -from datetime import datetime -from pathlib import Path - -SRC_URL = "https://raw.githubusercontent.com/17mon/china_ip_list/master/china_ip_list.txt" - -SCRIPT_DIR = Path(__file__).resolve().parent -LITE_DIR = SCRIPT_DIR.parent.parent -PAC_FILE = LITE_DIR / "pac_ip_cn.json" -BACKUP_DIR = LITE_DIR / "backups" - - -def fetch_lines(tries=2): - for i in range(tries): - try: - with urllib.request.urlopen(SRC_URL, timeout=20) as r: - return r.read().decode("utf-8", "replace").splitlines() - except Exception: - if i == tries - 1: - raise - - -def main(): - parser = argparse.ArgumentParser(description="更新 pac_ip_cn.json") - parser.add_argument("--dry-run", action="store_true", help="只统计，不写文件") - args = parser.parse_args() - - print(f"拉取: {SRC_URL}") - cidrs, skipped = [], 0 - for line in fetch_lines(): - s = line.strip() - if not s or s.startswith("#"): - continue - try: - net = ipaddress.ip_network(s, strict=False) - if net.version == 4: - cidrs.append(str(net)) - else: - skipped += 1 - except ValueError: - skipped += 1 - cidrs.sort(key=lambda c: (int(ipaddress.ip_network(c, strict=False).network_address), - ipaddress.ip_network(c, strict=False).prefixlen)) - print(f"共 {len(cidrs)} 条 IPv4 CIDR，跳过 {skipped} 条") - - if args.dry_run: - print("--dry-run: 未写文件") - return - - now = datetime.now() - comment = ( - f"中国 IPv4 地址段（CIDR）。智能分流依据：解析 IP 命中列表 → 直连，" - f"未命中 → 走加密隧道。由 others/update_pac_ip_cn/update_pac_ip_cn.py " - f"从 17mon/china_ip_list 生成于 {now.strftime('%Y-%m-%d %H:%M')}；" - f"客户端按文件内 version/date 从 CDN 在线更新到 %LOCALAPPDATA%\\RRPP。" - ) - out = {"_comment": comment, - "version": now.strftime("%Y%m%d%H%M"), - "date": now.strftime("%Y-%m-%d"), - "cidrs": cidrs} - - BACKUP_DIR.mkdir(exist_ok=True) - bak = BACKUP_DIR / f"pac_ip_cn.json.bak_{now.strftime('%Y%m%d_%H%M%S')}" - if PAC_FILE.is_file(): - bak.write_bytes(PAC_FILE.read_bytes()) - print(f"已备份: {bak}") - - PAC_FILE.write_text(json.dumps(out, ensure_ascii=False, indent=1) + "\n", "utf-8") - size = PAC_FILE.stat().st_size / 1024 - print(f"已写入: {PAC_FILE}（{len(cidrs)} 条，{size:.0f} KB）") - - -if __name__ == "__main__": - main() diff --git a/lite/test.bat b/lite/test.bat deleted file mode 100644 index 02f5b53..0000000 --- a/lite/test.bat +++ /dev/null @@ -1,4 +0,0 @@ -@echo off -cd /d "%~dp0" -call ..\.venv\Scripts\python.exe main.py -pause diff --git a/lite/test/PyQt5/QtCore.py b/lite/test/PyQt5/QtCore.py deleted file mode 100644 index a72f50f..0000000 --- a/lite/test/PyQt5/QtCore.py +++ /dev/null @@ -1,10 +0,0 @@ -class Qt: pass -class QTimer: pass - -class _Signal: - def connect(self, *a, **k): pass - def emit(self, *a, **k): pass - -def pyqtSignal(*a, **k): - return _Signal() -class QRectF: pass diff --git a/lite/test/PyQt5/QtGui.py b/lite/test/PyQt5/QtGui.py deleted file mode 100644 index bfc830d..0000000 --- a/lite/test/PyQt5/QtGui.py +++ /dev/null @@ -1,6 +0,0 @@ -class QFont: pass -class QIcon: pass -class QPixmap: pass -class QImage: pass -class QColor: pass -class QPainter: pass diff --git a/lite/test/PyQt5/QtWidgets.py b/lite/test/PyQt5/QtWidgets.py deleted file mode 100644 index 3c978ee..0000000 --- a/lite/test/PyQt5/QtWidgets.py +++ /dev/null @@ -1,28 +0,0 @@ -class QApplication: pass -class QWidget: pass -class QLabel: - def __init__(self, text=""): - self.text = text - def setStyleSheet(self, *a): pass - def setToolTip(self, *a): pass - def setFixedSize(self, *a): pass - def setFixedHeight(self, *a): pass - def setAlignment(self, *a): pass - def setPixmap(self, *a): pass - def setText(self, *a): pass - def setCursor(self, *a): pass - def unsetCursor(self): pass - def hide(self): pass - def show(self): pass -class QPushButton: pass -class QVBoxLayout: pass -class QHBoxLayout: pass -class QFrame: pass -class QMessageBox: pass -class QTextEdit: pass -class QCheckBox: pass -class QMenu: pass -class QDialog: Accepted = 1 -class QLineEdit: Password = 2 -class QListWidget: pass -class QListWidgetItem: pass diff --git a/lite/test/PyQt5/__init__.py b/lite/test/PyQt5/__init__.py deleted file mode 100644 index 0436b2f..0000000 --- a/lite/test/PyQt5/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# PyQt5 桩模块：仅供 WSL 下 import main 做路由逻辑测试， -# 决策函数（_decide_direct 等）不依赖任何 Qt 对象。 diff --git a/lite/test/check_node_menu.py b/lite/test/check_node_menu.py deleted file mode 100644 index 92fd61d..0000000 --- a/lite/test/check_node_menu.py +++ /dev/null @@ -1,127 +0,0 @@ -# -*- coding: utf-8 -*- -"""离屏验证节点菜单：IPv6-only 节点的标注与置灰（真实 PyQt5，不出窗口）。 - -需要 Windows venv 的真实 PyQt5： - ..\\.venv\\Scripts\\python.exe test\\check_node_menu.py -读取 test/nodes.json 作为节点数据，只构建菜单不进事件循环。 -""" -import os, sys -os.environ["QT_QPA_PLATFORM"] = "offscreen" -from pathlib import Path - -LITE = Path(__file__).resolve().parent.parent -# 脚本所在目录（test/）会被解释器自动放到 sys.path 首位，那里有 PyQt5 桩， -# 会顶掉真实 PyQt5 —— 必须移除，改用本程序自己的 venv -sys.path = [str(LITE)] + [p for p in sys.path if Path(p).resolve() != Path(__file__).resolve().parent] - -sys.stdout.reconfigure(encoding="utf-8") - -import main as pp -from PyQt5.QtWidgets import QApplication - -# 构造窗口时不触发用户数据初始化或远程同步；本测试只验证静态菜单和本地节点状态。 -pp._user_init_local = lambda: None -pp._user_init_remote = lambda *args, **kwargs: None -pp._nodes_sync_remote = lambda *args, **kwargs: pp._NODES_SYNC_DONE.set() -pp._NODES_SYNC_DONE.set() - -app = QApplication(sys.argv) -w = pp.MainWindow() -w.timer.stop() # 停掉状态轮询，纯静态检查 -failures = [] - -pp._USER_DIR = LITE / "test" # 真实 nodes.json 所在目录 -nodes = pp._reload_nodes() -print(f"本机有全局 IPv6: {pp._local_has_ipv6()}") -print(f"节点数: {len(nodes)}") - -menu = w._build_node_menu(nodes) -print("\n菜单项（enabled / 文案 / tooltip）:") -for act in menu.actions(): - print(f" {str(act.isEnabled()):5} | {act.text()}") - if act.toolTip(): - print(f" └ {act.toolTip()}") - -# 点击被禁用的项不应切换节点 -before = pp._selected_index() -for act in menu.actions(): - if not act.isEnabled(): - act.trigger() -print(f"\n点击禁用项后节点下标: {before} -> {pp._selected_index()} (应不变)") -if pp._selected_index() != before: - failures.append("点击不可用节点后发生了切换") - -# ── 上次选中的节点已不可用：启动时自动回退到第一个可达节点 ────────── -print("\n启动校验（_verify_selected_node）:") -bad = [i for i, n in enumerate(nodes) if not pp._node_reachable(n)] -if not bad: - print(" 本机所有节点均可达，无法验证回退（本机有 IPv6？）") -else: - import json - _bak = (pp._USER_DIR / "settings.json").read_bytes() if (pp._USER_DIR / "settings.json").is_file() else None - bad_idx = bad[0] - # 模拟上次持久化选中了不可用节点：内存下标与 current_node 必须一致， - # 否则 _verify_selected_node 会先按真实保存的域名重新匹配，测试并非回退分支。 - pp._settings_save({**pp._settings_load(), "current_node": nodes[bad_idx].get("domain") or ""}) - pp._SELECTED_NODE = bad_idx - switched = pp._verify_selected_node() - expect = pp._first_reachable_index(nodes) - ok = switched and pp._selected_index() == expect - print(f" 上次选中 {nodes[bad_idx].get('domain')}(下标 {bad_idx}) 不可用 -> " - f"回退到下标 {pp._selected_index()} ({nodes[pp._selected_index()].get('domain')}) " - f"{'PASS' if ok else 'FAIL'}") - if not ok: - failures.append("不可用节点未回退到首个可达节点") - saved = json.loads((pp._USER_DIR / "settings.json").read_text("utf-8")).get("current_node") if (pp._USER_DIR / "settings.json").is_file() else "" - ok2 = saved == str(nodes[expect].get("domain") or "") - print(f" 回退结果已写入设置 current_node={saved!r} {'PASS' if ok2 else 'FAIL'}") - if not ok2: - failures.append("节点回退未写回 current_node") - # 再校验一次：已经是可达节点，不应再切换 - ok3 = pp._verify_selected_node() is False - print(f" 再次校验不重复切换 {'PASS' if ok3 else 'FAIL'}") - if not ok3: - failures.append("已回退节点被重复切换") - if _bak is None: - (pp._USER_DIR / "settings.json").unlink(missing_ok=True) - else: - (pp._USER_DIR / "settings.json").write_bytes(_bak) - -# ── 服务端调整节点顺序：按域名重匹配而不是死守下标 ────────────────── -print("\n服务端调序（_rematch_saved_domain）:") -if len(nodes) < 2: - print(" 节点不足 2 个，跳过") -else: - import json - dom_old = nodes[0].get("domain") - ordered = [nodes[1], nodes[0]] + nodes[2:] # 模拟服务端交换了前两个节点 - pp._NODES = ordered - # 设置里记录的是原下标 0 的域名，重匹配后下标应跟随域名（新列表里是 1） - pp._settings_save({**pp._settings_load(), "current_node": dom_old}) - switched = pp._rematch_saved_domain() - ok = switched and ordered[pp._selected_index()].get("domain") == dom_old - print(f" current_node={dom_old} 调序后跟随域名 -> 下标 {pp._selected_index()} " - f"{'PASS' if ok else 'FAIL'}") - if not ok: - failures.append("节点调序后没有按域名重新匹配") - # 已是正确域名时再跑一次：不应重复切换 - ok2 = pp._rematch_saved_domain() is False - print(f" 再次重匹配不重复切换 {'PASS' if ok2 else 'FAIL'}") - if not ok2: - failures.append("节点调序重匹配重复切换") - # 域名已不在列表（节点被移除）：不切换，保持现状 - pp._settings_save({**pp._settings_load(), "current_node": "gone.example.com"}) - ok3 = pp._rematch_saved_domain() is False - print(f" 节点被移除时不动 {'PASS' if ok3 else 'FAIL'}") - if not ok3: - failures.append("已移除节点仍发生切换") - pp._NODES = None # 还原缓存，后续用例重读文件 - # 还原 settings.json，避免测试污染 - if _bak is None: - (pp._USER_DIR / "settings.json").unlink(missing_ok=True) - else: - (pp._USER_DIR / "settings.json").write_bytes(_bak) - -if failures: - raise SystemExit("FAIL: " + "; ".join(failures)) -print("OK") diff --git a/lite/test/nodes.json b/lite/test/nodes.json deleted file mode 100644 index e01510f..0000000 --- a/lite/test/nodes.json +++ /dev/null @@ -1,21 +0,0 @@ -[ - { - "domain": "example.com", - "port": 443, - "country": "TEST", - "remark": "普通测试节点" - }, - { - "domain": "example.org", - "port": 443, - "country": "TEST", - "remark": "备用测试节点" - }, - { - "domain": "ipv6.test.invalid", - "port": 443, - "country": "TEST", - "ipv6_only": true, - "remark": "IPv6 测试节点" - } -] diff --git a/lite/test/probe_nodes.py b/lite/test/probe_nodes.py deleted file mode 100644 index f8115f0..0000000 --- a/lite/test/probe_nodes.py +++ /dev/null @@ -1,151 +0,0 @@ -# -*- coding: utf-8 -*- -"""连通性探测：用 test/nodes.json 的真实节点，走 _tls_start 转发逻辑实测。 - -仅需标准库；不启动 GUI（main 的 GUI 在 __main__ 保护下）。 -用法: python3 test/probe_nodes.py [JP US ...] 不带参数测全部 -""" -import os, sys, time, json, ssl, socket, urllib.request, urllib.error -from pathlib import Path - -HERE = Path(__file__).resolve().parent -LITE = HERE.parent -sys.path.insert(0, str(HERE)) # PyQt5 桩 -sys.path.insert(0, str(LITE)) - -for k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"): - os.environ.pop(k, None) -# 不能设 NO_PROXY=*：urllib 会把 127.0.0.1 也当直连，显式 ProxyHandler 被绕过 -os.environ["NO_PROXY"] = "" - -import main as pp - -NODES = json.loads((HERE / "nodes.json").read_text("utf-8")) -want = [a.upper() for a in sys.argv[1:]] -if want: - NODES = [n for n in NODES if str(n.get("country", "")).upper() in want] - -TIMEOUT = 12 - - -def free_port(): - s = socket.socket(); s.bind(("127.0.0.1", 0)) - p = s.getsockname()[1]; s.close(); return p - - -def http_method(node): - return (node.get("methods") or {}).get("http") or {} - - -def as_flat_node(node): - """把 nodes.json 的 methods.http 结构摊平成代码 _node_val() 期望的形状。 - 仅用于探测：暴露「代码读不到真实节点字段」这一问题。""" - m = http_method(node) - return dict(node, port=int(m.get("port")), username=m.get("username", ""), - password=m.get("password", "")) - - -def probe_connect(host, port, sni=None, connect_timeout=8): - """直连探测：DNS -> TCP -> TLS -> 发 CONNECT 127.0.0.1:9 看上游代理响应。""" - res = {"dns": None, "tcp": None, "tls": None, "proxy": None, "err": None} - t0 = time.time() - try: - infos = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM) - res["dns"] = sorted({i[4][0] for i in infos}) - except Exception as e: - res["err"] = f"DNS: {e}" - return res - addr = res["dns"][0] - af = socket.AF_INET6 if ":" in addr else socket.AF_INET - try: - s = socket.socket(af, socket.SOCK_STREAM) - s.settimeout(connect_timeout) - s.connect((addr, port)) - res["tcp"] = round(time.time() - t0, 2) - except Exception as e: - res["err"] = f"TCP: {e}" - return res - try: - ctx = ssl.create_default_context() - s = ctx.wrap_socket(s, server_hostname=sni or host) - res["tls"] = s.version() - except Exception as e: - res["err"] = f"TLS: {e}" - try: s.close() - except Exception: pass - return res - try: - s.sendall(b"CONNECT 127.0.0.1:9 HTTP/1.1\r\nHost: 127.0.0.1:9\r\n\r\n") - res["proxy"] = s.recv(200).split(b"\r\n", 1)[0].decode("latin-1", "replace") - except Exception as e: - res["err"] = f"CONNECT: {e}" - try: s.close() - except Exception: pass - return res - - -def probe_via_forwarder(node): - """真实转发器链路：_tls_start -> 经 127.0.0.1 代理请求，取出口 IP 与归属地。""" - m = http_method(node) - if not m: - return "（该节点无 http 方法，跳过）" - old_node, old_port = pp._current_node, pp._TLS_LOCAL_PORT - port = free_port() - try: - # _proxy_auth_basic() 内部走 _current_node()，固定成本次待测节点 - pp._current_node = lambda: node - pp._TLS_LOCAL_PORT = port - pp._tls_stop() - pp._tls_start(node["domain"], int(m["port"]), use_tls=True, - bypass_lan=False, smart=False, quiet=True) - ph = urllib.request.ProxyHandler({"http": f"127.0.0.1:{port}", - "https": f"127.0.0.1:{port}"}) - op = urllib.request.build_opener(ph) - t0 = time.time() - # 明文 HTTP 代理（走 GET 绝对 URI）与 CONNECT 隧道各测一条 - with op.open("http://ip-api.com/json/?fields=query,country,regionName,isp", - timeout=TIMEOUT) as r: - info = json.loads(r.read(512).decode("utf-8", "replace")) - with op.open("https://ifconfig.me/ip", timeout=TIMEOUT) as r: - https_ip = r.read(64).decode().strip() - dt = round(time.time() - t0, 2) - return (f"OK {dt}s 出口IP={info.get('query')} " - f"[{info.get('country')} {info.get('regionName')} / {info.get('isp')}] " - f"HTTPS隧道出口={https_ip}") - except urllib.error.HTTPError as e: - return f"HTTP {e.code}: {e.read(120).decode('utf-8', 'replace').strip()[:100]}" - except Exception as e: - return f"失败: {type(e).__name__}: {e}" - finally: - pp._tls_stop() - pp._current_node, pp._TLS_LOCAL_PORT = old_node, old_port - - -print(f"节点数: {len(NODES)}") -n0 = NODES[0] -print(f"代码 _endpoint_domain()={pp._endpoint_domain()!r} " - f"_endpoint_port()={pp._endpoint_port()!r} _endpoint_country()={pp._endpoint_country()!r}") -print(f"代码 _proxy_auth_basic()={pp._proxy_auth_basic()!r}") -print(f"nodes.json[0] 顶层键 = {list(n0.keys())}") -print(f"nodes.json[0] domain={n0.get('domain')!r} methods.http.port=" - f"{http_method(n0).get('port')!r}") -print() - -for n in NODES: - c = n.get("country", "?") - m = http_method(n) - print("=" * 78) - print(f"[{c}] {n.get('domain')}:{m.get('port')} provider={n.get('provider')} " - f"traffic={n.get('traffic')} level={n.get('available_level')}") - if not m: - print(" 该节点无 http 方法，跳过") - continue - r = probe_connect(n["domain"], int(m["port"])) - print(f" DNS : {r['dns']}") - print(f" TCP : {r['tcp']}s" if r["tcp"] is not None else " TCP : -") - print(f" TLS : {r['tls']}" if r.get("tls") else " TLS : -") - print(f" CONNECT: {r['proxy']}" if r.get("proxy") else " CONNECT: -") - if r.get("err"): - print(f" 错误 : {r['err'].splitlines()[0][:200]}") - print(f" 经转发器: {probe_via_forwarder(n)}") - print(f" 经转发器(摊平节点): {probe_via_forwarder(as_flat_node(n))}") - print() diff --git a/lite/test/test_core_modules.py b/lite/test/test_core_modules.py deleted file mode 100644 index 202f824..0000000 --- a/lite/test/test_core_modules.py +++ /dev/null @@ -1,114 +0,0 @@ -# -*- coding: utf-8 -*- -"""拆分后 rrpp_core 的独立回归测试（不依赖 Qt 或真实代理连接）。""" - -import sys -import tempfile -import unittest -from pathlib import Path - - -LITE_DIR = Path(__file__).resolve().parent.parent -sys.path.insert(0, str(LITE_DIR)) - -from rrpp_core.forwarder_primitives import ( - force_http_close, - inject_proxy_auth, - parse_connect_authority, - parse_http_absolute, -) -from rrpp_core.node_api import NodeAuthError, b64url -from rrpp_core.node_secrets import NodeSecretsStore -from rrpp_core.network_primitives import is_private_host, is_private_ipv4_int, ttl_cache -from rrpp_core.pac_data import atomic_write, read_version, validate_document -from rrpp_core.routing_primitives import ( - build_suffix_trie, - merge_ip_ranges, - parse_domain_rules, - suffix_trie_matches, -) -from rrpp_core.settings_store import JsonSettingsStore -from rrpp_core.traffic import TrafficStats, format_bytes - - -class CoreModuleTests(unittest.TestCase): - def test_domain_rules_and_trie(self): - suffixes, exact, regexes, invalid = parse_domain_rules( - ["example.com", "full:only.example", "regexp:^edge-[0-9]+\\.example$", "regexp:["] - ) - self.assertEqual(invalid, 1) - self.assertIn("example.com", suffixes) - self.assertIn("only.example", exact) - self.assertTrue(regexes[0].search("edge-42.example")) - trie = build_suffix_trie(suffixes) - self.assertTrue(suffix_trie_matches("cdn.example.com", trie)) - self.assertFalse(suffix_trie_matches("example.net", trie)) - self.assertEqual( - merge_ip_ranges(["10.0.0.0/25", "10.0.0.128/25"], 4), - [(167772160, 167772415)], - ) - - def test_forwarder_primitives(self): - self.assertEqual( - parse_connect_authority("CONNECT [::1]:443 HTTP/1.1"), ("::1", 443) - ) - self.assertEqual( - parse_http_absolute("GET https://example.com/path HTTP/1.1"), - ("example.com", 443), - ) - request = b"GET / HTTP/1.1\r\nHost: example.com\r\n\r\n" - injected = inject_proxy_auth(request, "token") - self.assertIn(b"Authorization: Basic token\r\n", injected) - self.assertEqual(inject_proxy_auth(injected, "different"), injected) - self.assertIn(b"Connection: close\r\n", force_http_close(request)) - - def test_settings_pac_and_node_secrets_storage(self): - with tempfile.TemporaryDirectory() as temporary: - directory = Path(temporary) - settings = JsonSettingsStore(lambda: directory / "settings.json") - self.assertEqual(settings.load(), {}) - settings.save({"local_port": 1080}) - self.assertEqual(settings.load(), {"local_port": 1080}) - - payload = {"version": "1", "domains": ["example.com"]} - atomic_write(directory, "pac_domains", payload) - self.assertEqual(read_version(directory / "pac_domains.json"), ("1", "")) - self.assertTrue(validate_document("pac_domains", payload)) - - secrets = NodeSecretsStore(lambda: directory) - self.assertEqual(secrets.decrypt_bytes(secrets.encrypt_text("round-trip")), "round-trip") - secrets.save_auth("test-user", "test-password") - self.assertEqual( - secrets.load_auth(), {"username": "test-user", "password": "test-password"} - ) - - def test_traffic_and_node_protocol_helpers(self): - times = iter((10.0, 12.0)) - stats = TrafficStats(clock=lambda: next(times)) - stats.reset() - stats.add(up=20, down=10) - self.assertEqual(stats.rate(), (10, 5, 20, 10)) - self.assertEqual(format_bytes(1536), "1.5 KB") - self.assertEqual(b64url("a/b="), "YS9iPQ") - error = NodeAuthError("denied", 403) - self.assertEqual((str(error), error.code), ("denied", 403)) - - def test_network_primitives_and_ttl_cache(self): - self.assertTrue(is_private_ipv4_int(0x0A010203)) - self.assertTrue(is_private_host("::1")) - self.assertFalse(is_private_host("8.8.8.8")) - calls = [] - - @ttl_cache(300) - def probe(value): - calls.append(value) - return value - - self.assertEqual((probe("x"), probe("x")), ("x", "x")) - self.assertEqual(calls, ["x"]) - probe.cache_clear() - probe("x") - self.assertEqual(calls, ["x", "x"]) - - -if __name__ == "__main__": - unittest.main(verbosity=2) diff --git a/lite/test/test_pac_routing.py b/lite/test/test_pac_routing.py deleted file mode 100644 index be4eb07..0000000 --- a/lite/test/test_pac_routing.py +++ /dev/null @@ -1,739 +0,0 @@ -# -*- coding: utf-8 -*- -"""PAC（智能分流）模式路由决策测试。 - -核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？ - -判定链（main._decide_direct）： - 0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效） - 1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定） - 2. 内网 IP 且勾选 bypass_lan -> 直连 - 3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道 - -直接 import main 调用真实决策函数，数据用真实的 pac_ip_cn.json / -pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。 -""" - -import sys -import threading -from pathlib import Path - -TEST_DIR = Path(__file__).resolve().parent -LITE_DIR = TEST_DIR.parent - -sys.path.insert(0, str(TEST_DIR)) # 假 PyQt5 桩 -sys.path.insert(0, str(LITE_DIR)) - -import json -import socket -import main as pp -from main import _decide_direct, _is_china_ip, _is_force_proxy, _is_private_ip, _is_whitelisted - -# PAC 数据已不再随项目保留：从 CDN 拉取到临时用户目录（真实数据端到端） -import tempfile -pp._USER_DIR = Path(tempfile.mkdtemp()) / "RRPP" -pp._user_init_remote(base_url="https://cdn.rrrosmontis.icu/RRPP/PAC/") - -pp._load_cn_ranges() -pp._load_force_proxy_suffixes() -if not pp._CN_RANGES: - print("警告: 未能获取 pac_ip_cn.json（CDN 不可达？），中国 IP 判定将失效") -print(f"中国IP段: {len(pp._CN_RANGES or [])} 段, 强制域名表: {len(pp._FORCE_PROXY_SUFFIXES or [])} 条\n") - - -def resolved_ips(host): - try: - return sorted({a[4][0] for a in socket.getaddrinfo(host, 443, socket.AF_INET)}) - except Exception: - return [] - - -def mark(ip): - if _is_private_ip(ip): - return "内网" - if _is_china_ip(ip): - return "中国IP" - return "国外IP" - - -PATH = {True: "直连", False: "隧道"} -MODES = [ - ("PAC+bypass_lan", dict(bypass_lan=True, smart=True)), # 默认勾选状态 - ("PAC无bypass_lan", dict(bypass_lan=False, smart=True)), - ("固定代理模式", dict(bypass_lan=False, smart=False)), # 全部走隧道 -] - -# (名称, host, 期望 PAC+bypass_lan 决策: True=直连 / False=隧道) -cases = [ - ("境外域名 google.com", "google.com", False), - ("境外域名 github.com", "github.com", False), - ("境外域名 youtube.com", "youtube.com", False), - ("域名表强制 claude.ai", "claude.ai", False), - ("域名表强制 *.google 后缀", "antigravity.google", False), - ("国内域名 www.baidu.com", "www.baidu.com", True), - ("国内域名 www.qq.com", "www.qq.com", True), - ("内网 localhost", "localhost", True), - ("内网 IP 192.168.1.1", "192.168.1.1", True), - ("境外 IP 8.8.8.8", "8.8.8.8", False), - ("国内 IP 114.114.114.114", "114.114.114.114", True), -] - -failed = 0 -for name, host, expect in cases: - ips = resolved_ips(host) - force = _is_force_proxy(host) - marks = f"[{','.join(mark(ip) for ip in ips)}]" if ips else "[解析失败]" - row = f"{name:<22} {host:<18} {str(ips) if ips else '无IP':<45} {marks}" - results = [] - for mode_name, kw in MODES: - got = _decide_direct(host, **kw) - results.append(f"{mode_name}={PATH[got]}") - pac = _decide_direct(host, **MODES[0][1]) - ok = pac == expect - failed += 0 if ok else 1 - print(f"{row}\n {'域名表命中' if force else '域名表未命中':<10} | {' '.join(results)} | 期望 {PATH[expect]} -> {'PASS' if ok else 'FAIL'}") - -print(f"\n{'='*70}") -print(f"结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}") -print("=" * 70) -print("PAC 模式(智能分流)下: 域名解析结果全部为国外 IP -> 走隧道代理 ✓" - if not failed else "存在 FAIL，请检查上方明细") - -# ── 附加测试：不在域名表的被墙境外站 ───────────────────── -# 这类站点不能靠域名表兜底，必须仅凭「解析结果为国外 IP」强制走隧道， -# 否则会尝试直连 -> 被墙。 -print("\n" + "=" * 70) -print("附加: 不在 pac_domains.json 的被墙境外站（只能靠 IP 判定走隧道）") -print("=" * 70) - -extra_candidates = [ - "dailymotion.com", "vimeo.com", "notion.so", "linear.app", - "quora.com", "feedly.com", "dropbox.com", "tumblr.com", - "flickr.com", "pastebin.com", "fandom.com", "archive.org", - "godaddy.com", "imgur.com", "kickstarter.com", "shopify.com", -] - -extra_failed = 0 -skipped = 0 -for host in extra_candidates: - if _is_force_proxy(host): - print(f"{host:<20} 已在域名表中，跳过") - skipped += 1 - continue - ips = resolved_ips(host) - if not ips: - print(f"{host:<20} 解析失败，跳过") - skipped += 1 - continue - foreign = all(not _is_private_ip(ip) and not _is_china_ip(ip) for ip in ips) - if not foreign: - print(f"{host:<20} 解析到非国外 IP {ips}，跳过（不是有效的境外用例）") - skipped += 1 - continue - got = _decide_direct(host, bypass_lan=True, smart=True) # PAC 模式 - if got is not False: - # 决策函数内部会重新解析域名，CDN 轮询 DNS 可能与其上方展示的解析结果 - # 不一致：清缓存重试一次，仍不一致则按 DNS 不稳定跳过（非代码问题） - _is_china_ip.cache_clear() - _is_private_ip.cache_clear() - ips = resolved_ips(host) - got = _decide_direct(host, bypass_lan=True, smart=True) - if got is not False: - print(f"{host:<20} 解析/判定随 DNS 轮询不一致 {ips}，跳过（非代码问题）") - skipped += 1 - continue - print(f"{host:<20} 重试后走隧道（DNS 瞬时差异） PASS") - continue - extra_failed += 0 - print(f"{host:<20} {str(ips):<50} [{'，'.join(mark(ip) for ip in ips)}] -> " - f"{PATH[got]} (期望隧道) PASS") - -print(f"\n附加测试: {len(extra_candidates)} 个候选, 有效用例 {len(extra_candidates) - skipped} 个, " - f"{'全部 PASS' if not extra_failed else f'{extra_failed} 个 FAIL'}") -if extra_failed: - failed += extra_failed -print("=" * 70) - -# ── 代理白名单测试（用户可配置：命中即直连，最优先，含 IPv4/IPv6/CIDR） ───── -print("\n" + "=" * 70) -print("代理白名单: 命中规则强制直连（优先于域名表/固定代理，支持 IPv4/IPv6/CIDR）") -print("=" * 70) - -wl_cases = [ - # (规则, 测试host, 期望命中) - ("example.com", "example.com", True), - ("example.com", "www.example.com", True), - ("example.com", "sub.www.example.com", True), - ("example.com", "example.org", False), - ("claude.ai", "claude.ai", True), # 域名表强制隧道，白名单优先 - ("localhost", "localhost", True), - ("8.8.8.8", "8.8.8.8", True), # IPv4 字面量 - ("8.8.8.8", "8.8.4.4", False), - ("10.0.0.0/8", "10.1.2.3", True), # IPv4 CIDR - ("192.0.2.0/24", "192.0.2.55", True), # IPv4 CIDR - ("2001:db8::/32", "2001:db8::1", True), # IPv6 网段 - ("fe80::1", "fe80::1", True), # IPv6 单地址 -] - -# 注入白名单规则（绕过 settings.json 直接改内存缓存；版本 +1 使匹配缓存失效）， -# 另加一条"禁用"的规则验证不参与匹配 -pp._WHITELIST = [{"value": "203.0.113.0/24", "enabled": False}] + \ - [{"value": r, "enabled": True} for r, _, _ in wl_cases] -wl_ver = pp._WHITELIST_VERSION + 1 -pp._WHITELIST_VERSION = wl_ver - -wl_failed = 0 -for rule, host, expect in wl_cases: - got = _is_whitelisted(host, wl_ver) - ok = got == expect - wl_failed += 0 if ok else 1 - print(f"规则 {rule:<16} host {host:<22} 命中={got} 期望={expect} -> {'PASS' if ok else 'FAIL'}") -got = _is_whitelisted("203.0.113.9", wl_ver) # 禁用的规则不应命中 -print(f"禁用规则 203.0.113.0/24 host 203.0.113.9 -> 命中={got} 期望=False -> {'PASS' if not got else 'FAIL'}") -wl_failed += 0 if not got else 1 - -# 白名单在三种模式下都强制直连（固定代理模式原本全部走隧道） -print("\n白名单在三种模式下的决策（命中白名单应为直连）:") -for mode_name, kw in MODES: - got = _decide_direct("www.example.com", **kw) - ok = got is True - wl_failed += 0 if ok else 1 - print(f" {mode_name:<16} www.example.com -> {PATH[got]} (期望 直连) {'PASS' if ok else 'FAIL'}") -# 白名单优先于域名表：claude.ai 在 pac_domains.json 中强制隧道，白名单命中应直连 -got = _decide_direct("claude.ai", bypass_lan=True, smart=True) -ok = got is True -wl_failed += 0 if ok else 1 -print(f" {'域名表冲突':<14} claude.ai -> {PATH[got]} (期望 直连，白名单优先) {'PASS' if ok else 'FAIL'}") -# 未命中白名单的境外域名仍走隧道（白名单不影响其他路由） -got = _decide_direct("google.com", bypass_lan=True, smart=True) -ok = got is False -wl_failed += 0 if ok else 1 -print(f" {'未命中白名单':<14} google.com -> {PATH[got]} (期望 隧道) {'PASS' if ok else 'FAIL'}") - -# 恢复空白名单，避免影响后续用例 -pp._WHITELIST = None -pp._WHITELIST_VERSION += 1 -if wl_failed: - failed += wl_failed -print("=" * 70) - -# ── IPv6-only 节点标注 ───────────────────────────────────── -# IPv6 专用节点只提供 AAAA 记录，本机没有 IPv6 出口时连不上： -# 客户端据此在节点菜单标注并置灰。判定三级：ipv6_only 字段 → -# remark 里的 "(IPv6)"（服务端 rrpp_client 现网唯一的线索）→ 域名解析兜底。 -print("\n" + "=" * 70) -print("IPv6 专用节点：识别（三级判定）与可达性") -print("=" * 70) - -ipv6_failed = 0 -has_v6 = pp._local_has_ipv6() -print(f"本机是否有全局 IPv6 地址: {has_v6}") - -# 用真实 nodes.json（test 目录）；没有则跳过 -_nodes_file = TEST_DIR / "nodes.json" -real_nodes = json.loads(_nodes_file.read_text("utf-8")) if _nodes_file.is_file() else [] -if real_nodes: - pp._IPV6_OK.clear() # 清缓存，做真实探测 - for n in real_nodes: - c = n.get("country") or "?" - io = pp._node_ipv6_only(n) - reach = pp._node_reachable(n) - mark = "[IPv6]" if reach else "[不可用：本机无 IPv6]" - print(f" [{c}] {n.get('domain'):<24} ipv6_only={str(io):<6} " - f"可达={str(reach):<6} 菜单文案: {n.get('domain')} ({c})" - + (f"　{mark}" if io else "")) - # ipv6_only 节点在本机无 IPv6 时必须判定为不可达；有 IPv6 时仍取 - # 实际 TCP 探测结果（本机有 IPv6 不代表每个远端节点都必然在线）。 - if io: - expect = "实际探测" if has_v6 else False - ok = isinstance(reach, bool) if has_v6 else reach is False - ipv6_failed += 0 if ok else 1 - print(f" IPv6 专用节点可达性: {reach} (期望 {expect}) " - f"{'PASS' if ok else 'FAIL'}") - else: - ok = reach is True - ipv6_failed += 0 if ok else 1 - print(f" 普通节点恒可达: {reach} {'PASS' if ok else 'FAIL'}") - # 缓存生效：再次判定不发新探测（结果一致） - cached = list(pp._IPV6_OK.items()) - if cached: - print(f" 可达性缓存: {cached}") - -# 三级判定：字段 → remark 里的 "(IPv6)" 标记 → 域名解析兜底。 -# 服务端 subscription?method=rrpp_client 只吐旧扁平格式，ipv6_only 字段被丢掉， -# 只在 remark 里留 "(IPv6)"（后端 countries_to_rrpp_client），故第 2 级是现网主路径。 -synthetic = [ - ({"domain": "example.com"}, False, "无标记（example.com 有 A 记录）"), - ({"domain": "example.com", "ipv6_only": False}, False, "ipv6_only=false"), - ({"domain": "example.com", "ipv6_only": True}, True, "ipv6_only=true"), - ({"domain": "example.com", "remark": "GlobalVM 350G (IPv6)"}, True, "remark 含 (IPv6)"), - ({"domain": "example.com", "remark": "Azure 15G"}, False, "remark 无 IPv6 标记"), - ({"domain": "example.com", "remark": "IPv6 专用机"}, True, "remark 含 IPv6（无括号）"), - ({"domain": "example.com", "remark": "some_ipv6_thing"}, True, "remark 小写 ipv6"), -] -for node, expect, desc in synthetic: - got = pp._node_ipv6_only(node) - ok = got == expect - ipv6_failed += 0 if ok else 1 - print(f" {desc:<24} _node_ipv6_only={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") -pp._IPV6_OK.clear() # 合成用例可能污染缓存 - -if ipv6_failed: - failed += ipv6_failed -print("=" * 70) - -# ── CONNECT authority 解析：IPv6 字面量带方括号 ────────────────── -print("\n" + "=" * 70) -print("CONNECT authority 解析（IPv6 字面量剥方括号）") -print("=" * 70) - -conn_cases = [ - ("CONNECT [::1]:443 HTTP/1.1", ("::1", 443)), - ("CONNECT example.com:443 HTTP/1.1", ("example.com", 443)), - ("CONNECT iam.aeolianchi.me:443 HTTP/1.1", ("iam.aeolianchi.me", 443)), - ("CONNECT [::1] HTTP/1.1", None), # 无端口 - ("CONNECT example.com HTTP/1.1", None), # 无端口 - ("CONNECT [fe80::1%25eth0]:8443 HTTP/1.1", ("fe80::1%25eth0", 8443)), - ("CONNECT 127.0.0.1:1080 HTTP/1.1", ("127.0.0.1", 1080)), -] -conn_failed = 0 -for line, expect in conn_cases: - got = pp._parse_connect_authority(line) - ok = got == expect - conn_failed += 0 if ok else 1 - print(f" {line:<45} -> {got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") -if conn_failed: - failed += conn_failed -print("=" * 70) - -# ── 上游代理认证注入：按完整头名精确匹配 ──────────────────────── -print("\n" + "=" * 70) -print("认证注入：仅精确的 authorization 头名阻止注入") -print("=" * 70) - -def _auth_lines(data: bytes): - head = data.split(b"\r\n\r\n", 1)[0] - return [l for l in head.split(b"\r\n")[1:] if l.lower().startswith(b"authorization:")] - -auth_cases = [ - # (描述, 请求, 期望)：数字=注入后 authorization 行数；"unchanged"=原样跳过 - ("无任何认证头", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nHost: iam.aeolianchi.me\r\n\r\n", 1), - ("Proxy-Authorization", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nProxy-Authorization: Basic xyz\r\n\r\n", 1), - ("已有 Authorization", b"GET http://example.com/ HTTP/1.1\r\nAuthorization: Bearer tok\r\n\r\n", "unchanged"), - ("小写 authorization", b"GET http://example.com/ HTTP/1.1\r\nauthorization: basic abc\r\n\r\n", "unchanged"), - ("X-Authorization 前缀", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nX-Authorization-Go: 1\r\n\r\n", 1), - ("无头空块", b"GET http://example.com/ HTTP/1.1\r\n\r\n", 1), -] -auth_failed = 0 -for desc, req, expect in auth_cases: - out = pp._inject_proxy_auth(req) - if expect == "unchanged": - ok = out == req - print(f" {desc:<22} 原样跳过 {'PASS' if ok else 'FAIL'}") - else: - lines = _auth_lines(out) - ok = len(lines) == expect and out.split(b"\r\n", 2)[1] == lines[0] - # 注入位置必须紧跟请求行（上游取首个 Authorization 时读到我们的认证） - print(f" {desc:<22} authorization 行数={len(lines)} (期望 {expect}) " - f"{'PASS' if ok else 'FAIL'}") - auth_failed += 0 if ok else 1 -if auth_failed: - failed += auth_failed -print("=" * 70) - -# ── 本程序 PAC 地址判定（启动清理只认自己的残留） ──────────────── -print("\n" + "=" * 70) -print("PAC 地址归属判定（_is_our_pac_url）") -print("=" * 70) - -pac_cases = [ - ("http://127.0.0.1:39453/proxy.pac", True), # 本程序 PAC 服务 - ("http://127.0.0.1:7897/pac?auth=xyz", False), # 其他代理软件（如 Clash） - ("http://127.0.0.1:1080/other.pac", False), # 路径不符 - ("http://192.168.1.1:8080/proxy.pac", False), # 非本机 - ("", False), -] -pac_failed = 0 -for url, expect in pac_cases: - got = pp._is_our_pac_url(url) - ok = got == expect - pac_failed += 0 if ok else 1 - print(f" {url:<42} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") -if pac_failed: - failed += pac_failed -print("=" * 70) - -# ── 强制代理规则格式：后缀 / full / regexp + 关键域名兜底 ─────────── -print("\n" + "=" * 70) -print("强制代理规则格式（后缀 / full / regexp / 关键域名兜底）") -print("=" * 70) - -rule_failed = 0 -rule_dir = Path(tempfile.mkdtemp()) / "RRPP" -rule_dir.mkdir(parents=True) -(rule_dir / "pac_domains.json").write_text(json.dumps({"domains": [ - "suffix.example", "full:only.example", "regexp:^edge-[0-9]+\\.example$", -]}), "utf-8") - -# 此段隔离真实 CDN 表，确保三种语义分别受测。 -old_user_dir = pp._USER_DIR -old_suffixes = pp._FORCE_PROXY_SUFFIXES -old_exact = pp._FORCE_PROXY_EXACT -old_regexes = pp._FORCE_PROXY_REGEXES -old_loaded = pp._FORCE_PROXY_LOADED -pp._USER_DIR = rule_dir -pp._FORCE_PROXY_SUFFIXES = None -pp._FORCE_PROXY_EXACT = None -pp._FORCE_PROXY_REGEXES = None -pp._FORCE_PROXY_LOADED = False -pp._is_force_proxy.cache_clear() -pp._load_force_proxy_suffixes() - -rule_cases = [ - ("普通后缀匹配子域", "cdn.suffix.example", True), - ("full 精确匹配", "only.example", True), - ("full 不匹配子域", "cdn.only.example", False), - ("regexp 匹配", "edge-42.example", True), - ("regexp 不匹配", "edge-x.example", False), - ("关键 Google CDN 兜底", "assets.googlecdn.com", True), - ("无关域名", "unmatched.example", False), -] -for desc, host, expect in rule_cases: - got = pp._is_force_proxy(host) - ok = got == expect - rule_failed += 0 if ok else 1 - print(f" {desc:<22} {host:<30} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") - -pp._USER_DIR = old_user_dir -pp._FORCE_PROXY_SUFFIXES = old_suffixes -pp._FORCE_PROXY_EXACT = old_exact -pp._FORCE_PROXY_REGEXES = old_regexes -pp._FORCE_PROXY_LOADED = old_loaded -pp._is_force_proxy.cache_clear() -if rule_failed: - failed += rule_failed -print("=" * 70) - -# ── 明确直连域名：优先级 / Trie / 固定代理隔离 ─────────────────── -print("\n" + "=" * 70) -print("明确直连域名（后缀 / full / regexp / 强制代理优先 / 固定代理隔离）") -print("=" * 70) - -direct_failed = 0 -direct_dir = Path(tempfile.mkdtemp()) / "RRPP" -direct_dir.mkdir(parents=True) -(direct_dir / "pac_direct_domains.json").write_text(json.dumps({"domains": [ - "direct.example", "full:only-direct.example", "regexp:^edge-direct-[0-9]+\\.example$", - "claude.ai", -]}), "utf-8") -old_user_dir = pp._USER_DIR -old_direct_trie = pp._DIRECT_SUFFIX_TRIE -old_direct_exact = pp._DIRECT_EXACT -old_direct_regexes = pp._DIRECT_REGEXES -old_direct_loaded = pp._DIRECT_LOADED -old_data_path = pp._data_path -try: - pp._USER_DIR = direct_dir - pp._data_path = lambda name: None # 隔离仓库内置种子，覆盖到临时用户数据。 - pp._DIRECT_SUFFIX_TRIE = None - pp._DIRECT_EXACT = None - pp._DIRECT_REGEXES = None - pp._DIRECT_LOADED = False - pp._is_direct_domain.cache_clear() - pp._load_direct_domains() - direct_cases = [ - ("后缀匹配子域", "cdn.direct.example", True), - ("full 精确匹配", "only-direct.example", True), - ("full 不匹配子域", "cdn.only-direct.example", False), - ("regexp 匹配", "edge-direct-42.example", True), - ("regexp 不匹配", "edge-direct-x.example", False), - ] - for desc, host, expect in direct_cases: - got = pp._is_direct_domain(host) - ok = got == expect - direct_failed += 0 if ok else 1 - print(f" {desc:<22} {host:<30} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") - smart_direct = pp._decide_direct("cdn.direct.example", True, True) - fixed_proxy = not pp._decide_direct("cdn.direct.example", True, False) - force_wins = not pp._decide_direct("claude.ai", True, True) - direct_target = pp._direct_connect_target("cdn.direct.example", 443, True, True) - target_keeps_host = direct_target == ("cdn.direct.example", 443) - print(f" PAC 模式命中明确直连: {smart_direct} {'PASS' if smart_direct else 'FAIL'}") - print(f" 固定代理模式不使用明确直连表: {fixed_proxy} {'PASS' if fixed_proxy else 'FAIL'}") - print(f" 强制代理优先于明确直连: {force_wins} {'PASS' if force_wins else 'FAIL'}") - print(f" 明确直连保持原始主机名: {target_keeps_host} {'PASS' if target_keeps_host else 'FAIL'}") - direct_failed += 0 if smart_direct else 1 - direct_failed += 0 if fixed_proxy else 1 - direct_failed += 0 if force_wins else 1 - direct_failed += 0 if target_keeps_host else 1 - pp._pac_write("pac_direct_domains", {"domains": ["updated-direct.example"]}) - pp._invalidate_pac_routing_cache("pac_direct_domains") - direct_reloaded = (not pp._is_direct_domain("direct.example") - and pp._is_direct_domain("updated-direct.example")) - print(f" 明确直连表热更新立即生效: {direct_reloaded} " - f"{'PASS' if direct_reloaded else 'FAIL'}") - direct_failed += 0 if direct_reloaded else 1 -finally: - pp._USER_DIR = old_user_dir - pp._data_path = old_data_path - pp._DIRECT_SUFFIX_TRIE = old_direct_trie - pp._DIRECT_EXACT = old_direct_exact - pp._DIRECT_REGEXES = old_direct_regexes - pp._DIRECT_LOADED = old_direct_loaded - pp._is_direct_domain.cache_clear() -if direct_failed: - failed += direct_failed -print("=" * 70) - -# ── PAC 更新立即生效 + DNS 解析一致性 ────────────────────────── -print("\n" + "=" * 70) -print("PAC 热更新缓存失效 + 国内直连使用同一次 DNS 解析结果") -print("=" * 70) - -hot_failed = 0 -hot_dir = Path(tempfile.mkdtemp()) / "RRPP" -hot_dir.mkdir(parents=True) -old_user_dir = pp._USER_DIR -old_cn_ranges = pp._CN_RANGES -old_cn_ranges_v6 = pp._CN_RANGES_V6 -old_cn_loaded = pp._CN_RANGES_LOADED -old_suffixes = pp._FORCE_PROXY_SUFFIXES -old_exact = pp._FORCE_PROXY_EXACT -old_regexes = pp._FORCE_PROXY_REGEXES -old_force_loaded = pp._FORCE_PROXY_LOADED -old_getaddrinfo = socket.getaddrinfo -try: - pp._USER_DIR = hot_dir - pp._CN_RANGES = None - pp._CN_RANGES_V6 = None - pp._CN_RANGES_LOADED = False - pp._FORCE_PROXY_SUFFIXES = None - pp._FORCE_PROXY_EXACT = None - pp._FORCE_PROXY_REGEXES = None - pp._FORCE_PROXY_LOADED = False - pp._is_china_ip.cache_clear() - pp._is_force_proxy.cache_clear() - - pp._pac_write("pac_domains", {"domains": ["old-rule.example"]}) - pp._pac_write("pac_ip_cn", {"cidrs": ["114.114.114.0/24"]}) - pp._load_force_proxy_suffixes() - pp._load_cn_ranges() - before = pp._is_force_proxy("old-rule.example") - pp._pac_write("pac_domains", {"domains": ["new-rule.example"]}) - pp._invalidate_pac_routing_cache("pac_domains") - domains_reloaded = (not pp._is_force_proxy("old-rule.example") - and pp._is_force_proxy("new-rule.example")) - print(f" 域名表热更新立即生效: 初始={before}, 更新后={domains_reloaded} " - f"{'PASS' if before and domains_reloaded else 'FAIL'}") - hot_failed += 0 if before and domains_reloaded else 1 - - initial_cn = pp._is_china_ip("114.114.114.114") - pp._pac_write("pac_ip_cn", {"cidrs": ["8.8.8.0/24"]}) - pp._invalidate_pac_routing_cache("pac_ip_cn") - ip_reloaded = (not pp._is_china_ip("114.114.114.114") - and pp._is_china_ip("8.8.8.8")) - print(f" IP 段热更新立即生效: 初始={initial_cn}, 更新后={ip_reloaded} " - f"{'PASS' if initial_cn and ip_reloaded else 'FAIL'}") - hot_failed += 0 if initial_cn and ip_reloaded else 1 - - # 固定本次 DNS 返回值；_direct_connect_target 必须只解析一次并直连该 IP。 - pp._pac_write("pac_ip_cn", {"cidrs": ["114.114.114.0/24"]}) - pp._invalidate_pac_routing_cache("pac_ip_cn") - dns_calls = [] - def fake_getaddrinfo(host, *args, **kwargs): - dns_calls.append(host) - if host == "mixed.example": - ips = ["114.114.114.114", "8.8.8.8"] - else: - ips = ["114.114.114.114"] - return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, 0)) for ip in ips] - socket.getaddrinfo = fake_getaddrinfo - target = pp._direct_connect_target("cn-cdn.example", 443, False, True) - stable_dns = target == ("114.114.114.114", 443) and dns_calls == ["cn-cdn.example"] - mixed = pp._direct_connect_target("mixed.example", 443, False, True) - mixed_safe = mixed is None - print(f" 国内 CDN 固定解析 IP 且不二次解析: {stable_dns} {'PASS' if stable_dns else 'FAIL'}") - print(f" 混合中国/境外 CDN 记录改走隧道: {mixed_safe} {'PASS' if mixed_safe else 'FAIL'}") - hot_failed += 0 if stable_dns else 1 - hot_failed += 0 if mixed_safe else 1 - - # IPv6 中国 CIDR 与 IPv6 私网/回环分流。240e::/16 是测试用的公网 IPv6 段。 - pp._pac_write("pac_ip_cn", {"cidrs": ["240e::/16"]}) - pp._invalidate_pac_routing_cache("pac_ip_cn") - ipv6_cn = (pp._is_china_ip("240e:1234::1") - and pp._direct_connect_target("240e:1234::1", 443, False, True) - == ("240e:1234::1", 443)) - ipv6_private = all(pp._is_private_ip(ip) for ip in ("::1", "fc00::1", "fe80::1")) - ipv6_public = not pp._is_private_ip("2001:4860:4860::8888") - print(f" IPv6 中国 CIDR 智能直连: {ipv6_cn} {'PASS' if ipv6_cn else 'FAIL'}") - print(f" IPv6 回环/ULA/链路本地直连: {ipv6_private and ipv6_public} " - f"{'PASS' if ipv6_private and ipv6_public else 'FAIL'}") - hot_failed += 0 if ipv6_cn else 1 - hot_failed += 0 if ipv6_private and ipv6_public else 1 -finally: - socket.getaddrinfo = old_getaddrinfo - pp._USER_DIR = old_user_dir - pp._CN_RANGES = old_cn_ranges - pp._CN_RANGES_V6 = old_cn_ranges_v6 - pp._CN_RANGES_LOADED = old_cn_loaded - pp._FORCE_PROXY_SUFFIXES = old_suffixes - pp._FORCE_PROXY_EXACT = old_exact - pp._FORCE_PROXY_REGEXES = old_regexes - pp._FORCE_PROXY_LOADED = old_force_loaded - pp._is_china_ip.cache_clear() - pp._is_force_proxy.cache_clear() -if hot_failed: - failed += hot_failed -print("=" * 70) - -# ── 智能国内直连失败回退上游隧道 ─────────────────────────────── -print("\n" + "=" * 70) -print("智能国内直连失败：仅该路径回退一次上游隧道") -print("=" * 70) - -fallback_failed = 0 -direct = ("114.114.114.114", 443) -upstream = ("node.example", 1205) -calls = [] -remote_obj = object() -def fake_connect(target, timeout): - calls.append((target, timeout)) - if target == direct: - raise OSError("mock direct connect failure") - if target == upstream: - return remote_obj - raise AssertionError(f"unexpected target: {target}") - -remote, via_tunnel = pp._connect_route( - direct, True, upstream[0], upstream[1], None, False, fake_connect) -fallback_ok = (remote is remote_obj and via_tunnel is True - and calls == [(direct, 10), (upstream, 10)]) -calls.clear() -try: - pp._connect_route(direct, False, upstream[0], upstream[1], None, False, fake_connect) - explicit_not_fallback = False -except OSError: - explicit_not_fallback = calls == [(direct, 10)] -print(f" 智能中国公网直连失败回退隧道: {fallback_ok} {'PASS' if fallback_ok else 'FAIL'}") -print(f" 显式直连失败不回退: {explicit_not_fallback} " - f"{'PASS' if explicit_not_fallback else 'FAIL'}") -fallback_failed += 0 if fallback_ok else 1 -fallback_failed += 0 if explicit_not_fallback else 1 -if fallback_failed: - failed += fallback_failed -print("=" * 70) - -# ── 启动连通性必须验证境外隧道 ───────────────────────────────── -print("\n" + "=" * 70) -print("启动连通性：google.com 隧道失败必须阻止“开启成功”") -print("=" * 70) - -class _FakeResponse: - def __enter__(self): - return self - def __exit__(self, *args): - return False - def read(self, size=-1): - return b"" - -class _FakeOpener: - def __init__(self, fail_google=False): - self.fail_google = fail_google - def open(self, url, timeout): - if self.fail_google and "www.google.com" in url: - raise OSError("mock tunnel failure") - return _FakeResponse() - - -class _CancellingResponse(_FakeResponse): - def __init__(self, cancel_event): - self.cancel_event = cancel_event - def read(self, size=-1): - self.cancel_event.set() - return b"" - - -class _CancellingOpener: - def __init__(self, cancel_event): - self.cancel_event = cancel_event - self.urls = [] - def open(self, url, timeout): - self.urls.append(url) - return _CancellingResponse(self.cancel_event) - - -class _GoogleCancellingOpener(_CancellingOpener): - def open(self, url, timeout): - self.urls.append(url) - if "www.google.com" in url: - return _CancellingResponse(self.cancel_event) - return _FakeResponse() - -verify_failed = 0 -ok_results = pp._verify_proxy_connectivity(opener=_FakeOpener()) -ok = ok_results.get("google.com") is True -try: - pp._verify_proxy_connectivity(opener=_FakeOpener(fail_google=True)) - blocked = False -except RuntimeError: - blocked = True -cancel_event = threading.Event() -cancelling_opener = _CancellingOpener(cancel_event) -try: - pp._verify_proxy_connectivity(opener=cancelling_opener, cancel_event=cancel_event) - cancel_blocked = False -except pp._StartCancelled: - cancel_blocked = True -google_not_started = all("www.google.com" not in url for url in cancelling_opener.urls) -google_cancel_event = threading.Event() -google_cancelling_opener = _GoogleCancellingOpener(google_cancel_event) -try: - pp._verify_proxy_connectivity(opener=google_cancelling_opener, - cancel_event=google_cancel_event) - google_cancelled = False -except pp._StartCancelled: - google_cancelled = True -google_started = any("www.google.com" in url for url in google_cancelling_opener.urls) -print(f" 境外隧道成功可通过: {ok} {'PASS' if ok else 'FAIL'}") -print(f" 境外隧道失败会阻止开启: {blocked} {'PASS' if blocked else 'FAIL'}") -print(f" Baidu 探测中取消会中断后续 Google 探测: {cancel_blocked and google_not_started} " - f"{'PASS' if cancel_blocked and google_not_started else 'FAIL'}") -print(f" Google 探测中取消会终止本次开启: {google_cancelled and google_started} " - f"{'PASS' if google_cancelled and google_started else 'FAIL'}") -verify_failed += 0 if ok else 1 -verify_failed += 0 if blocked else 1 -verify_failed += 0 if cancel_blocked and google_not_started else 1 -verify_failed += 0 if google_cancelled and google_started else 1 -if verify_failed: - failed += verify_failed -print("=" * 70) - -# ── DNS 分流 TTL 缓存 ──────────────────────────────────────── -print("\n" + "=" * 70) -print("TTL 缓存：cache_clear 兼容 + 缓存命中") -print("=" * 70) - -ttl_failed = 0 -assert hasattr(pp._is_china_ip, "cache_clear"), "cache_clear 缺失" -assert hasattr(pp._is_private_ip, "cache_clear"), "cache_clear 缺失" -assert hasattr(pp._is_whitelisted, "cache_clear"), "cache_clear 缺失" -pp._is_china_ip.cache_clear() -pp._is_private_ip.cache_clear() -pp._is_whitelisted.cache_clear() -a1, a2 = pp._is_private_ip("10.1.2.3"), pp._is_private_ip("10.1.2.3") -ok = a1 is True and a2 is True -ttl_failed += 0 if ok else 1 -print(f" 内网 IP 判定缓存命中: {a1}/{a2} {'PASS' if ok else 'FAIL'}") -b1, b2 = pp._is_china_ip("114.114.114.114"), pp._is_china_ip("114.114.114.114") -ok = b1 is True and b2 is True -ttl_failed += 0 if ok else 1 -print(f" 中国 IP 判定缓存命中: {b1}/{b2} {'PASS' if ok else 'FAIL'}") -pp._is_china_ip.cache_clear() -pp._is_private_ip.cache_clear() -print(f" cache_clear 后重查: {pp._is_china_ip('114.114.114.114')} " - f"{'PASS' if pp._is_china_ip('114.114.114.114') is True else 'FAIL'}") -if ttl_failed: - failed += ttl_failed -print("=" * 70) - -print(f"\n最终结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}") -sys.exit(1 if failed else 0)