From 0603224de2cf71dd745820c16c4bafc5599e37e1 Mon Sep 17 00:00:00 2001 From: RoyirinRosmontis Date: Wed, 16 Sep 2026 19:10:12 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=AE=E5=A4=8D=EF=BC=9A=E5=AE=8C=E5=96=84?= =?UTF-8?q?=20PAC=20=E5=88=86=E6=B5=81=E4=B8=8E=E5=90=AF=E5=8A=A8=E5=8F=96?= =?UTF-8?q?=E6=B6=88?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- lite/proxy_lite.py | 664 +++++++++++++++++++++++++--------- lite/test/test_pac_routing.py | 349 ++++++++++++++++++ 2 files changed, 846 insertions(+), 167 deletions(-) diff --git a/lite/proxy_lite.py b/lite/proxy_lite.py index 0831691..3109f62 100644 --- a/lite/proxy_lite.py +++ b/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: - continue - n = int(ip) - if ( - (n & 0xFF000000) == 0x0A000000 or - (n & 0xFFF00000) == 0xAC100000 or - (n & 0xFFFF0000) == 0xC0A80000 or - (n & 0xFF000000) == 0x7F000000 or - (n & 0xFFFF0000) == 0xA9FE0000 - ): + if ip.version == 4 and _is_private_int(int(ip)): + return True + if ip.version == 6 and (ip.is_loopback or ip.is_link_local + or ip in ipaddress.ip_network("fc00::/7")): 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(): - """加载 pac_ip_cn.json 中的中国 IPv4 段，合并为整数区间供二分查找。""" - global _CN_RANGES, _CN_RANGES_LOADED - if _CN_RANGES_LOADED: - return _CN_RANGES - _CN_RANGES_LOADED = True - try: - 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 +def _merge_ip_ranges(cidrs, version): + """将指定 IP 版本的 CIDR 合并为整数闭区间，供二分查找。""" + import ipaddress + ranges = [] + for c in cidrs: + net = ipaddress.ip_network(str(c).strip(), strict=False) + if net.version == version: ranges.append((int(net.network_address), int(net.broadcast_address))) - ranges.sort() - merged = [] - for s, e in ranges: - if merged and s <= merged[-1][1] + 1: - if e > merged[-1][1]: - merged[-1] = (merged[-1][0], e) - else: - merged.append((s, e)) - _CN_RANGES = merged - log(f"已加载中国 IP 段: {len(cidrs)} 条 CIDR，合并为 {len(merged)} 段") - except Exception as e: - log_red(f"pac_ip_cn.json 加载失败: {e}") - return _CN_RANGES + ranges.sort() + merged = [] + for start, end in ranges: + if merged and start <= merged[-1][1] + 1: + if end > merged[-1][1]: + merged[-1] = (merged[-1][0], end) + else: + merged.append((start, end)) + return merged + + +def _load_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: - return _FORCE_PROXY_SUFFIXES - _FORCE_PROXY_LOADED = True - try: - import json - path = _user_path("pac_domains.json") - if path is None: + with _PAC_DATA_LOCK: + if _FORCE_PROXY_LOADED: + return _FORCE_PROXY_SUFFIXES + try: + import json + path = _user_path("pac_domains.json") + 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)} 条关键域名兜底") - 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 + finally: + _FORCE_PROXY_LOADED = True + return _FORCE_PROXY_SUFFIXES + + +def _build_suffix_trie(suffixes): + """按反向域名标签建立 Trie，适合大规模后缀规则的 O(标签数) 匹配。""" + root = {} + for suffix in suffixes: + node = root + for label in reversed(suffix.split(".")): + node = node.setdefault(label, {}) + node[""] = True + return root + + +def _suffix_trie_matches(host: str, trie) -> bool: + node = trie + for label in reversed(host.rstrip(".").lower().split(".")): + node = node.get(label) + if node is None: + return False + if node.get(""): + return True + return bool(node and node.get("")) + + +def _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: - 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 = () - return _FORCE_PROXY_SUFFIXES + if rule.startswith("full:"): + host = rule[len("full:"):].rstrip(".") + if host: + exact.add(host) + continue + if rule.startswith("regexp:"): + pattern = raw_rule[len("regexp:"):] + if not pattern: + invalid_regexes += 1 + continue + try: + regexes.append(re.compile(pattern, re.IGNORECASE)) + except re.error: + invalid_regexes += 1 + continue + suffix = rule.lstrip("*.").rstrip(".") + if suffix: + suffixes.add(suffix) + _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): - """判断目标是否命中中国 IP 段；域名会先解析。 +def _resolve_china_direct_ip(host: str): + """返回可安全直连的中国 IPv4/IPv6，或 None。 - 任一解析结果命中中国段 -> True（直连）；全部不命中 -> False（走隧道）； - 解析失败或数据未加载 -> None（无法判定，安全起见走隧道）。 + 同一地址族的所有解析结果都必须在对应中国 IP 段内，才允许直连；混合 + 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() - if not ranges: - return None - for ip in addresses: - if ip.version != 4: - continue - n = int(ip) - # 0.0.0.0 / 内网 / 回环解析结果疑似污染或局域网，不参与判定 - if n == 0 or _is_private_int(n): - continue - i = bisect.bisect_right(ranges, (n, 0xFFFFFFFF)) - if i > 0 and ranges[i - 1][0] <= n <= ranges[i - 1][1]: - return True - return False + _load_cn_ranges() + + def family_target(version, ranges): + if not ranges: + return None + family = [ip for ip in addresses if ip.version == version] + if not family or any(not ip.is_global for ip in family): + return None + limit = (1 << (32 if version == 4 else 128)) - 1 + for ip in family: + n = int(ip) + i = bisect.bisect_right(ranges, (n, limit)) + if not (i > 0 and ranges[i - 1][0] <= n <= ranges[i - 1][1]): + 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): - direct_target = target + if 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 = socket.create_connection(direct_target, 10) + remote, via_tunnel = _connect_route( + 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): - if is_start: + def _set_toggle_button_style(self, is_start=True, is_cancelling=False): + 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() - def _start_thread(self, use_pac, use_bypass_lan): + self._start_cancel_event = threading.Event() + 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: - # 本地生成全量转发 PAC（不再从服务器拉取），全部流量交给转发器做智能分流 - pac_content = _build_pac(f"{PROXY_HOST}:{PROXY_PORT}") + try: + _raise_if_start_cancelled(cancel_event) + 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) - # 在内存中启动本地 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}") - 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: + log("不代理内网 IP：内网地址将直连") + log("验证代理连通性...") + _verify_proxy_connectivity(cancel_event=cancel_event) + except Exception: try: - opener.open(url, timeout=10) - log(f" {name} 可达") - except Exception as e: - log(f" {name} 不可达: {e}") + # 连通性未通过（或用户取消）绝不能保留指向本地转发器的系统代理。 + self._cleanup_start_resources() + except Exception as cleanup_error: + 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: diff --git a/lite/test/test_pac_routing.py b/lite/test/test_pac_routing.py index eb58829..2d5e328 100644 --- a/lite/test/test_pac_routing.py +++ b/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)