修复：Nuitka onefile 更新路径

.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

lite/MODULE_SPLIT_PLAN.md

@@ -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 基线；保留失败现场的测试输出以便定位。不得通过覆盖用户的运行配置、订阅或构建目录来“修复”回归。

lite/build_pyinstaller.bat

@@ -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

lite/claude.bat

@@ -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

lite/main.py

@@ -6,7 +6,7 @@ from pathlib import Path
 from PyQt5.QtWidgets import (
     QApplication, QWidget, QLabel, QPushButton,
     QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit,
-    QCheckBox, QMenu, QDialog,
+    QCheckBox, QMenu, QDialog, QProgressDialog,
 )
 from PyQt5.QtCore import Qt, QTimer, pyqtSignal
 from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor
@@ -61,10 +61,21 @@ _BASE_WINDOW_HEIGHT = 470  # 内容区 + 加大日志区 + 底部两行提示小
 # 已被打包为 EXE（PyInstaller 的 sys.frozen / Nuitka 的 __compiled__）：
 #   BASE_DIR   -> 内置数据目录：PyInstaller 用 _MEIPASS，Nuitka onefile 用
 #                 __file__ 所在解压临时目录（无 _MEIPASS 时回退）
-#   PERSIST_DIR -> EXE 所在目录，服务安装等需要持久化的操作都放这里
+#   PERSIST_DIR -> EXE 所在目录，服务安装等需要持久化的操作都放这里。
+# Nuitka onefile 的 sys.executable 指向 TEMP\onefile_* 解包子进程，不能用它
+# 定位用户启动的原始 EXE；优先使用 __compiled__.containing_dir，旧版 Nuitka
+# 再回退到 sys.argv[0]（官方保证它指向原始 onefile 路径）。
+def _frozen_executable_dir():
+    if "__compiled__" in globals():
+        compiled_dir = getattr(globals()["__compiled__"], "containing_dir", None)
+        if compiled_dir:
+            return Path(compiled_dir).resolve()
+        return Path(sys.argv[0]).resolve().parent
+    return Path(sys.executable).resolve().parent
+
 if getattr(sys, 'frozen', False) or "__compiled__" in globals():
     BASE_DIR = Path(getattr(sys, "_MEIPASS", None) or Path(__file__).resolve().parent)
-    PERSIST_DIR = Path(sys.executable).resolve().parent
+    PERSIST_DIR = _frozen_executable_dir()
 else:
     BASE_DIR = Path(__file__).resolve().parent
     PERSIST_DIR = BASE_DIR
@@ -1652,6 +1663,7 @@ class MainWindow(QWidget):
     _op_done = pyqtSignal(str, bool)
     _update_done = pyqtSignal()
     _update_download_done = pyqtSignal(bool, str)
+    _update_download_progress = pyqtSignal(int, int, int)
 
     _status_ready = pyqtSignal(object)
     _node_switched = pyqtSignal()   # 启动校验时代替用户切换了节点（刷新标题栏）
@@ -1665,6 +1677,7 @@ class MainWindow(QWidget):
         self._start_in_progress = False
         self._proxy_state_lock = threading.RLock()
         self._status_busy = False      # 防止状态轮询线程堆叠
+        self._update_progress_dialog = None
         # 把全局 log() 函数接到 GUI 日志区
         global _log_func
         _log_func = self.log_signal.emit
@@ -1672,6 +1685,7 @@ class MainWindow(QWidget):
         self._op_done.connect(self._after_op)
         self._update_done.connect(self._on_update_done)
         self._update_download_done.connect(self._on_update_download_done)
+        self._update_download_progress.connect(self._on_update_download_progress)
         self._status_ready.connect(self._apply_status)
         self._node_switched.connect(self._on_node_switched)
 
@@ -2338,8 +2352,35 @@ class MainWindow(QWidget):
                 QMessageBox.Yes,
             )
             if reply == QMessageBox.Yes:
+                self._show_update_download_progress()
                 threading.Thread(target=self._download_update, daemon=True).start()
 
+    def _show_update_download_progress(self):
+        """在 GUI 线程创建下载进度窗口；网络线程只通过 signal 回传状态。"""
+        dialog = QProgressDialog("正在准备下载更新…", None, 0, 100, self)
+        dialog.setWindowTitle("正在下载更新")
+        dialog.setWindowModality(Qt.WindowModal)
+        dialog.setMinimumDuration(0)
+        dialog.setAutoClose(False)
+        dialog.setAutoReset(False)
+        dialog.setValue(0)
+        dialog.show()
+        self._update_progress_dialog = dialog
+
+    def _on_update_download_progress(self, percent, downloaded, total):
+        """把后台下载进度渲染到进度窗口。"""
+        dialog = self._update_progress_dialog
+        if dialog is None:
+            return
+        if total > 0:
+            dialog.setRange(0, 100)
+            dialog.setValue(max(0, min(100, percent)))
+            dialog.setLabelText(
+                f"正在下载更新… {percent}%（{_fmt_bytes(downloaded)} / {_fmt_bytes(total)}）")
+        else:
+            dialog.setRange(0, 0)
+            dialog.setLabelText(f"正在下载更新… 已下载 {_fmt_bytes(downloaded)}")
+
     def _download_update(self):
         """后台下载新版本 EXE 到 EXE 所在目录。
 
@@ -2357,58 +2398,98 @@ class MainWindow(QWidget):
 
             # 对 URL 进行安全编码，兼容空格或已编码空格
             url = urllib.parse.quote(url, safe=':/?#[]@!$&\'()*+,;=%')
-            save_path = PERSIST_DIR / f"Royirin Rosmontis Private Proxy {self._latest_version}.exe"
+            # 冻结程序必须放到当前 EXE 的目录，便于用户关闭旧版后就地替换；
+            # 源码调试则仍写入源码目录，避免误把测试文件写到 Python 安装目录。
+            if getattr(sys, 'frozen', False) or "__compiled__" in globals():
+                download_dir = _frozen_executable_dir()
+            else:
+                download_dir = PERSIST_DIR
+            save_path = download_dir / f"Royirin Rosmontis Private Proxy {self._latest_version}.exe"
+            part_path = save_path.with_suffix(save_path.suffix + ".part")
             log(f"正在下载新版本: {url}")
             log(f"保存到: {save_path}")
 
-            def fetch(target):
-                # 同更新检查：按「使用代理更新」设置选择隧道或直连
-                if _update_via_proxy():
+            use_proxy = _update_via_proxy()
+            started_here = _ensure_forwarder() if use_proxy else False
+            try:
+                if use_proxy:
                     opener = urllib.request.build_opener(urllib.request.ProxyHandler(
                         {"http": f"{PROXY_HOST}:{PROXY_PORT}",
                          "https": f"{PROXY_HOST}:{PROXY_PORT}"}))
                 else:
                     opener = _pac_no_proxy_opener()
-                with opener.open(urllib.request.Request(target), timeout=120) as resp:
-                    return resp.read()
 
-            started_here = _ensure_forwarder() if _update_via_proxy() else False
-            try:
-                data = fetch(url)
+                if part_path.exists():
+                    part_path.unlink()
+                downloaded = 0
+                hasher = hashlib.sha256()
+                last_percent = -1
+                with opener.open(urllib.request.Request(url), timeout=120) as resp, part_path.open("wb") as fp:
+                    try:
+                        response_size = int(resp.headers.get("Content-Length") or 0)
+                    except (TypeError, ValueError):
+                        response_size = 0
+                    total = self._asset_size or response_size
+                    self._update_download_progress.emit(0, 0, total)
+                    log(f"更新下载进度: 0%（0 / {_fmt_bytes(total)}）" if total else "更新下载进度: 已开始")
+                    while True:
+                        chunk = resp.read(256 * 1024)
+                        if not chunk:
+                            break
+                        fp.write(chunk)
+                        hasher.update(chunk)
+                        downloaded += len(chunk)
+                        if total:
+                            percent = min(100, downloaded * 100 // total)
+                            if percent != last_percent:
+                                last_percent = percent
+                                log(f"更新下载进度: {percent}%（{_fmt_bytes(downloaded)} / {_fmt_bytes(total)}）")
+                                self._update_download_progress.emit(percent, downloaded, total)
+
                 checksum_text = ""
                 if self._checksum_url:
-                    checksum_text = fetch(self._checksum_url).decode("ascii", "replace")
+                    checksum_url = urllib.parse.quote(
+                        self._checksum_url, safe=':/?#[]@!$&\'()*+,;=%')
+                    with opener.open(urllib.request.Request(checksum_url), timeout=120) as resp:
+                        checksum_text = resp.read().decode("ascii", "replace")
             finally:
                 _release_forwarder(started_here)
 
-            if len(data) < 1024:
-                self._update_download_done.emit(False, "下载文件过小，可能不是有效的 EXE")
-                return
-            if self._asset_size and len(data) != self._asset_size:
-                self._update_download_done.emit(
-                    False, f"下载不完整：实际 {len(data)} 字节，应为 {self._asset_size} 字节")
-                return
+            if downloaded < 1024:
+                raise RuntimeError("下载文件过小，可能不是有效的 EXE")
+            if self._asset_size and downloaded != self._asset_size:
+                raise RuntimeError(
+                    f"下载不完整：实际 {downloaded} 字节，应为 {self._asset_size} 字节")
             if checksum_text:
                 want = checksum_text.strip().split()[0].lower()
-                got = hashlib.sha256(data).hexdigest()
+                got = hasher.hexdigest()
                 if len(want) != 64 or want != got:
-                    self._update_download_done.emit(False, "更新包校验失败（SHA-256 不匹配），已放弃")
-                    return
+                    raise RuntimeError("更新包校验失败（SHA-256 不匹配），已放弃")
                 log("更新包 SHA-256 校验通过")
 
-            save_path.write_bytes(data)
+            part_path.replace(save_path)
             log(f"新版本已下载: {save_path}")
             self._update_download_done.emit(True, str(save_path))
         except Exception as e:
+            try:
+                if 'part_path' in locals():
+                    part_path.unlink(missing_ok=True)
+            except Exception:
+                pass
             log(f"下载更新失败: {e}")
             self._update_download_done.emit(False, str(e))
 
     def _on_update_download_done(self, success, message):
         """下载完成后提示用户手动替换 EXE。"""
+        dialog = self._update_progress_dialog
+        self._update_progress_dialog = None
+        if dialog is not None:
+            dialog.close()
+            dialog.deleteLater()
         if success:
             QMessageBox.information(
                 self, "下载完成",
-                f"新版本已下载到:\n{message}\n\n请关闭本程序后，用该文件替换当前运行的程序，然后重新启动。"
+                f"新版本已下载到当前程序所在文件夹:\n{message}\n\n请关闭本程序后，用该文件替换当前运行的程序，然后重新启动。"
             )
         else:
             QMessageBox.warning(self, "下载失败", f"无法下载新版本:\n{message}")

lite/others/nodes_tool/nodes_tool.py

@@ -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()

lite/others/update_pac_domains/update_pac_domains.py

@@ -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()

lite/others/update_pac_ip_cn/update_pac_ip_cn.py

@@ -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()

lite/test.bat

@@ -1,4 +0,0 @@
-@echo off
-cd /d "%~dp0"
-call ..\.venv\Scripts\python.exe main.py
-pause

lite/test/PyQt5/QtCore.py

@@ -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

lite/test/PyQt5/QtGui.py

@@ -1,6 +0,0 @@
-class QFont: pass
-class QIcon: pass
-class QPixmap: pass
-class QImage: pass
-class QColor: pass
-class QPainter: pass

lite/test/PyQt5/QtWidgets.py

@@ -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

lite/test/PyQt5/__init__.py

@@ -1,2 +0,0 @@
-# PyQt5 桩模块：仅供 WSL 下 import main 做路由逻辑测试，
-# 决策函数（_decide_direct 等）不依赖任何 Qt 对象。

lite/test/check_node_menu.py

@@ -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")

lite/test/nodes.json

@@ -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 测试节点"
-  }
-]

lite/test/probe_nodes.py

@@ -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()

lite/test/test_core_modules.py

@@ -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)

lite/test/test_pac_routing.py

@@ -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)