修复：完善 PAC 分流与启动取消

2026-09-16 19:10:12 +08:00
parent 9376e35025
commit 0603224de2
2 changed files with 846 additions and 167 deletions
@@ -14,6 +14,7 @@ pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。
 """

 import sys
+import threading
 from pathlib import Path

 TEST_DIR = Path(__file__).resolve().parent
@@ -360,6 +361,351 @@ 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 兼容 + 缓存命中")
@@ -387,3 +733,6 @@ print(f"  cache_clear 后重查: {pp._is_china_ip('114.114.114.114')} "
 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)