Merge remote-tracking branch 'origin/main'

仓库内 fake-site/index.html 为旧版（md5 e0f5cbcb），与四个生产节点实际
服务的版本不一致。已用线上（us 节点）实际抓取版本替换，恢复所有节点

README.md

@@ -1,11 +1,17 @@
+# Royirin Rosmontis Private Proxy
+Supported by Global Rosmanka Cloud Nexus Inc. —— https://rosmanka.rrrosmontis.icu
 ## 警告
 - 包含 WireGuard 的 common 版本转发器永久不再维护，推荐使用 Lite 版本
 > common 版本现在有许多未修复的 bug (如部分 UI 的显示问题) 、令人难以忍受的代码逻辑 (如 PAC 规则直接写在代码里)和安全性问题 (认证凭据的暴露与证书忽略)，而 Lite 版本修复了这些问题，并会收到持续的更新 
-- 本程序完全没有 Linux 与 CLI 功能兼容，默认构建为带有图形界面的 Windows .exe
+- 本程序完全没有 Linux 与 CLI 功能兼容，默认构建为带有图形界面的 Windows 应用 (.exe)
 
-
-## 如何区分 common 版本与 Lite 版本
+### 如何区分 common 版本与 Lite 版本
 - Lite 版本的版本号前四位固定为 9999，后续为从月份开始的构建时间戳，而 common 版本的版本号为包含年份的完整的构建时间戳
 
 ## 服务器侧配置手册 (部分内容过时)
-https://rrrosmontis.icu/articles/index.html?id=00005
+- https://rrrosmontis.icu/articles/index.html?id=00005
+- 更推荐参考本仓库 server_config 文件夹中的 md 文件手册
+
+## 若要直接使用仓库客户端对应的服务
+请前往 https://sso.rrrosmontis.icu 注册 Rosmanka ID，然后与管理员商定使用细则
+royirin@rrrosmontis.icu

lite/test/check_node_menu.py

@@ -72,4 +72,33 @@ else:
     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

@@ -0,0 +1,738 @@
+# -*- 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)

server_config/fake-site/index.html

@@ -26,7 +26,7 @@
                 <span class="code-label">Error code 200</span>
             </h1>
             <div>
-                Visit <a href="https://example.com" target="_blank" rel="noopener noreferrer">example.com</a> for more information.
+                Visit <a href="https://rrrosmontis.icu" target="_blank" rel="noopener noreferrer">rrrosmontis.icu</a> for more information.
             </div>
             <div class="mt-3">2026-06-28 05:00:19 UTC</div>
         </header>
@@ -90,7 +90,7 @@
                     <span class="cf-footer-separator sm:hidden">&bull;</span>
                 </span>
                 
-                <span class="cf-footer-item sm:block sm:mb-1"><span>Performance &amp; security by</span> <a rel="noopener noreferrer" href="https://example.com" id="brand_link" target="_blank">Example Site</a></span>
+                <span class="cf-footer-item sm:block sm:mb-1"><span>Performance &amp; security by</span> <a rel="noopener noreferrer" href="https://rrrosmontis.icu" id="brand_link" target="_blank">Royirin Rosmontis</a></span>
             </p>
         </div><!-- /.error-footer -->
     </div>