feat(lite): 持久化代理模式配置并发布 9999091816

.gitignore

@@ -1,2 +1,12 @@
 __pycache__/
 *.pyc
+
+# Local development state and reproducible build outputs.
+.claude/
+lite/.claude/
+lite/backups/
+lite/build/
+lite/dist/
+lite/RELEASE_WORKFLOW.md
+*.bak
+*.bak-*

README.md

@@ -3,7 +3,7 @@ Supported by Global Rosmanka Cloud Nexus Inc. —— https://rosmanka.rrrosmonti
 ## 警告
 - 包含 WireGuard 的 common 版本转发器永久不再维护，推荐使用 Lite 版本
 > common 版本现在有许多未修复的 bug (如部分 UI 的显示问题) 、令人难以忍受的代码逻辑 (如 PAC 规则直接写在代码里)和安全性问题 (认证凭据的暴露与证书忽略)，而 Lite 版本修复了这些问题，并会收到持续的更新 
-- 本程序完全没有 Linux 与 CLI 功能兼容，默认构建为带有图形界面的 Windows .exe
+- 本程序完全没有 Linux 与 CLI 功能兼容，默认构建为带有图形界面的 Windows 应用 (.exe)
 
 ### 如何区分 common 版本与 Lite 版本
 - Lite 版本的版本号前四位固定为 9999，后续为从月份开始的构建时间戳，而 common 版本的版本号为包含年份的完整的构建时间戳

lite/RRPP_Lite.spec

@@ -0,0 +1,85 @@
+# -*- mode: python ; coding: utf-8 -*-
+
+from pathlib import Path
+from PyInstaller.utils.hooks import collect_all
+
+# 强制收集 PyQt5 全部模块及二进制文件
+qt_datas, qt_binaries, qt_hiddenimports = collect_all('PyQt5')
+
+# 只内置 config.json（普通设置）。nodes.json / pac_*.json 属用户数据：
+# 运行时读 %LOCALAPPDATA%\RRPP（首启自动创建模板 / 从 CDN 拉取），不进 EXE。
+_JSON_DATAS = [(str(Path(SPECPATH) / 'config.json'), '.')]
+print(f"[spec] 打包 json 数据文件: {[Path(src).name for src, _ in _JSON_DATAS]}")
+
+# 自定义 manifest：普通权限运行 + 声明 PerMonitorV2 DPI 感知，
+# 避免程序在不同缩放比例的显示器之间移动时被 Windows 位图拉伸。
+_MANIFEST = r'''<?xml version="1.0" encoding="UTF-8" standalone="yes"?>
+<assembly xmlns="urn:schemas-microsoft-com:asm.v1" manifestVersion="1.0">
+  <assemblyIdentity version="1.0.0.0" name="RRPP_Lite" type="win32"/>
+  <description>Royirin Rosmontis Private Proxy Lite</description>
+  <trustInfo xmlns="urn:schemas-microsoft-com:asm.v2">
+    <security>
+      <requestedPrivileges>
+        <requestedExecutionLevel level="asInvoker" uiAccess="false"/>
+      </requestedPrivileges>
+    </security>
+  </trustInfo>
+  <compatibility xmlns="urn:schemas-microsoft-com:compatibility.v1">
+    <application>
+      <supportedOS Id="{8e0f7a12-bfb3-4fe8-b9a5-48fd50a15a9a}"/>
+      <supportedOS Id="{1f676c76-80e1-4239-95bb-83d0f6d0da78}"/>
+    </application>
+  </compatibility>
+  <application xmlns="urn:schemas-microsoft-com:asm.v3">
+    <windowsSettings>
+      <dpiAwareness xmlns="http://schemas.microsoft.com/SMI/2016/WindowsSettings">PerMonitorV2, PerMonitor, system</dpiAwareness>
+      <dpiAware xmlns="http://schemas.microsoft.com/SMI/2005/WindowsSettings">true/pm</dpiAware>
+    </windowsSettings>
+  </application>
+  <dependency>
+    <dependentAssembly>
+      <assemblyIdentity type="win32" name="Microsoft.Windows.Common-Controls" version="6.0.0.0" processorArchitecture="*" publicKeyToken="6595b64144ccf1df" language="*"/>
+    </dependentAssembly>
+  </dependency>
+</assembly>
+'''
+
+a = Analysis(
+    ['main.py'],
+    pathex=['..\\.venv\\Lib\\site-packages'],
+    binaries=qt_binaries,
+    datas=[('ico.ico', '.')] + _JSON_DATAS + qt_datas,
+    hiddenimports=qt_hiddenimports,
+    hookspath=[],
+    hooksconfig={},
+    runtime_hooks=[],
+    # 排除 Nuitka 构建链相关模块（venv 里装了 Nuitka，但 PyInstaller 产物不需要）
+    excludes=['nuitka', 'zstandard', 'ordered_set'],
+    noarchive=False,
+    optimize=0,
+)
+pyz = PYZ(a.pure)
+
+exe = EXE(
+    pyz,
+    a.scripts,
+    a.binaries,
+    a.datas,
+    [],
+    name='Royirin Rosmontis Private Proxy Lite Python',
+    debug=False,
+    bootloader_ignore_signals=False,
+    strip=False,
+    # UPX 会压缩资源段（含图标），冷缓存下 Explorer 可能提取不到图标导致任务栏空白
+    upx=False,
+    upx_exclude=[],
+    runtime_tmpdir=None,
+    console=False,
+    disable_windowed_traceback=False,
+    argv_emulation=False,
+    target_arch=None,
+    codesign_identity=None,
+    entitlements_file=None,
+    icon=['ico.ico'],
+    manifest=_MANIFEST,
+)

lite/build_lite.bat

@@ -0,0 +1,19 @@
+@echo off
+cd /d "%~dp0"
+call ..\.venv\Scripts\python.exe -m PyInstaller --version >nul 2>&1
+if errorlevel 1 (
+    echo PyInstaller not installed. Run first:
+    echo   ..\.venv\Scripts\python.exe -m pip install pyinstaller
+    pause
+    exit /b 1
+)
+call ..\.venv\Scripts\python.exe -m PyInstaller RRPP_Lite.spec --noconfirm
+if errorlevel 1 (
+    echo.
+    echo Build failed.
+    pause
+    exit /b 1
+)
+echo.
+echo Build complete. Output: dist\Royirin Rosmontis Private Proxy Lite Python.exe
+pause

lite/build_nuitka.bat

@@ -0,0 +1,42 @@
+@echo off
+cd /d "%~dp0"
+call ..\.venv\Scripts\python.exe -m nuitka --version >nul 2>&1
+if errorlevel 1 (
+    echo Nuitka not installed. Run first:
+    echo   ..\.venv\Scripts\python.exe -m pip install nuitka zstandard
+    pause
+    exit /b 1
+)
+rem Remove prior artifacts and the SCons incremental state before every build.
+rem The constants blob is embedded outside normal dependency tracking, so reuse
+rem can leave new UI strings and settings keys out of the executable.
+rd /s /q "dist\main.build" 2>nul
+rd /s /q "dist\main.dist" 2>nul
+rd /s /q "dist\main.onefile-build" 2>nul
+rem Use Nuitka COFF objects for embedded constants to avoid compiler cache reuse.
+rem Zig C23 embed caches do not reliably track embedded blob changes.
+set NUITKA_RESOURCE_MODE=coff_obj
+rem Convert config.json version to Windows file-version segments (each <= 65535).
+for /f %%v in ('..\.venv\Scripts\python.exe -c "import json;v=json.load(open('config.json',encoding='utf-8'))['version'];print(f'9999.{int(v[4:8])}.{int(v[8:12])}.0')"') do set "FILEVER=%%v"
+call ..\.venv\Scripts\python.exe -m nuitka ^
+  --standalone --onefile ^
+  --enable-plugin=pyqt5 ^
+  --include-data-files=config.json=config.json ^
+  --windows-icon-from-ico=ico.ico ^
+  --windows-console-mode=disable ^
+  --company-name="Global Rosmanka Cloud Nexus Inc" ^
+  --product-name="R.R.P.P." ^
+  --file-version=%FILEVER% ^
+  --output-dir=dist ^
+  --output-filename="Royirin Rosmontis Private Proxy Lite.exe" ^
+  --assume-yes-for-downloads ^
+  main.py
+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/config.json

@@ -1,6 +1,6 @@
 {
-  "version": "",
-  "update_url": "",
-  "pac_base_url": "",
-  "nodes_url": ""
+  "version": "999909181618",
+  "pac_base_url": "https://cdn.rrrosmontis.icu/RRPP/PAC/",
+  "nodes_url": "https://api.rrrosmontis.icu/v1/rrpp/subscription?method=rrpp_client",
+  "update_url": "https://git.rrrosmontis.icu/api/v1/repos/RoyirinRosmontis/Private-Proxy/releases/latest"
 }

lite/pac_direct_domains.json

@@ -0,0 +1,6 @@
+{
+  "version": "",
+  "date": "",
+  "description": "PAC 智能分流的明确直连域名模板。强制代理规则和用户白名单优先级更高。",
+  "domains": []
+}

lite/rrpp_core/__init__.py

@@ -0,0 +1 @@
+"""RRPP Lite 的无 GUI 核心逻辑。"""

lite/rrpp_core/forwarder_primitives.py

@@ -0,0 +1,91 @@
+"""本地转发器的无状态报文解析与改写函数。"""
+
+from urllib.parse import urlsplit
+
+
+HTTP_METHODS = (
+    b"CONNECT ", b"GET ", b"POST ", b"PUT ", b"DELETE ",
+    b"HEAD ", b"OPTIONS ", b"PATCH ",
+)
+
+
+def starts_with_http_method(data: bytes) -> bool:
+    return any(data.startswith(method) for method in HTTP_METHODS)
+
+
+def inject_proxy_auth(data: bytes, basic_auth: str) -> bytes:
+    """为完整 HTTP 代理请求注入上游 Basic Authorization。"""
+    if not starts_with_http_method(data):
+        return data
+    end = data.find(b"\r\n\r\n")
+    if end == -1:
+        return data
+    head = data[:end]
+    first_end = head.find(b"\r\n")
+    request_line = head if first_end == -1 else head[:first_end]
+    headers = b"" if first_end == -1 else head[first_end + 2:]
+    for line in headers.split(b"\r\n"):
+        if b":" not in line:
+            continue
+        name = line[:line.index(b":")].strip().lower()
+        if name == b"authorization":
+            return data
+    auth_line = f"Authorization: Basic {basic_auth}\r\n".encode()
+    return request_line + b"\r\n" + auth_line + headers + data[end:]
+
+
+def parse_connect_authority(first_line: str):
+    """从 CONNECT 首行提取 ``(host, port)``，失败时返回 ``None``。"""
+    parts = first_line.split(" ")
+    if len(parts) < 2:
+        return None
+    authority = parts[1]
+    if authority.startswith("[") and "]" in authority:
+        try:
+            host = authority[1:authority.index("]")]
+            tail = authority[authority.index("]") + 1:]
+            if tail.startswith(":") and tail[1:].isdigit():
+                return host, int(tail[1:])
+        except ValueError:
+            pass
+    if ":" not in authority:
+        return None
+    host, _, port_s = authority.rpartition(":")
+    if not host or not port_s.isdigit():
+        return None
+    return host, int(port_s)
+
+
+def parse_http_absolute(first_line: str):
+    """从明文 HTTP 代理首行提取 ``(host, port)``，失败时返回 ``None``。"""
+    parts = first_line.split(" ")
+    if len(parts) < 2 or not parts[1].lower().startswith(("http://", "https://")):
+        return None
+    parsed = urlsplit(parts[1])
+    if not parsed.hostname:
+        return None
+    port = parsed.port or (443 if parsed.scheme == "https" else 80)
+    return parsed.hostname, port
+
+
+def force_http_close(buffer: bytes) -> bytes:
+    """直连明文 HTTP 时强制 Connection: close，避免错复用连接。"""
+    end = buffer.find(b"\r\n\r\n")
+    if end == -1:
+        return buffer
+    head, tail = buffer[:end], buffer[end:]
+    lines = head.split(b"\r\n")
+    out = [lines[0]]
+    conn_seen = False
+    for line in lines[1:]:
+        low = line.lower()
+        if low.startswith(b"connection:"):
+            out.append(b"Connection: close")
+            conn_seen = True
+        elif low.startswith(b"proxy-connection:"):
+            continue
+        else:
+            out.append(line)
+    if not conn_seen:
+        out.append(b"Connection: close")
+    return b"\r\n".join(out) + tail

lite/rrpp_core/network_primitives.py

@@ -0,0 +1,82 @@
+"""路由、白名单和转发器共用的网络基础函数。"""
+
+import functools
+import ipaddress
+import socket
+import threading
+import time
+
+
+def ttl_cache(seconds: int, maxsize: int = 16384):
+    """带 TTL 的小型线程安全缓存，并保留 ``cache_clear`` 兼容接口。"""
+    def decorator(function):
+        store = {}
+        lock = threading.Lock()
+
+        @functools.wraps(function)
+        def wrapper(*args, **kwargs):
+            key = (args, tuple(sorted(kwargs.items())))
+            now = time.time()
+            with lock:
+                hit = store.get(key)
+            if hit and now - hit[0] < seconds:
+                return hit[1]
+            value = function(*args, **kwargs)
+            with lock:
+                store[key] = (now, value)
+                if len(store) > maxsize:
+                    cutoff = now - seconds
+                    for cached_key, (timestamp, _) in list(store.items()):
+                        if timestamp < cutoff:
+                            del store[cached_key]
+                    if len(store) > maxsize:
+                        store.clear()
+            return value
+
+        def cache_clear():
+            with lock:
+                store.clear()
+
+        wrapper.cache_clear = cache_clear
+        return wrapper
+    return decorator
+
+
+def is_private_ipv4_int(value: int) -> bool:
+    """整数形式 IPv4 是否属于常见私网、回环或链路本地段。"""
+    return (
+        (value & 0xFF000000) == 0x0A000000
+        or (value & 0xFFF00000) == 0xAC100000
+        or (value & 0xFFFF0000) == 0xC0A80000
+        or (value & 0xFF000000) == 0x7F000000
+        or (value & 0xFFFF0000) == 0xA9FE0000
+    )
+
+
+def resolve_host_ips(host: str):
+    """解析主机名的 IPv4/IPv6 集合；字面量 IP 返回自身，失败返回空集合。"""
+    try:
+        return {ipaddress.ip_address(host)}
+    except ValueError:
+        pass
+    try:
+        return {
+            ipaddress.ip_address(address[4][0])
+            for address in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM)
+        }
+    except Exception:
+        return set()
+
+
+def is_private_host(host: str) -> bool:
+    """目标是否命中私网、回环或 IPv6 ULA/链路本地地址。"""
+    for address in resolve_host_ips(host):
+        if address.version == 4 and is_private_ipv4_int(int(address)):
+            return True
+        if address.version == 6 and (
+            address.is_loopback
+            or address.is_link_local
+            or address in ipaddress.ip_network("fc00::/7")
+        ):
+            return True
+    return False

lite/rrpp_core/node_api.py

@@ -0,0 +1,36 @@
+"""节点订阅的认证请求协议。"""
+
+import base64
+import json
+import urllib.error
+import urllib.request
+
+
+class NodeAuthError(Exception):
+    """订阅服务拒绝请求时携带的 HTTP 状态和用户可读原因。"""
+
+    def __init__(self, message, code=0):
+        super().__init__(message)
+        self.code = code
+
+
+def b64url(value: str) -> str:
+    """SSO 查询参数使用的无 padding URL-safe Base64。"""
+    return base64.urlsafe_b64encode(str(value).encode("utf-8")).decode("ascii").rstrip("=")
+
+
+def fetch_nodes_auth(url, user, password, timeout=10):
+    """直连拉取需 SSO 查询参数认证的 nodes.json。"""
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    separator = "&" if "?" in url else "?"
+    request_url = f"{url}{separator}user={b64url(user)}&password={b64url(password)}"
+    try:
+        with opener.open(request_url, timeout=timeout) as response:
+            return json.loads(response.read().decode("utf-8"))
+    except urllib.error.HTTPError as error:
+        body = ""
+        try:
+            body = error.read().decode("utf-8", "replace").strip()
+        except Exception:
+            pass
+        raise NodeAuthError(body or f"请求被拒绝 (HTTP {error.code})", error.code) from None

lite/rrpp_core/node_secrets.py

@@ -0,0 +1,95 @@
+"""节点与登录凭据的本机加密存储。"""
+
+import ctypes
+import json
+import os
+
+
+class _DpapiBlob(ctypes.Structure):
+    _fields_ = [
+        ("cbData", ctypes.c_uint32),
+        ("pbData", ctypes.POINTER(ctypes.c_char)),
+    ]
+
+
+class NodeSecretsStore:
+    """使用 Windows DPAPI 保存节点和 SSO 凭据；非 Windows 保持明文兼容。"""
+
+    def __init__(self, user_dir_getter):
+        self._user_dir_getter = user_dir_getter
+
+    @staticmethod
+    def protect_bytes(data: bytes) -> bytes:
+        crypt32 = ctypes.WinDLL("crypt32")
+        kernel32 = ctypes.WinDLL("kernel32")
+        crypt32.CryptProtectData.argtypes = [
+            ctypes.POINTER(_DpapiBlob), ctypes.c_wchar_p, ctypes.POINTER(_DpapiBlob),
+            ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DpapiBlob),
+        ]
+        crypt32.CryptProtectData.restype = ctypes.c_bool
+        kernel32.LocalFree.argtypes = [ctypes.c_void_p]
+        buffer = ctypes.create_string_buffer(data, len(data))
+        blob_in = _DpapiBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_char)))
+        blob_out = _DpapiBlob()
+        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)
+
+    @staticmethod
+    def unprotect_bytes(data: bytes) -> bytes:
+        crypt32 = ctypes.WinDLL("crypt32")
+        kernel32 = ctypes.WinDLL("kernel32")
+        crypt32.CryptUnprotectData.argtypes = [
+            ctypes.POINTER(_DpapiBlob), ctypes.POINTER(ctypes.c_wchar_p), ctypes.POINTER(_DpapiBlob),
+            ctypes.c_void_p, ctypes.c_void_p, ctypes.c_uint32, ctypes.POINTER(_DpapiBlob),
+        ]
+        crypt32.CryptUnprotectData.restype = ctypes.c_bool
+        kernel32.LocalFree.argtypes = [ctypes.c_void_p]
+        buffer = ctypes.create_string_buffer(data, len(data))
+        blob_in = _DpapiBlob(len(data), ctypes.cast(buffer, ctypes.POINTER(ctypes.c_char)))
+        description = ctypes.c_wchar_p()
+        blob_out = _DpapiBlob()
+        if not crypt32.CryptUnprotectData(ctypes.byref(blob_in), ctypes.byref(description), 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 description.value:
+                kernel32.LocalFree(description)
+
+    def encrypt_text(self, text: str) -> bytes:
+        if os.name != "nt":
+            return text.encode("utf-8")
+        return self.protect_bytes(text.encode("utf-8"))
+
+    def decrypt_bytes(self, blob: bytes) -> str:
+        if os.name != "nt":
+            return blob.decode("utf-8")
+        try:
+            return self.unprotect_bytes(blob).decode("utf-8")
+        except Exception:
+            return blob.decode("utf-8")
+
+    def save_auth(self, username, password):
+        directory = self._user_dir_getter()
+        directory.mkdir(parents=True, exist_ok=True)
+        text = json.dumps({"username": username, "password": password}, ensure_ascii=False)
+        (directory / "auth.json").write_bytes(self.encrypt_text(text))
+
+    def load_auth(self):
+        try:
+            path = self._user_dir_getter() / "auth.json"
+            if not path.is_file():
+                return None
+            data = json.loads(self.decrypt_bytes(path.read_bytes()))
+            if not (isinstance(data, dict) and data.get("username") and data.get("password")):
+                return None
+            return data
+        except Exception:
+            return None

lite/rrpp_core/pac_data.py

@@ -0,0 +1,53 @@
+"""PAC 数据文件的网络传输与安全落盘。"""
+
+import json
+import os
+import urllib.request
+from pathlib import Path
+
+
+def read_version(path):
+    """读取 PAC JSON 的 version/date；无效文件返回空字符串。"""
+    try:
+        data = json.loads(Path(path).read_text("utf-8"))
+        return str(data.get("version") or ""), str(data.get("date") or "")
+    except Exception:
+        return "", ""
+
+
+def no_proxy_opener():
+    """创建忽略环境/系统代理的直连 HTTP opener。"""
+    return urllib.request.build_opener(urllib.request.ProxyHandler({}))
+
+
+def fetch_json(url, timeout=15, tries=2):
+    """以直连方式拉取 JSON；最后一次失败时保留原始异常。"""
+    opener = no_proxy_opener()
+    for attempt in range(tries):
+        try:
+            with opener.open(url, timeout=timeout) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except Exception:
+            if attempt == tries - 1:
+                raise
+
+
+def validate_document(name, data):
+    """校验各 PAC 文件的最小结构，避免错误响应覆盖有效缓存。"""
+    if not isinstance(data, dict):
+        return False
+    if name in ("pac_domains", "pac_direct_domains"):
+        return isinstance(data.get("domains"), list)
+    if name == "pac_ip_cn":
+        return isinstance(data.get("cidrs"), list)
+    return False
+
+
+def atomic_write(directory, name, data):
+    """将 PAC JSON 以 tmp + replace 原子写入指定用户目录。"""
+    directory = Path(directory)
+    directory.mkdir(parents=True, exist_ok=True)
+    target = directory / f"{name}.json"
+    temporary_path = directory / f"{name}.json.tmp"
+    temporary_path.write_text(json.dumps(data, ensure_ascii=False), "utf-8")
+    os.replace(temporary_path, target)

lite/rrpp_core/routing_primitives.py

@@ -0,0 +1,81 @@
+"""PAC 路由引擎使用的纯算法，不依赖文件、网络或 Qt。"""
+
+import ipaddress
+import re
+
+
+def merge_ip_ranges(cidrs, version):
+    """将指定 IP 版本的 CIDR 合并为整数闭区间，供二分查找。"""
+    ranges = []
+    for cidr in cidrs:
+        network = ipaddress.ip_network(str(cidr).strip(), strict=False)
+        if network.version == version:
+            ranges.append((int(network.network_address), int(network.broadcast_address)))
+    ranges.sort()
+    merged = []
+    for start, end in ranges:
+        if merged and start <= merged[-1][1] + 1:
+            if end > merged[-1][1]:
+                merged[-1] = (merged[-1][0], end)
+        else:
+            merged.append((start, end))
+    return merged
+
+
+def build_suffix_trie(suffixes):
+    """按反向域名标签建立 Trie，适合大规模后缀规则的 O(标签数) 匹配。"""
+    root = {}
+    for suffix in suffixes:
+        node = root
+        for label in reversed(suffix.split(".")):
+            node = node.setdefault(label, {})
+        node[""] = True
+    return root
+
+
+def suffix_trie_matches(host: str, trie) -> bool:
+    """判断主机名是否命中一个后缀 Trie。"""
+    node = trie
+    for label in reversed(host.rstrip(".").lower().split(".")):
+        node = node.get(label)
+        if node is None:
+            return False
+        if node.get(""):
+            return True
+    return bool(node and node.get(""))
+
+
+def parse_domain_rules(raw_rules, seed_suffixes=()):
+    """解析后缀、``full:`` 与 ``regexp:`` 三种 PAC 域名规则。
+
+    返回 ``(suffixes, exact, regexes, invalid_regex_count)``。调用方决定
+    这些规则是“强制代理”还是“明确直连”，从而避免解析层携带业务策略。
+    """
+    suffixes = set(seed_suffixes)
+    exact = set()
+    regexes = []
+    invalid_regexes = 0
+    for raw in raw_rules or ():
+        raw_rule = str(raw).strip()
+        rule = raw_rule.lower()
+        if not raw_rule:
+            continue
+        if rule.startswith("full:"):
+            host = rule[len("full:"):].rstrip(".")
+            if host:
+                exact.add(host)
+            continue
+        if rule.startswith("regexp:"):
+            pattern = raw_rule[len("regexp:"):]
+            if not pattern:
+                invalid_regexes += 1
+                continue
+            try:
+                regexes.append(re.compile(pattern, re.IGNORECASE))
+            except re.error:
+                invalid_regexes += 1
+            continue
+        suffix = rule.lstrip("*.").rstrip(".")
+        if suffix:
+            suffixes.add(suffix)
+    return suffixes, frozenset(exact), tuple(regexes), invalid_regexes

lite/rrpp_core/settings_store.py

@@ -0,0 +1,34 @@
+"""用户设置的 JSON 存储实现。"""
+
+import json
+import os
+
+
+class JsonSettingsStore:
+    """读取与原子写回一个 JSON 对象。
+
+    ``path_getter`` 在每次操作时调用，保证测试或运行时切换用户目录后不会
+    意外写到旧目录。
+    """
+
+    def __init__(self, path_getter):
+        self._path_getter = path_getter
+
+    def load(self):
+        path = self._path_getter()
+        try:
+            if not path.is_file():
+                return {}
+            data = json.loads(path.read_text("utf-8"))
+            return data if isinstance(data, dict) else {}
+        except Exception:
+            return {}
+
+    def save(self, data):
+        path = self._path_getter()
+        path.parent.mkdir(parents=True, exist_ok=True)
+        temporary_path = path.with_suffix(path.suffix + ".tmp")
+        temporary_path.write_text(
+            json.dumps(data, ensure_ascii=False, indent=2) + "\n", "utf-8"
+        )
+        os.replace(temporary_path, path)

lite/rrpp_core/third_party_subscriptions.py

@@ -0,0 +1,53 @@
+"""第三方 RRPP Client JSON 订阅的下载、校验与展示辅助。"""
+
+import json
+import urllib.request
+from urllib.parse import urlsplit, urlunsplit
+
+
+def fetch_nodes(url: str, timeout=15):
+    """直连读取一个第三方订阅，避免首次导入受系统代理残留影响。"""
+    opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
+    request = urllib.request.Request(
+        url,
+        headers={"Accept": "application/json", "User-Agent": "RRPP-Lite"},
+    )
+    with opener.open(request, timeout=timeout) as response:
+        return validate_nodes(json.loads(response.read().decode("utf-8-sig")))
+
+
+def validate_nodes(data):
+    """校验 RRPP Client 节点数组，返回可安全写入本地缓存的副本。"""
+    if not isinstance(data, list) or not data:
+        raise ValueError("订阅应为非空的 JSON 节点数组")
+    result = []
+    for index, node in enumerate(data, 1):
+        if not isinstance(node, dict):
+            raise ValueError(f"第 {index} 项不是节点对象")
+        domain = str(node.get("domain") or "").strip()
+        method = (node.get("methods") or {}).get("http") or {}
+        port = method.get("port", node.get("port"))
+        try:
+            port = int(port)
+        except (TypeError, ValueError):
+            port = 0
+        if not domain:
+            raise ValueError(f"第 {index} 项缺少 domain")
+        if not 1 <= port <= 65535:
+            raise ValueError(f"第 {index} 项的 HTTP 代理端口无效")
+        result.append(dict(node))
+    return result
+
+
+def mask_url(url: str) -> str:
+    """用于 UI 展示，隐藏查询参数和 URL 用户信息中的敏感令牌。"""
+    try:
+        parts = urlsplit(str(url))
+        host = parts.hostname or ""
+        if parts.port:
+            host = f"{host}:{parts.port}"
+        if parts.username:
+            host = "***@" + host
+        return urlunsplit((parts.scheme, host, parts.path, "…" if parts.query else "", ""))
+    except Exception:
+        return "已隐藏订阅地址"

lite/rrpp_core/traffic.py

@@ -0,0 +1,55 @@
+"""线程安全的本地转发流量统计。"""
+
+import threading
+import time
+
+
+class TrafficStats:
+    """记录单次代理会话的上下行总量与采样速率。"""
+
+    def __init__(self, clock=time.time):
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._up = 0
+        self._down = 0
+        self._prev_up = 0
+        self._prev_down = 0
+        self._prev_ts = 0.0
+
+    def add(self, up=0, down=0):
+        with self._lock:
+            self._up += up
+            self._down += down
+
+    def reset(self):
+        with self._lock:
+            self._up = self._down = 0
+            self._prev_up = self._prev_down = 0
+            self._prev_ts = self._clock()
+
+    def rate(self):
+        with self._lock:
+            now = self._clock()
+            if self._prev_ts > 0 and now > self._prev_ts:
+                elapsed = now - self._prev_ts
+                up_rate = int((self._up - self._prev_up) / elapsed)
+                down_rate = int((self._down - self._prev_down) / elapsed)
+            else:
+                up_rate = down_rate = 0
+            self._prev_up = self._up
+            self._prev_down = self._down
+            self._prev_ts = now
+            return up_rate, down_rate, self._up, self._down
+
+    def snapshot(self):
+        with self._lock:
+            return self._up, self._down
+
+
+def format_bytes(value):
+    """将字节数格式化为 UI 展示文本。"""
+    number = float(value)
+    for unit in ("B", "KB", "MB", "GB", "TB"):
+        if number < 1024 or unit == "TB":
+            return f"{int(number)} {unit}" if unit == "B" else f"{number:.1f} {unit}"
+        number /= 1024

lite/rrpp_ui/__init__.py

@@ -0,0 +1,4 @@
+"""RRPP Lite 的 Qt 界面层。
+
+核心代理与 PAC 逻辑不得依赖本包，以便后续在无 GUI 环境中独立测试。
+"""

lite/rrpp_ui/theme.py

@@ -0,0 +1,11 @@
+"""RRPP Lite 界面的共享色彩常量。"""
+
+C_TITLE = "#8899aa"
+C_GREEN = "#51cf66"
+C_YELLOW = "#ffd43b"
+C_RED = "#ff6b6b"
+C_BG = "#ffffff"
+C_CARD = "#f0f2f5"
+C_BORDER = "#d0d5dd"
+C_TEXT = "#333333"
+C_SUBTEXT = "#889099"

lite/rrpp_ui/widgets.py

@@ -0,0 +1,104 @@
+"""无业务状态的可复用 Qt 控件。"""
+
+from PyQt5.QtCore import Qt, QRectF, pyqtSignal
+from PyQt5.QtGui import QColor, QPainter
+from PyQt5.QtWidgets import QCheckBox, QLabel
+
+from .theme import C_BORDER, C_GREEN, C_RED, C_TITLE
+
+
+_NL_BASE = f"color: {C_TITLE}; font-size: 12px;"
+
+
+class NodeLabel(QLabel):
+    """可点击标签：hover 显示下划线和手型光标，点击发出 clicked 信号。"""
+
+    clicked = pyqtSignal()
+
+    def __init__(self, text="", tooltip="点击切换节点", base=None, hover=None):
+        super().__init__(text)
+        self._base = base or _NL_BASE
+        self._hover = hover or self._base + " text-decoration: underline;"
+        self.setStyleSheet(self._base)
+        if tooltip:
+            self.setToolTip(tooltip)
+
+    def enterEvent(self, event):
+        self.setStyleSheet(self._hover)
+        self.setCursor(Qt.PointingHandCursor)
+        super().enterEvent(event)
+
+    def leaveEvent(self, event):
+        self.setStyleSheet(self._base)
+        self.unsetCursor()
+        super().leaveEvent(event)
+
+    def mouseReleaseEvent(self, event):
+        if event.button() == Qt.LeftButton:
+            self.clicked.emit()
+        super().mouseReleaseEvent(event)
+
+
+class ToggleSwitch(QCheckBox):
+    """苹果风格开关：绿/灰胶囊加滑动圆钮，自绘且不依赖图片资源。"""
+
+    def __init__(self, checked=False, parent=None):
+        super().__init__(parent)
+        self.setChecked(checked)
+        self.setFixedSize(42, 22)
+        self.setCursor(Qt.PointingHandCursor)
+
+    def paintEvent(self, event):
+        painter = QPainter(self)
+        painter.setRenderHint(QPainter.Antialiasing)
+        painter.setPen(Qt.NoPen)
+        painter.setBrush(QColor(C_GREEN) if self.isChecked() else QColor("#c8ccd4"))
+        painter.drawRoundedRect(QRectF(0, 0, 42, 22), 11, 11)
+        knob_x = 20.0 if self.isChecked() else 2.0
+        painter.setBrush(QColor("#ffffff"))
+        painter.drawEllipse(QRectF(knob_x, 2.0, 18.0, 18.0))
+        painter.end()
+
+    def mousePressEvent(self, event):
+        if event.button() == Qt.LeftButton:
+            self.setChecked(not self.isChecked())
+            self.repaint()
+            event.accept()
+            return
+        super().mousePressEvent(event)
+
+
+class StatusDot(QLabel):
+    """代理状态小圆点。"""
+
+    def __init__(self):
+        super().__init__()
+        self.setFixedSize(14, 14)
+        self.set_off()
+
+    def set_on(self):
+        self.setStyleSheet(f"""
+            QLabel {{
+                background-color: {C_GREEN};
+                border-radius: 7px;
+                border: 1px solid rgba(81, 207, 102, 50);
+            }}
+        """)
+
+    def set_off(self):
+        self.setStyleSheet(f"""
+            QLabel {{
+                background-color: {C_RED};
+                border-radius: 7px;
+                border: 1px solid rgba(255, 107, 107, 50);
+            }}
+        """)
+
+    def set_disabled(self):
+        self.setStyleSheet(f"""
+            QLabel {{
+                background-color: {C_BORDER};
+                border-radius: 7px;
+                border: 1px solid rgba(136, 153, 170, 50);
+            }}
+        """)

lite/test/check_node_menu.py

@@ -1,104 +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 proxy_lite as pp
-from PyQt5.QtWidgets import QApplication
-
-# 构造窗口时不写用户数据目录（避免污染），构造后再指向真实 nodes.json
-pp._user_init_local = lambda: None
-
-app = QApplication(sys.argv)
-w = pp.MainWindow()
-w.timer.stop()                      # 停掉状态轮询，纯静态检查
-
-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()} (应不变)")
-
-# ── 上次选中的节点已不可用：启动时自动回退到第一个可达节点 ──────────
-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]
-    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'}")
-    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'}")
-    # 再校验一次：已经是可达节点，不应再切换
-    ok3 = pp._verify_selected_node() is False
-    print(f"  再次校验不重复切换 {'PASS' if ok3 else 'FAIL'}")
-    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'}")
-    # 已是正确域名时再跑一次：不应重复切换
-    ok2 = pp._rematch_saved_domain() is False
-    print(f"  再次重匹配不重复切换 {'PASS' if ok2 else 'FAIL'}")
-    # 域名已不在列表（节点被移除）：不切换，保持现状
-    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'}")
-    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)
-
-print("OK")

lite/test/test_pac_routing.py

@@ -1,738 +0,0 @@
-# -*- coding: utf-8 -*-
-"""PAC（智能分流）模式路由决策测试。
-
-核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？
-
-判定链（proxy_lite._decide_direct）：
-  0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效）
-  1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定）
-  2. 内网 IP 且勾选 bypass_lan -> 直连
-  3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道
-
-直接 import proxy_lite 调用真实决策函数，数据用真实的 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 proxy_lite as pp
-from proxy_lite 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 时必须判定为不可达，否则必须可达
-        if io:
-            expect = has_v6 or reach  # 本机有 IPv6 -> 应可达；无 IPv6 -> 不应可达
-            ok = reach == expect
-            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)