Merge remote-tracking branch 'origin/main'

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/proxy_lite.py

@@ -602,6 +602,52 @@ def proxy_off():
     total = up + down
     log(f"本次转发流量共 {_fmt_bytes(total)}")
 
+
+class _StartCancelled(RuntimeError):
+    """用户在启动连通性探测完成前取消了本次开启操作。"""
+
+
+def _raise_if_start_cancelled(cancel_event):
+    if cancel_event is not None and cancel_event.is_set():
+        raise _StartCancelled("用户取消了开启代理")
+
+
+def _verify_proxy_connectivity(timeout=10, opener=None, cancel_event=None):
+    """验证本地代理的国内直连和境外隧道路径。
+
+    国内探测只作诊断；境外 google.com 探测是启动成功的必要条件，避免认证、
+    隧道或节点故障时仍把 UI 标记为“已开启”。返回每条路径的可达性。
+    """
+    import urllib.request
+    if opener is None:
+        local_proxy = f"http://{PROXY_HOST}:{PROXY_PORT}"
+        opener = urllib.request.build_opener(
+            urllib.request.ProxyHandler({"http": local_proxy, "https": local_proxy}))
+    results = {}
+    test_urls = [
+        ("baidu.com", "http://www.baidu.com", False),
+        ("google.com", "https://www.google.com/generate_204", True),
+    ]
+    for name, url, required in test_urls:
+        _raise_if_start_cancelled(cancel_event)
+        try:
+            with opener.open(url, timeout=timeout) as response:
+                response.read(1)
+            _raise_if_start_cancelled(cancel_event)
+            results[name] = True
+            log(f"  {name} 可达")
+        except _StartCancelled:
+            raise
+        except Exception as e:
+            # 取消时，本地转发器可能已被关闭，从而让正在进行的请求报错；
+            # 这不是“开启失败”，也不能在稍后把界面重新切成已开启。
+            _raise_if_start_cancelled(cancel_event)
+            results[name] = False
+            log(f"  {name} 不可达: {e}")
+            if required:
+                raise RuntimeError(f"境外隧道验证失败（{name}）：{e}") from e
+    return results
+
 # ── 本地认证转发器 ─────────────────────────────────────
 
 # ── 流量统计 ──────────────────────────────────────────
@@ -797,7 +843,7 @@ def _ttl_cache(seconds: int, maxsize: int = 16384):
 
 @_ttl_cache(300, maxsize=2048)
 def _is_private_ip(host: str) -> bool:
-    """判断目标是否命中常用内网/回环 IPv4；域名会解析后判断。"""
+    """判断目标是否命中常用内网、回环或链路本地地址（IPv4/IPv6）。"""
     try:
         import ipaddress
         ip = ipaddress.ip_address(host)
@@ -806,7 +852,7 @@ def _is_private_ip(host: str) -> bool:
         try:
             import socket
             addresses = []
-            for addr in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM):
+            for addr in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM):
                 try:
                     addresses.append(ipaddress.ip_address(addr[4][0]))
                 except ValueError:
@@ -814,27 +860,27 @@ def _is_private_ip(host: str) -> bool:
         except Exception:
             return False
     for ip in addresses:
-        if ip.version != 4:
+        if ip.version == 4 and _is_private_int(int(ip)):
-            continue
+            return True
-        n = int(ip)
+        if ip.version == 6 and (ip.is_loopback or ip.is_link_local
-        if (
+                                or ip in ipaddress.ip_network("fc00::/7")):
-            (n & 0xFF000000) == 0x0A000000 or
-            (n & 0xFFF00000) == 0xAC100000 or
-            (n & 0xFFFF0000) == 0xC0A80000 or
-            (n & 0xFF000000) == 0x7F000000 or
-            (n & 0xFFFF0000) == 0xA9FE0000
-        ):
             return True
     return False
 
 # ── IP 智能分流（域名表优先，非中国 IP 走隧道，中国 IP 直连） ─────
 
 _CN_RANGES = None           # 合并排序后的 (start, end) 整数段
+_CN_RANGES_V6 = None        # 合并排序后的中国 IPv6 整数段（数据源提供时启用）
 _CN_RANGES_LOADED = False
 _FORCE_PROXY_SUFFIXES = None
 _FORCE_PROXY_EXACT = None
 _FORCE_PROXY_REGEXES = None
 _FORCE_PROXY_LOADED = False
+_DIRECT_SUFFIX_TRIE = None
+_DIRECT_EXACT = None
+_DIRECT_REGEXES = None
+_DIRECT_LOADED = False
+_PAC_DATA_LOCK = threading.RLock()
 
 # PAC 数据由 CDN 独立下发；此处保留登录/AI 服务的关键后缀兜底。
 # 即使远端表暂时未更新或不可读，这些域名也不能因解析到中国 CDN IP 而直连。
@@ -1031,10 +1077,11 @@ _TLS_LOCAL_PORT = _port_val()  # 本地转发器监听端口（设置 > 内置 c
 PROXY_PORT = _TLS_LOCAL_PORT
 
 # ── PAC 数据在线更新（CDN 直连，无需认证） ─────────────
-# 更新源地址取 config.json 的 pac_base_url，目录下直接放两个数据文件:
+# 更新源地址取 config.json 的 pac_base_url，目录下直接放三个数据文件:
 #   {pac_base_url}/pac_domains.json    强制隧道域名表（version/date 字段随文件下发）
+#   {pac_base_url}/pac_direct_domains.json  国内/用户指定直连域名表（同上）
 #   {pac_base_url}/pac_ip_cn.json      中国 IP 段（同上）
-_PAC_FILES = ("pac_domains", "pac_ip_cn")
+_PAC_FILES = ("pac_domains", "pac_direct_domains", "pac_ip_cn")
 
 def _pac_read_version(path):
     """读取 PAC 数据文件内的 version/date 字段。"""
@@ -1062,7 +1109,7 @@ def _pac_fetch_json(url, timeout=15, tries=2):
                 raise
 
 def _pac_validate(name, data):
-    if name == "pac_domains":
+    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)
@@ -1076,6 +1123,31 @@ def _pac_write(name, data):
     tmp.write_text(json.dumps(data, ensure_ascii=False), "utf-8")
     os.replace(tmp, _USER_DIR / f"{name}.json")
 
+
+def _invalidate_pac_routing_cache(name=None):
+    """清除 PAC 数据更新后的进程内缓存，使新规则无需重启立即生效。"""
+    global _CN_RANGES, _CN_RANGES_V6, _CN_RANGES_LOADED
+    global _FORCE_PROXY_SUFFIXES, _FORCE_PROXY_EXACT, _FORCE_PROXY_REGEXES, _FORCE_PROXY_LOADED
+    global _DIRECT_SUFFIX_TRIE, _DIRECT_EXACT, _DIRECT_REGEXES, _DIRECT_LOADED
+    with _PAC_DATA_LOCK:
+        if name in (None, "pac_ip_cn"):
+            _CN_RANGES = None
+            _CN_RANGES_V6 = None
+            _CN_RANGES_LOADED = False
+            _is_china_ip.cache_clear()
+        if name in (None, "pac_domains"):
+            _FORCE_PROXY_SUFFIXES = None
+            _FORCE_PROXY_EXACT = None
+            _FORCE_PROXY_REGEXES = None
+            _FORCE_PROXY_LOADED = False
+            _is_force_proxy.cache_clear()
+        if name in (None, "pac_direct_domains"):
+            _DIRECT_SUFFIX_TRIE = None
+            _DIRECT_EXACT = None
+            _DIRECT_REGEXES = None
+            _DIRECT_LOADED = False
+            _is_direct_domain.cache_clear()
+
 # ── nodes.json 加密存储（Windows DPAPI，本机本用户可解） ─────────────
 # 非 Windows（dev/测试）不加密，读写明文；读取对密文/明文都兼容。
 
@@ -1275,7 +1347,9 @@ def _user_init_remote(base_url=None):
                 if rver:
                     if rver > lver:
                         _pac_write(name, data)
-                        log(f"PAC 数据已更新: {name}.json {rver} ({rdate})")
+                        _invalidate_pac_routing_cache(name)
+                        if name != "pac_direct_domains":
+                            log(f"PAC 数据已更新: {name}.json {rver} ({rdate})")
                         updated += 1
                     # 否则已是最新，跳过
                 else:
@@ -1284,7 +1358,9 @@ def _user_init_remote(base_url=None):
                     if local.is_file() and local.read_bytes() == new_bytes:
                         continue
                     _pac_write(name, data)
-                    log(f"PAC 数据已下载: {name}.json 无版本号 ({rdate})")
+                    _invalidate_pac_routing_cache(name)
+                    if name != "pac_direct_domains":
+                        log(f"PAC 数据已下载: {name}.json 无版本号 ({rdate})")
                     updated += 1
             except Exception as e:
                 log(f"PAC 数据 {name}.json 更新失败，使用本地副本: {e}")
@@ -1292,39 +1368,50 @@ def _user_init_remote(base_url=None):
         if updated == 0 and failed == 0 and (_USER_DIR / "pac_domains.json").is_file():
             log("PAC 数据已是最新")
 
-def _load_cn_ranges():
+def _merge_ip_ranges(cidrs, version):
-    """加载 pac_ip_cn.json 中的中国 IPv4 段，合并为整数区间供二分查找。"""
+    """将指定 IP 版本的 CIDR 合并为整数闭区间，供二分查找。"""
-    global _CN_RANGES, _CN_RANGES_LOADED
+    import ipaddress
-    if _CN_RANGES_LOADED:
+    ranges = []
-        return _CN_RANGES
+    for c in cidrs:
-    _CN_RANGES_LOADED = True
+        net = ipaddress.ip_network(str(c).strip(), strict=False)
-    try:
+        if net.version == version:
-        import json, ipaddress
-        path = _user_path("pac_ip_cn.json")
-        if path is None:
-            log_red("未找到 pac_ip_cn.json，智能分流不可用（全部走隧道）")
-            return _CN_RANGES
-        data = json.loads(Path(path).read_text("utf-8"))
-        cidrs = data.get("cidrs") or []
-        ranges = []
-        for c in cidrs:
-            net = ipaddress.ip_network(str(c).strip(), strict=False)
-            if net.version != 4:
-                continue
             ranges.append((int(net.network_address), int(net.broadcast_address)))
-        ranges.sort()
+    ranges.sort()
-        merged = []
+    merged = []
-        for s, e in ranges:
+    for start, end in ranges:
-            if merged and s <= merged[-1][1] + 1:
+        if merged and start <= merged[-1][1] + 1:
-                if e > merged[-1][1]:
+            if end > merged[-1][1]:
-                    merged[-1] = (merged[-1][0], e)
+                merged[-1] = (merged[-1][0], end)
-            else:
+        else:
-                merged.append((s, e))
+            merged.append((start, end))
-        _CN_RANGES = merged
+    return merged
-        log(f"已加载中国 IP 段: {len(cidrs)} 条 CIDR，合并为 {len(merged)} 段")
+
-    except Exception as e:
+
-        log_red(f"pac_ip_cn.json 加载失败: {e}")
+def _load_cn_ranges():
-    return _CN_RANGES
+    """加载 pac_ip_cn.json 的中国 IPv4/IPv6 CIDR，合并为整数区间。"""
+    global _CN_RANGES, _CN_RANGES_V6, _CN_RANGES_LOADED
+    with _PAC_DATA_LOCK:
+        if _CN_RANGES_LOADED:
+            return _CN_RANGES
+        try:
+            import json
+            path = _user_path("pac_ip_cn.json")
+            if path is None:
+                log_red("未找到 pac_ip_cn.json，智能分流不可用（全部走隧道）")
+                return _CN_RANGES
+            data = json.loads(Path(path).read_text("utf-8"))
+            cidrs = data.get("cidrs") or []
+            _CN_RANGES = _merge_ip_ranges(cidrs, 4)
+            _CN_RANGES_V6 = _merge_ip_ranges(cidrs, 6)
+            v4_count = sum(1 for c in cidrs if ":" not in str(c))
+            v6_count = len(cidrs) - v4_count
+            log(f"已加载中国 IP 段: IPv4 {v4_count} 条/合并 {len(_CN_RANGES)} 段，"
+                f"IPv6 {v6_count} 条/合并 {len(_CN_RANGES_V6)} 段")
+        except Exception as e:
+            log_red(f"pac_ip_cn.json 加载失败: {e}")
+        finally:
+            _CN_RANGES_LOADED = True
+        return _CN_RANGES
 
 def _load_force_proxy_suffixes():
     """加载 pac_domains.json 的后缀、精确和正则规则。
@@ -1334,59 +1421,136 @@ def _load_force_proxy_suffixes():
     当成带冒号的普通后缀，导致它们永远不命中。
     """
     global _FORCE_PROXY_SUFFIXES, _FORCE_PROXY_EXACT, _FORCE_PROXY_REGEXES, _FORCE_PROXY_LOADED
-    if _FORCE_PROXY_LOADED:
+    with _PAC_DATA_LOCK:
-        return _FORCE_PROXY_SUFFIXES
+        if _FORCE_PROXY_LOADED:
-    _FORCE_PROXY_LOADED = True
+            return _FORCE_PROXY_SUFFIXES
-    try:
+        try:
-        import json
+            import json
-        path = _user_path("pac_domains.json")
+            path = _user_path("pac_domains.json")
-        if path is None:
+            if path is None:
+                _FORCE_PROXY_SUFFIXES = sorted(_CORE_FORCE_PROXY_SUFFIXES)
+                _FORCE_PROXY_EXACT = frozenset()
+                _FORCE_PROXY_REGEXES = ()
+                log(f"未找到 pac_domains.json，使用 {len(_FORCE_PROXY_SUFFIXES)} 条关键域名兜底")
+                return _FORCE_PROXY_SUFFIXES
+            data = json.loads(Path(path).read_text("utf-8"))
+            suffixes = set(_CORE_FORCE_PROXY_SUFFIXES)
+            exact = set()
+            regexes = []
+            invalid_regexes = 0
+            for raw in data.get("domains") 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。
+                    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
+                suffixes.add(rule.lstrip("*.").rstrip("."))
+            suffixes.discard("")
+            _FORCE_PROXY_SUFFIXES = suffixes
+            _FORCE_PROXY_EXACT = frozenset(exact)
+            _FORCE_PROXY_REGEXES = tuple(regexes)
+            log(f"已加载强制代理规则: 后缀 {len(suffixes)} 条、精确 {len(exact)} 条、"
+                f"正则 {len(regexes)} 条 (pac_domains.json)" +
+                (f"，跳过 {invalid_regexes} 条非法正则" if invalid_regexes else ""))
+        except Exception as e:
+            log_red(f"pac_domains.json 加载失败: {e}")
             _FORCE_PROXY_SUFFIXES = sorted(_CORE_FORCE_PROXY_SUFFIXES)
             _FORCE_PROXY_EXACT = frozenset()
             _FORCE_PROXY_REGEXES = ()
-            log(f"未找到 pac_domains.json，使用 {len(_FORCE_PROXY_SUFFIXES)} 条关键域名兜底")
+        finally:
-            return _FORCE_PROXY_SUFFIXES
+            _FORCE_PROXY_LOADED = True
-        data = json.loads(Path(path).read_text("utf-8"))
+        return _FORCE_PROXY_SUFFIXES
-        suffixes = set(_CORE_FORCE_PROXY_SUFFIXES)
+
-        exact = set()
+
-        regexes = []
+def _build_suffix_trie(suffixes):
-        invalid_regexes = 0
+    """按反向域名标签建立 Trie，适合大规模后缀规则的 O(标签数) 匹配。"""
-        for raw in data.get("domains") or []:
+    root = {}
-            raw_rule = str(raw).strip()
+    for suffix in suffixes:
-            rule = raw_rule.lower()
+        node = root
-            if not raw_rule:
+        for label in reversed(suffix.split(".")):
-                continue
+            node = node.setdefault(label, {})
-            if rule.startswith("full:"):
+        node[""] = True
-                host = rule[len("full:"):].rstrip(".")
+    return root
-                if host:
+
-                    exact.add(host)
+
-                continue
+def _suffix_trie_matches(host: str, trie) -> bool:
-            if rule.startswith("regexp:"):
+    node = trie
-                # 正则本身可能含大写字符类或大小写敏感字面量；只忽略匹配时的
+    for label in reversed(host.rstrip(".").lower().split(".")):
-                # 域名大小写，不能在编译前改写 pattern。
+        node = node.get(label)
-                pattern = raw_rule[len("regexp:"):]
+        if node is None:
-                if not pattern:
+            return False
-                    invalid_regexes += 1
+        if node.get(""):
+            return True
+    return bool(node and node.get(""))
+
+
+def _load_direct_domains():
+    """加载 pac_direct_domains.json：PAC 智能分流中明确直连的域名规则。"""
+    global _DIRECT_SUFFIX_TRIE, _DIRECT_EXACT, _DIRECT_REGEXES, _DIRECT_LOADED
+    with _PAC_DATA_LOCK:
+        if _DIRECT_LOADED:
+            return _DIRECT_SUFFIX_TRIE
+        try:
+            import json
+            path = _user_path("pac_direct_domains.json")
+            if path is None:
+                _DIRECT_SUFFIX_TRIE, _DIRECT_EXACT, _DIRECT_REGEXES = {}, frozenset(), ()
+                return _DIRECT_SUFFIX_TRIE
+            data = json.loads(Path(path).read_text("utf-8"))
+            suffixes, exact, regexes = set(), set(), []
+            invalid_regexes = 0
+            for raw in data.get("domains") or []:
+                raw_rule = str(raw).strip()
+                rule = raw_rule.lower()
+                if not raw_rule:
                     continue
-                try:
+                if rule.startswith("full:"):
-                    regexes.append(re.compile(pattern, re.IGNORECASE))
+                    host = rule[len("full:"):].rstrip(".")
-                except re.error:
+                    if host:
-                    invalid_regexes += 1
+                        exact.add(host)
-                continue
+                    continue
-            suffixes.add(rule.lstrip("*.").rstrip("."))
+                if rule.startswith("regexp:"):
-        suffixes.discard("")
+                    pattern = raw_rule[len("regexp:"):]
-        _FORCE_PROXY_SUFFIXES = suffixes
+                    if not pattern:
-        _FORCE_PROXY_EXACT = frozenset(exact)
+                        invalid_regexes += 1
-        _FORCE_PROXY_REGEXES = tuple(regexes)
+                        continue
-        log(f"已加载强制代理规则: 后缀 {len(suffixes)} 条、精确 {len(exact)} 条、"
+                    try:
-            f"正则 {len(regexes)} 条 (pac_domains.json)" +
+                        regexes.append(re.compile(pattern, re.IGNORECASE))
-            (f"，跳过 {invalid_regexes} 条非法正则" if invalid_regexes else ""))
+                    except re.error:
-    except Exception as e:
+                        invalid_regexes += 1
-        log_red(f"pac_domains.json 加载失败: {e}")
+                    continue
-        _FORCE_PROXY_SUFFIXES = sorted(_CORE_FORCE_PROXY_SUFFIXES)
+                suffix = rule.lstrip("*.").rstrip(".")
-        _FORCE_PROXY_EXACT = frozenset()
+                if suffix:
-        _FORCE_PROXY_REGEXES = ()
+                    suffixes.add(suffix)
-    return _FORCE_PROXY_SUFFIXES
+            _DIRECT_SUFFIX_TRIE = _build_suffix_trie(suffixes)
+            _DIRECT_EXACT = frozenset(exact)
+            _DIRECT_REGEXES = tuple(regexes)
+            log(f"已加载明确直连规则: 后缀 {len(suffixes)} 条、精确 {len(exact)} 条、"
+                f"正则 {len(regexes)} 条 (pac_direct_domains.json)" +
+                (f"，跳过 {invalid_regexes} 条非法正则" if invalid_regexes else ""))
+        except Exception as e:
+            log_red(f"pac_direct_domains.json 加载失败: {e}")
+            _DIRECT_SUFFIX_TRIE, _DIRECT_EXACT, _DIRECT_REGEXES = {}, frozenset(), ()
+        finally:
+            _DIRECT_LOADED = True
+        return _DIRECT_SUFFIX_TRIE
+
 
 def _is_private_int(n: int) -> bool:
     """整数形式的 IPv4 是否属于常用内网/回环段。"""
@@ -1397,11 +1561,13 @@ def _is_private_int(n: int) -> bool:
             (n & 0xFFFF0000) == 0xA9FE0000)
 
 @_ttl_cache(300, maxsize=16384)
-def _is_china_ip(host: str):
+def _resolve_china_direct_ip(host: str):
-    """判断目标是否命中中国 IP 段；域名会先解析。
+    """返回可安全直连的中国 IPv4/IPv6，或 None。
 
-    任一解析结果命中中国段 -> True（直连）；全部不命中 -> False（走隧道）；
+    同一地址族的所有解析结果都必须在对应中国 IP 段内，才允许直连；混合
-    解析失败或数据未加载 -> None（无法判定，安全起见走隧道）。
+    CDN 记录不会因其中一个中国 IP 而误判。优先 IPv4 保持原有兼容性；
+    数据源提供 IPv6 CIDR 后，IPv6-only 目标同样可直连。返回本次解析得到
+    的地址，让转发器直接连接它，避免判定后再次 DNS 解析到不同的地址。
     """
     import ipaddress
     try:
@@ -1411,7 +1577,7 @@ def _is_china_ip(host: str):
         try:
             import socket
             addresses = []
-            for addr in socket.getaddrinfo(host, None, socket.AF_INET, socket.SOCK_STREAM):
+            for addr in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM):
                 try:
                     addresses.append(ipaddress.ip_address(addr[4][0]))
                 except ValueError:
@@ -1420,20 +1586,37 @@ def _is_china_ip(host: str):
             return None
     if not addresses:
         return None
-    ranges = _load_cn_ranges()
+    _load_cn_ranges()
-    if not ranges:
+
-        return None
+    def family_target(version, ranges):
-    for ip in addresses:
+        if not ranges:
-        if ip.version != 4:
+            return None
-            continue
+        family = [ip for ip in addresses if ip.version == version]
-        n = int(ip)
+        if not family or any(not ip.is_global for ip in family):
-        # 0.0.0.0 / 内网 / 回环解析结果疑似污染或局域网，不参与判定
+            return None
-        if n == 0 or _is_private_int(n):
+        limit = (1 << (32 if version == 4 else 128)) - 1
-            continue
+        for ip in family:
-        i = bisect.bisect_right(ranges, (n, 0xFFFFFFFF))
+            n = int(ip)
-        if i > 0 and ranges[i - 1][0] <= n <= ranges[i - 1][1]:
+            i = bisect.bisect_right(ranges, (n, limit))
-            return True
+            if not (i > 0 and ranges[i - 1][0] <= n <= ranges[i - 1][1]):
-    return False
+                return None
+        return str(family[0])
+
+    # IPv4 仍优先；IPv6-only 目标会在 PAC 数据加入 IPv6 CIDR 后走下一分支。
+    return family_target(4, _CN_RANGES) or family_target(6, _CN_RANGES_V6)
+
+
+@_ttl_cache(300, maxsize=16384)
+def _is_china_ip(host: str):
+    """目标是否可按中国 IPv4/IPv6 直连；无法安全判定时返回 False。"""
+    return _resolve_china_direct_ip(host) is not None
+
+
+_is_china_ip_cache_clear = _is_china_ip.cache_clear
+def _clear_china_ip_cache():
+    _is_china_ip_cache_clear()
+    _resolve_china_direct_ip.cache_clear()
+_is_china_ip.cache_clear = _clear_china_ip_cache
 
 @functools.lru_cache(maxsize=8192)
 def _is_force_proxy(host: str) -> bool:
@@ -1448,6 +1631,19 @@ def _is_force_proxy(host: str) -> bool:
             h in exact or
             any(pattern.search(h) is not None for pattern in regexes))
 
+
+@functools.lru_cache(maxsize=32768)
+def _is_direct_domain(host: str) -> bool:
+    """PAC 明确直连规则命中的域名直连（后缀 / full / regexp）。"""
+    trie = _load_direct_domains()
+    exact = _DIRECT_EXACT or ()
+    regexes = _DIRECT_REGEXES or ()
+    if not trie and not exact and not regexes:
+        return False
+    h = host.rstrip(".").lower()
+    return (_suffix_trie_matches(h, trie) or h in exact or
+            any(pattern.search(h) is not None for pattern in regexes))
+
 def _resolve_host_ips(host: str):
     """解析域名的 IPv4/IPv6 地址集合（字面量 IP 返回自身）；解析失败返回空集合。"""
     import ipaddress
@@ -1491,20 +1687,46 @@ def _is_whitelisted(host: str, version: int) -> bool:
 
 def _decide_direct(host: str, bypass_lan: bool, smart: bool) -> bool:
     """判断连接是否直连。判定顺序（白名单最优先）：
-    白名单命中 -> 直连；域名表命中 -> 隧道；内网 + bypass_lan -> 直连；
+    白名单命中 -> 直连；强制代理域名 -> 隧道；明确直连域名 -> 直连；
-    中国 IP -> 直连，其余 -> 隧道。
+    内网 + bypass_lan -> 直连；中国 IP -> 直连；其余 -> 隧道。
     白名单在 PAC 与固定代理模式下均生效（固定模式除白名单外全部走隧道）。
     """
     if _is_whitelisted(host, _WHITELIST_VERSION):
         return True
     if smart and _is_force_proxy(host):
         return False
+    if smart and _is_direct_domain(host):
+        return True
     if bypass_lan and _is_private_ip(host):
         return True
     if smart:
         return _is_china_ip(host) is True
     return False
 
+
+def _direct_connect_target(host: str, port: int, bypass_lan: bool, smart: bool):
+    """返回直连 socket 目标，或 None 表示交给上游隧道。
+
+    仅智能分流判定出的中国公网地址固定为本次 DNS 解析的 IP；白名单和
+    内网直连保持原始主机名，保留用户显式规则与 IPv6 的原有语义。
+    """
+    if not _decide_direct(host, bypass_lan, smart):
+        return None
+    if _is_smart_china_direct(host, bypass_lan, smart):
+        ip = _resolve_china_direct_ip(host)
+        if ip is None:  # 缓存过期或 DNS 状态变化时，安全起见改走隧道。
+            return None
+        return ip, port
+    return host, port
+
+
+def _is_smart_china_direct(host: str, bypass_lan: bool, smart: bool) -> bool:
+    """当前直连是否来自智能分流的中国公网判定，而非用户显式直连规则。"""
+    return (smart and not _is_whitelisted(host, _WHITELIST_VERSION)
+            and not _is_direct_domain(host)
+            and not (bypass_lan and _is_private_ip(host))
+            and _resolve_china_direct_ip(host) is not None)
+
 def _parse_connect_authority(first_line: str):
     """从 CONNECT 首行提取 (host, port)，失败返回 None。
 
@@ -1565,6 +1787,24 @@ def _force_http_close(buf: bytes) -> bytes:
         out.append(b"Connection: close")
     return b"\r\n".join(out) + tail
 
+def _connect_route(direct_target, allow_direct_fallback, remote_host, remote_port, ctx, use_tls,
+                   create_connection=None):
+    """连接直连目标或上游隧道；智能中国直连失败时安全回退一次隧道。"""
+    import socket
+    create = create_connection or socket.create_connection
+    if direct_target:
+        try:
+            return create(direct_target, 10), False
+        except OSError as e:
+            if not allow_direct_fallback:
+                raise
+            log(f"[转发器] 国内直连 {direct_target[0]} 失败，回退上游隧道: {e}")
+    remote = create((remote_host, remote_port), 10)
+    if use_tls:
+        remote = ctx.wrap_socket(remote, server_hostname=remote_host)
+    return remote, True
+
+
 def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False, smart=False):
     """处理一条客户端连接：内网(bypass_lan)/中国IP(智能分流)目标直接连接；其余走远端隧道。"""
     import socket, select
@@ -1574,6 +1814,7 @@ def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False,
     is_connect = False
     injector = None
     direct_target = None
+    allow_direct_fallback = False
     max_headers = 65536
     try:
         while len(buf) < max_headers:
@@ -1600,34 +1841,35 @@ def _tls_relay(client, remote_host, remote_port, ctx, use_tls, bypass_lan=False,
                 target = _parse_connect_authority(first_line)
             else:
                 target = _parse_http_absolute(first_line)
-            if target and _decide_direct(target[0], bypass_lan, smart):
+            if target:
-                direct_target = target
+                direct_target = _direct_connect_target(
+                    target[0], target[1], bypass_lan, smart)
+                allow_direct_fallback = (direct_target is not None and
+                                         _is_smart_china_direct(target[0], bypass_lan, smart))
     except Exception as e:
         log(f"[转发器] 读取首包失败: {e}")
         client.close()
         return
 
-    count_traffic = direct_target is None
     remote = None
     try:
-        if direct_target:
+        remote, via_tunnel = _connect_route(
-            remote = socket.create_connection(direct_target, 10)
+            direct_target, allow_direct_fallback, remote_host, remote_port, ctx, use_tls)
+        count_traffic = via_tunnel
+        if not via_tunnel:
             if is_connect:
                 client.sendall(b"HTTP/1.1 200 Connection Established\r\n\r\n")
                 buf = b""
             else:
                 buf = _force_http_close(buf)
         else:
-            remote = socket.create_connection((remote_host, remote_port), 10)
-            if use_tls:
-                remote = ctx.wrap_socket(remote, server_hostname=remote_host)
             if is_http and b"\r\n\r\n" in buf:
                 if not is_connect:
                     injector = _make_request_injector()
                 buf = _inject_proxy_auth(buf)
         if buf:
             remote.sendall(buf)
-            if count_traffic:
+            if via_tunnel:
                 _traffic_add(up=len(buf))
     except Exception as e:
         log(f"[转发器] 连接目标失败: {e}")
@@ -2369,6 +2611,11 @@ class MainWindow(QWidget):
     def __init__(self):
         super().__init__()
         self._proxy_set_by_us = False  # 系统代理是否由本程序开启（退出时只清理自己设置的）
+        # 启动完成前仍允许用户取消。事件由 GUI 线程置位，后台启动线程只读取；
+        # 锁保护系统代理写入和取消清理之间的交接，避免遗留代理。
+        self._start_cancel_event = None
+        self._start_in_progress = False
+        self._proxy_state_lock = threading.RLock()
         self._status_busy = False      # 防止状态轮询线程堆叠
         # 把全局 log() 函数接到 GUI 日志区
         global _log_func
@@ -2595,8 +2842,20 @@ class MainWindow(QWidget):
             event.accept()  # 即使清理异常窗口也必须关闭
 
     # ── 开关操作 ──
-    def _set_toggle_button_style(self, is_start=True):
+    def _set_toggle_button_style(self, is_start=True, is_cancelling=False):
-        if is_start:
+        if is_cancelling:
+            self.btn_toggle.setText("取消开启")
+            self.btn_toggle.setStyleSheet("""
+                QPushButton {
+                    background-color: #cc8a2e; color: #fff; border: none;
+                    border-radius: 8px; padding: 10px 24px;
+                    font-size: 14px; font-weight: bold;
+                }
+                QPushButton:hover { background-color: #e1a246; }
+                QPushButton:pressed { background-color: #ad701d; }
+                QPushButton:disabled { background-color: #3a3020; color: #666; }
+            """)
+        elif is_start:
             self.btn_toggle.setText("开启代理")
             self.btn_toggle.setStyleSheet(f"""
                 QPushButton {{
@@ -2625,32 +2884,63 @@ class MainWindow(QWidget):
         """根据按钮当前文字判断是开启还是关闭。"""
         if self.btn_toggle.text() == "开启代理":
             self.on_start()
+        elif self.btn_toggle.text() == "取消开启":
+            self.on_cancel_start()
         else:
             self.on_stop()
 
     def on_start(self):
+        if self._start_in_progress:
+            return
         # 节点未配置：先弹出登录框，成功后继续；取消/失败则中止开启
         if _endpoint_domain() == "???":
             self._show_login_dialog()
             if _endpoint_domain() == "???":
                 log_red("未登录，无法开启代理")
                 return
-        self.btn_toggle.setEnabled(False)
         self.cb_pac.setEnabled(False)
         self.cb_bypass_lan.setEnabled(False)
         # 在 GUI 线程读取控件状态传参，避免子线程直接访问 Qt 控件（非线程安全）
         use_pac = self.cb_pac.isChecked()
         use_bypass_lan = self.cb_bypass_lan.isChecked()
-        threading.Thread(target=self._start_thread, args=(use_pac, use_bypass_lan), daemon=True).start()
+        self._start_cancel_event = threading.Event()
-    def _start_thread(self, use_pac, use_bypass_lan):
+        self._start_in_progress = True
+        # 网络探测可能需要数秒；此时保留按钮给用户取消，而不是把界面锁死。
+        self._set_toggle_button_style(is_cancelling=True)
+        self.btn_toggle.setEnabled(True)
+        threading.Thread(
+            target=self._start_thread,
+            args=(use_pac, use_bypass_lan, self._start_cancel_event),
+            daemon=True,
+        ).start()
+    def _start_thread(self, use_pac, use_bypass_lan, cancel_event):
         try:
-            self._start_proxy(use_pac, use_bypass_lan)
+            self._start_proxy(use_pac, use_bypass_lan, cancel_event)
+            _raise_if_start_cancelled(cancel_event)
             self._op_done.emit("start", True)
+        except _StartCancelled:
+            # on_cancel_start 的清理线程负责把界面切回关闭状态。
+            log("开启代理已取消")
         except Exception as e:
+            if cancel_event.is_set():
+                log("开启代理已取消")
+                return
             log(f"[错误] 开启失败: {e}")
             self._op_done.emit("start", False)
 
-    def _start_proxy(self, use_pac, use_bypass_lan):
+    def _cleanup_start_resources(self):
+        """回收本次启动已建立的资源；仅恢复本程序实际写入过的系统代理。"""
+        with self._proxy_state_lock:
+            proxy_owned = self._proxy_set_by_us
+            self._proxy_set_by_us = False
+        if proxy_owned:
+            proxy_off()
+        else:
+            # 取消可能恰好发生在“转发器已开、系统代理尚未写入”之间。
+            _tls_stop()
+            _stop_pac_server()
+
+    def _start_proxy(self, use_pac, use_bypass_lan, cancel_event=None):
         """启动本地转发器并设置系统代理（PAC/固定两种模式），可在后台线程调用。"""
         if _endpoint_domain() == "???":
             # 未配置节点时禁止开启：否则系统代理会指向连不通任何目标的转发器，导致断网
@@ -2666,39 +2956,61 @@ class MainWindow(QWidget):
         if use_pac:
             _load_cn_ranges()
             _load_force_proxy_suffixes()
+            _load_direct_domains()
+        _raise_if_start_cancelled(cancel_event)
         _tls_start(ep, _endpoint_port(), use_tls=True, bypass_lan=use_bypass_lan, smart=use_pac)
-        if use_pac:
+        try:
-            # 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流
+            _raise_if_start_cancelled(cancel_event)
-            pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}")
+            if use_pac:
+                # 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流
+                pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}")
+                if use_bypass_lan:
+                    pac_content = _apply_lan_bypass(pac_content)
+                # 在内存中启动本地 PAC HTTP 服务，不往 EXE 目录写任何文件
+                pac_port = _start_pac_server(pac_content)
+                pac_url = f"http://127.0.0.1:{pac_port}/proxy.pac"
+                log(f"使用 PAC: {pac_url}")
+                with self._proxy_state_lock:
+                    _raise_if_start_cancelled(cancel_event)
+                    proxy_on(pac_url=pac_url)
+                    self._proxy_set_by_us = True
+            else:
+                with self._proxy_state_lock:
+                    _raise_if_start_cancelled(cancel_event)
+                    proxy_on(bypass_lan=use_bypass_lan)
+                    self._proxy_set_by_us = True
+            _raise_if_start_cancelled(cancel_event)
             if use_bypass_lan:
-                pac_content = _apply_lan_bypass(pac_content)
+                log("不代理内网 IP：内网地址将直连")
-            # 在内存中启动本地 PAC HTTP 服务，不往 EXE 目录写任何文件
+            log("验证代理连通性...")
-            pac_port = _start_pac_server(pac_content)
+            _verify_proxy_connectivity(cancel_event=cancel_event)
-            pac_url = f"http://127.0.0.1:{pac_port}/proxy.pac"
+        except Exception:
-            log(f"使用 PAC: {pac_url}")
-            proxy_on(pac_url=pac_url)
-        else:
-            proxy_on(bypass_lan=use_bypass_lan)
-        if use_bypass_lan:
-            log("不代理内网 IP：内网地址将直连")
-        self._proxy_set_by_us = True  # 标记系统代理由本程序设置，退出时据此清理
-        log("验证代理连通性...")
-        import urllib.request
-        # 显式 ProxyHandler 强制经本地转发器；默认 opener 会套用系统/环境代理
-        # 设置（PAC 模式下注册表 ProxyEnable=0，urllib 认为无代理而绕开），造成假"可达"
-        opener = urllib.request.build_opener(
-            urllib.request.ProxyHandler({"http": f"{PROXY_HOST}:{PROXY_PORT}",
-                                         "https": f"{PROXY_HOST}:{PROXY_PORT}"}))
-        # baidu 验证国内直连路径，google 验证境外隧道路径
-        test_urls = [("baidu.com", "http://www.baidu.com"), ("google.com", "http://www.google.com")]
-        for name, url in test_urls:
             try:
-                opener.open(url, timeout=10)
+                # 连通性未通过（或用户取消）绝不能保留指向本地转发器的系统代理。
-                log(f"  {name} 可达")
+                self._cleanup_start_resources()
-            except Exception as e:
+            except Exception as cleanup_error:
-                log(f"  {name} 不可达: {e}")
+                log(f"[错误] 开启失败后的代理清理失败: {cleanup_error}")
+            raise
         log("开启成功")
 
+    def on_cancel_start(self):
+        """在 Baidu/Google 启动探测期间立即取消并回收代理。"""
+        if not self._start_in_progress or self._start_cancel_event is None:
+            return
+        self._start_cancel_event.set()
+        self.btn_toggle.setEnabled(False)
+        threading.Thread(target=self._cancel_start_thread, daemon=True).start()
+
+    def _cancel_start_thread(self):
+        try:
+            log("正在取消开启代理...")
+            self._cleanup_start_resources()
+            log("已取消开启代理")
+            self._op_done.emit("cancel", True)
+        except Exception as e:
+            log(f"[错误] 取消开启代理失败: {e}")
+            self._op_done.emit("cancel", False)
+
     def on_stop(self):
         """禁用按钮并在后台线程执行关闭流程。"""
         self.btn_toggle.setEnabled(False)
@@ -2826,13 +3138,23 @@ class MainWindow(QWidget):
 
     def _after_op(self, op, ok):
         """开启/关闭操作完成后刷新界面状态并恢复按钮/控件可用性。"""
+        if op in ("start", "cancel"):
+            self._start_in_progress = False
+            self._start_cancel_event = None
         self.refresh_status()
         self.btn_toggle.setEnabled(True)
         # 恢复模式/PAC 控件的时机：关闭完成后、或开启失败时。
         # 不能靠按钮文字判断——refresh_status 是异步的，此刻按钮样式还没更新
-        is_off = (op == "stop") or (op == "start" and not ok)
+        is_off = op == "stop" or (op == "cancel" and ok) or (op == "start" and not ok)
         self.cb_pac.setEnabled(is_off)
         self.cb_bypass_lan.setEnabled(is_off)
+        if op == "start" and ok:
+            self._set_toggle_button_style(is_start=False)
+        elif is_off:
+            self._set_toggle_button_style(is_start=True)
+        elif op == "cancel":
+            # 取消清理失败时保留“关闭”入口，供用户再次显式清理，而非误显示为未开启。
+            self._set_toggle_button_style(is_start=False)
         self._update_node_label()
     # ── 日志 ──
     _early_logs = []  # 窗口初始化期间（log_text 创建前）到达的日志缓存
@@ -2881,12 +3203,14 @@ class MainWindow(QWidget):
             self.px_dot.set_on()
             self.px_status.setText("已开启")
             self.px_status.setStyleSheet(f"color: {C_GREEN}; background: transparent;")
-            self._set_toggle_button_style(is_start=False)
+            if not self._start_in_progress:
+                self._set_toggle_button_style(is_start=False)
         else:
             self.px_dot.set_off()
             self.px_status.setText("已关闭")
             self.px_status.setStyleSheet(f"color: {C_RED}; background: transparent;")
-            self._set_toggle_button_style(is_start=True)
+            if not self._start_in_progress:
+                self._set_toggle_button_style(is_start=True)
 
         # 流量卡片（始终更新，未连接时显示 0）
         px_on = is_proxy_on()
@@ -3046,7 +3370,13 @@ class MainWindow(QWidget):
 
         本程序异常退出残留的代理设置由下次启动时 _cleanup_stale_proxy() 处理。
         """
-        if self._proxy_set_by_us:
+        if self._start_in_progress and self._start_cancel_event is not None:
+            self._start_cancel_event.set()
+            try:
+                self._cleanup_start_resources()
+            except Exception:
+                pass  # 清理失败也不能阻止退出，残留交给下次启动清理
+        elif self._proxy_set_by_us:
             try:
                 proxy_off()
             except Exception:

lite/test/test_pac_routing.py

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