diff --git a/lite/proxy_lite.py b/lite/proxy_lite.py index 52d0aa8..310db42 100644 --- a/lite/proxy_lite.py +++ b/lite/proxy_lite.py @@ -102,6 +102,7 @@ def _load_nodes(): _SELECTED_NODE = 0 # 当前选中节点下标（默认第一条） _SELECTED_RESTORED = False # 启动时是否已按设置恢复过上次选中的节点 +_NODES_SYNC_DONE = threading.Event() # 订阅同步线程完成信号（启动校验等它） def _current_node(): """当前使用的节点：默认取列表第一个；无节点时返回 None。""" @@ -270,20 +271,42 @@ def _first_reachable_index(nodes): return i return 0 -def _verify_selected_node(): - """启动时校验上次选中的节点在本机是否还可用。 +def _rematch_saved_domain(): + """按 settings 的 current_node（域名）在最新节点列表里重匹配下标。 - 典型场景：上次选了 IPv6 专用节点，这次本机没有 IPv6 出口（或该节点已不可达）， - 继续选中它只会在开启代理时才报错。这里自动回退到第一个可达节点并记录到设置。 - 返回 True 表示发生了切换（调用方据此刷新标题栏）。""" + 远程同步后服务端可能调整节点顺序：只按下标会静默连到别的节点。 + 返回 True 表示发生了切换。""" global _SELECTED_NODE nodes = _load_nodes() + if not nodes: + return False + saved = str(_settings_load().get("current_node") or "") + cur = nodes[_SELECTED_NODE] if 0 <= _SELECTED_NODE < len(nodes) else None + if saved and (not cur or str(cur.get("domain") or "") != saved): + match = next((i for i, n in enumerate(nodes) + if str(n.get("domain") or "") == saved), None) + if match is not None: + _SELECTED_NODE = match + log(f"节点列表已更新，继续使用上次选择的节点 {saved}") + return True + return False + +def _verify_selected_node(): + """启动校验，返回 True 表示发生了切换（调用方据此刷新标题栏）。 + + 两步：① 按设置记录的域名在最新节点列表里重匹配（服务端调过顺序时 + 下标会连到别的节点）；② 选中节点在本机不可达（典型：上次选了 IPv6 + 专用节点，这次本机没有 IPv6 出口）时自动回退到第一个可达节点。""" + global _SELECTED_NODE + nodes = _load_nodes() + switched = _rematch_saved_domain() if not nodes or not (0 <= _SELECTED_NODE < len(nodes)): - return False - if _node_reachable(nodes[_SELECTED_NODE]): - return False + return switched + cur = nodes[_SELECTED_NODE] + if _node_reachable(cur): + return switched idx = _first_reachable_index(nodes) - old = str(nodes[_SELECTED_NODE].get("domain") or "") + old = str(cur.get("domain") or "") new = str(nodes[idx].get("domain") or "") _SELECTED_NODE = idx _save_current_node(new) @@ -479,8 +502,76 @@ def _notify_proxy_change(): pass +def _reg_query_value(name): + """查询 HKCU Internet Settings 下指定值，返回 (类型, 数据)；值不存在返回 None。""" + r = _run(["reg", "query", PROXY_REG, "/v", name], warn=False) + if r.returncode != 0: + return None + for line in (r.stdout or "").splitlines(): + if name in line: + parts = line.strip().split(None, 2) # 形如 "ProxyServer REG_SZ 127.0.0.1:7897" + if len(parts) >= 3: + return parts[1], parts[2] + return None + +def _is_our_pac_url(value: str) -> bool: + """value 是否指向本程序的本地 PAC 服务（http://127.0.0.1:<端口>/proxy.pac）。""" + import re + return bool(re.fullmatch(r"http://127\.0\.0\.1:\d+/proxy\.pac", value.strip())) + +_PROXY_BACKUP_KEY = "prev_proxy_state" # settings.json 里的系统代理原值备份 +_proxy_prev = None # 运行期备份：None = 尚未备份 + +def _backup_proxy_state(): + """覆盖系统代理前把原值备份下来（内存 + settings.json，防崩溃丢失）。 + + 只在尚未备份时执行：本次运行里第一次开启代理时的注册表状态才是用户 + 原有的（可能是其他代理软件的值），之后反复开/关不应覆盖首次备份。""" + global _proxy_prev + if _proxy_prev is not None: + return + state = {} + for name in ("ProxyEnable", "ProxyServer", "ProxyOverride", "AutoConfigURL"): + v = _reg_query_value(name) + state[name] = list(v) if v else None + _proxy_prev = state + try: + _settings_save({**_settings_load(), _PROXY_BACKUP_KEY: state}) + except Exception: + pass # 持久化失败只影响崩溃后的恢复，不影响本次运行 + +def _restore_proxy_state(): + """恢复开启代理前的系统代理原值；没有备份时退化为清空（历史行为）。""" + global _proxy_prev + state = _proxy_prev + if state is None: + state = _settings_load().get(_PROXY_BACKUP_KEY) + if isinstance(state, dict): + for name in ("ProxyEnable", "ProxyServer", "ProxyOverride", "AutoConfigURL"): + v = state.get(name) + if v: + _run(["reg", "add", PROXY_REG, "/v", name, "/t", v[0], "/d", v[1], "/f"], warn=False) + else: + _run(["reg", "delete", PROXY_REG, "/v", name, "/f"], warn=False) + log("已恢复开启代理前的系统代理设置") + else: + # 无备份（旧版本升级后首次运行）：清空，与历史行为一致 + for name in ("ProxyServer", "ProxyOverride", "AutoConfigURL"): + _run(["reg", "delete", PROXY_REG, "/v", name, "/f"], warn=False) + _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "0", "/f"]) + _proxy_prev = None + # 备份已消费：清除持久化副本，避免下次启动误恢复 + try: + data = _settings_load() + if isinstance(data, dict) and _PROXY_BACKUP_KEY in data: + del data[_PROXY_BACKUP_KEY] + _settings_save(data) + except Exception: + pass + def proxy_on(pac_url=None, bypass_lan=False): """设置 Windows 系统代理。pac_url 不为空时启用 PAC 模式，否则使用固定代理。""" + _backup_proxy_state() # 覆盖前保存用户原有代理设置，关闭时恢复而不是清空 if pac_url: # PAC 模式：仅设置 AutoConfigURL，ProxyEnable 置 0， # 避免程序读到"固定代理已启用但服务器为空"的混乱状态 @@ -501,11 +592,7 @@ def proxy_on(pac_url=None, bypass_lan=False): def proxy_off(): """关闭系统代理并停止 TLS 转发器 / PAC 本地服务。""" log("正在关闭系统代理...") - _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "0", "/f"]) - # 清除 ProxyServer / ProxyOverride / AutoConfigURL，确保两种模式都清理干净 - _run(["reg", "delete", PROXY_REG, "/v", "ProxyServer", "/f"], warn=False) - _run(["reg", "delete", PROXY_REG, "/v", "ProxyOverride", "/f"], warn=False) - _run(["reg", "delete", PROXY_REG, "/v", "AutoConfigURL", "/f"], warn=False) + _restore_proxy_state() # 恢复用户原有代理设置，而不是简单清空 log("系统代理已关闭") _notify_proxy_change() _tls_stop() @@ -628,17 +715,28 @@ def _inject_proxy_auth(data: bytes) -> bytes: nginx 的 auth_basic 校验 Authorization 而非 Proxy-Authorization， 因此转发器向上游发送 Basic 认证时使用 Authorization 头。 + 只按完整头名精确匹配（行首到冒号）：Proxy-Authorization 等含 + "authorization" 字样的行不算已有认证，不能阻止注入；目标站自己的 + Authorization（明文 HTTP 请求）才跳过。注入位置紧跟请求行， + 保证上游（取首个 Authorization）读到的是我们的认证。 """ if not _starts_with_method(data): return data end = data.find(b"\r\n\r\n") if end == -1: return data - headers_lower = data[:end].lower() - if b"authorization:" in headers_lower: - return data + head = data[:end] + first_end = head.find(b"\r\n") + request_line = head if first_end == -1 else head[:first_end] + headers = b"" if first_end == -1 else head[first_end + 2:] + for line in headers.split(b"\r\n"): + if b":" not in line: + continue + name = line[:line.index(b":")].strip().lower() + if name == b"authorization": + return data # 已有目标站自己的 Authorization，不再注入 auth_line = f"Authorization: Basic {_proxy_auth_basic()}\r\n".encode() - return data[:end + 2] + auth_line + data[end + 2:] + return request_line + b"\r\n" + auth_line + headers + data[end:] def _make_request_injector(): """为明文 HTTP 代理连接生成逐条注入认证的流式处理器。 @@ -661,7 +759,43 @@ def _make_request_injector(): return b"" return feed -@functools.lru_cache(maxsize=256) +def _ttl_cache(seconds: int, maxsize: int = 16384): + """带过期时间的缓存装饰器（替代无 TTL 的 lru_cache）。 + + 域名分流判定结果长期缓存会让域名换 IP 后继续沿用旧的直连/隧道判断； + 到期自动失效重查。保留 cache_clear() 兼容（测试脚本在用）。""" + def deco(func): + store = {} + lock = threading.Lock() + + def wrapper(*args, **kwargs): + key = (args, tuple(sorted(kwargs.items()))) + now = time.time() + with lock: + hit = store.get(key) + if hit and now - hit[0] < seconds: + return hit[1] + val = func(*args, **kwargs) + with lock: + store[key] = (now, val) + if len(store) > maxsize: + cutoff = now - seconds + for k, (t, _) in list(store.items()): + if t < cutoff: + del store[k] + if len(store) > maxsize: + store.clear() # 极端情况下全清：最坏只是多几次解析 + return val + + def cache_clear(): + with lock: + store.clear() + + wrapper.cache_clear = cache_clear + return wrapper + return deco + +@_ttl_cache(300, maxsize=2048) def _is_private_ip(host: str) -> bool: """判断目标是否命中常用内网/回环 IPv4；域名会解析后判断。""" try: @@ -1029,31 +1163,34 @@ def _auth_load(): def _nodes_sync_remote(): """启动时用已保存凭据拉取最新节点配置（内容比对，变化才覆盖）。""" - cred = _auth_load() - if not cred: - return # 从未登录过，等登录框处理 - url = str(_config_val("nodes_url", "") or "").strip() - if not url: - return try: - import json - data = _fetch_nodes_auth(url, cred["username"], cred["password"]) - if not (isinstance(data, list) and data - and all(isinstance(n, dict) for n in data) - and str(data[0].get("domain") or "").strip()): - raise ValueError("返回的节点数据无效") - text = json.dumps(data, ensure_ascii=False, indent=2) + "\n" - local = _USER_DIR / "nodes.json" - if local.is_file() and _nodes_decrypt(local.read_bytes()) == text: - log("节点配置已是最新") + cred = _auth_load() + if not cred: + return # 从未登录过，等登录框处理 + url = str(_config_val("nodes_url", "") or "").strip() + if not url: return - _USER_DIR.mkdir(parents=True, exist_ok=True) - local.write_bytes(_nodes_encrypt(text)) - global _NODES - _NODES = None # 使后续读取（节点菜单/开启代理）拿到新配置 - log(f"节点配置已更新: {len(data)} 个节点 (nodes.json)") - except Exception as e: - log(f"节点配置在线更新失败，使用本地配置: {e}") + try: + import json + data = _fetch_nodes_auth(url, cred["username"], cred["password"]) + if not (isinstance(data, list) and data + and all(isinstance(n, dict) for n in data) + and str(data[0].get("domain") or "").strip()): + raise ValueError("返回的节点数据无效") + text = json.dumps(data, ensure_ascii=False, indent=2) + "\n" + local = _USER_DIR / "nodes.json" + if local.is_file() and _nodes_decrypt(local.read_bytes()) == text: + log("节点配置已是最新") + return + _USER_DIR.mkdir(parents=True, exist_ok=True) + local.write_bytes(_nodes_encrypt(text)) + global _NODES + _NODES = None # 使后续读取（节点菜单/开启代理）拿到新配置 + log(f"节点配置已更新: {len(data)} 个节点 (nodes.json)") + except Exception as e: + log(f"节点配置在线更新失败，使用本地配置: {e}") + finally: + _NODES_SYNC_DONE.set() # 无论成败都要放行启动校验 class _AuthError(Exception): """订阅接口返回的错误：code 为 HTTP 状态码，str 为服务端返回的中文提示。""" @@ -1208,7 +1345,7 @@ def _is_private_int(n: int) -> bool: (n & 0xFF000000) == 0x7F000000 or (n & 0xFFFF0000) == 0xA9FE0000) -@functools.lru_cache(maxsize=16384) +@_ttl_cache(300, maxsize=16384) def _is_china_ip(host: str): """判断目标是否命中中国 IP 段；域名会先解析。 @@ -1289,7 +1426,7 @@ def _whitelist_entry_match(host: str, entry: str) -> bool: # 域名条目：精确匹配或子域后缀匹配（与 pac_domains 同一规则） return h == entry or h.endswith("." + entry) -@functools.lru_cache(maxsize=8192) +@_ttl_cache(300, maxsize=8192) def _is_whitelisted(host: str, version: int) -> bool: """host 是否命中已启用的白名单规则（按版本键控缓存，规则变更即失效）。""" entries = [r["value"] for r in _whitelist_rules() if r.get("enabled")] @@ -1314,11 +1451,24 @@ def _decide_direct(host: str, bypass_lan: bool, smart: bool) -> bool: return False def _parse_connect_authority(first_line: str): - """从 CONNECT 首行提取 (host, port)，失败返回 None。""" + """从 CONNECT 首行提取 (host, port)，失败返回 None。 + + IPv6 字面量带方括号（CONNECT [::1]:443）：剥掉括号再按最后一个冒号拆分， + 否则会把 "[::1]" 当成主机名，直连/白名单判定全部失效。""" parts = first_line.split(" ") if len(parts) < 2: return None authority = parts[1] + if authority.startswith("[") and "]" in authority: + # 括号必须紧跟在主机名部分之后（[::1]:443）；"[::1" 无右括号等畸形输入 + # 会因 "]:" 检查失败而走下面的普通拆分，最终被 port_s.isdigit() 拦下 + try: + host = authority[1:authority.index("]")] + tail = authority[authority.index("]") + 1:] + if tail.startswith(":") and tail[1:].isdigit(): + return host, int(tail[1:]) + except ValueError: + pass if ":" not in authority: return None host, _, port_s = authority.rpartition(":") @@ -1501,27 +1651,28 @@ def _release_forwarder(started_here): def is_proxy_on(): """系统代理是否生效：固定代理已开启，或 AutoConfigURL 指向本地 PAC 服务。""" - r = _run(["reg", "query", PROXY_REG, "/v", "ProxyEnable"], warn=False) - if "0x1" in (r.stdout or ""): + v = _reg_query_value("ProxyEnable") + if v and "0x1" in (v[1] or ""): return True - r = _run(["reg", "query", PROXY_REG, "/v", "AutoConfigURL"], warn=False) - return "127.0.0.1" in (r.stdout or "") + v = _reg_query_value("AutoConfigURL") + return bool(v) and _is_our_pac_url(v[1] or "") def _cleanup_stale_proxy(): - """清理上次异常退出残留的系统代理设置。 + """清理上次异常退出残留的系统代理设置，并恢复用户原有设置。 TLS 转发器 / PAC 服务只活在进程内，程序启动时若注册表仍指向 本地转发端口或本地 PAC 地址，说明是残留——不清理会导致全网不通。 + 只认本程序的精确地址：其他软件（如 Clash）的 PAC 也在 127.0.0.1 上， + 不能因为地址里带 127.0.0.1 就误删。 """ try: - if not is_proxy_on(): + server = _reg_query_value("ProxyServer") + pac = _reg_query_value("AutoConfigURL") + ours = (bool(server) and (server[1] or "").strip() == f"{PROXY_HOST}:{PROXY_PORT}") or \ + (bool(pac) and _is_our_pac_url(pac[1] or "")) + if not ours: return - r = _run(["reg", "query", PROXY_REG, "/v", "ProxyServer"], warn=False) - r2 = _run(["reg", "query", PROXY_REG, "/v", "AutoConfigURL"], warn=False) - stale = (f"{PROXY_HOST}:{PROXY_PORT}" in (r.stdout or "") - or "127.0.0.1" in (r2.stdout or "")) - if stale: - proxy_off() + proxy_off() # 内部从备份恢复用户原有设置（无备份则清空） except Exception: pass @@ -2198,9 +2349,10 @@ class MainWindow(QWidget): def _verify_node_thread(self): """等节点配置在线同步完成后，校验当前选中节点是否可用，必要时自动回退。 - 探测要解析 DNS 并尝试 TCP 连接，放在后台线程，避免卡住启动。""" - import time - time.sleep(1.5) + 探测要解析 DNS 并尝试 TCP 连接，放在后台线程，避免卡住启动。 + 订阅同步可能较慢：等同步完成（最多 15 秒），同步后才校验， + 否则服务端换过节点顺序时校验的是旧列表。""" + _NODES_SYNC_DONE.wait(15) try: if _verify_selected_node(): self._node_switched.emit() @@ -2695,6 +2847,8 @@ class MainWindow(QWidget): # 先初始化结果，任何路径（含跳过）完成后 _on_update_done 都能安全读取 self._latest_version = "" self._download_url = "" + self._asset_size = 0 # release API 记录的附件大小，下载后比对 + self._checksum_url = "" # SHA-256 边车附件链接（无则跳过校验） try: if _skip_update_check() and not manual: return # 设置跳过（仅自动检查）：不检查也不打日志 @@ -2721,11 +2875,23 @@ class MainWindow(QWidget): req = urllib.request.Request(url) data = opener.open(req, timeout=20).read().decode("utf-8-sig") info = json.loads(data) - # 基于 Gitea 仓库 releases API：版本取 tag_name，下载取 exe 附件链接 + # 基于 Gitea 仓库 releases API：版本取 tag_name，下载取 exe 附件链接。 + # 附件可能有多个（发布流程附带 .sha256 边车），按名字精确匹配而不是 + # 固定取第一个，避免附件顺序变化时下载到错文件。 self._latest_version = str(info.get("tag_name") or "") assets = info.get("assets") or [] - self._download_url = (assets[0].get("browser_download_url") or "" - if assets else "") + wanted = f"RRPP_Lite_{self._latest_version}.exe" if self._latest_version else "" + exe_asset = next((a for a in assets if a.get("name") == wanted), None) + if exe_asset is None: + # 老 release 命名不齐时回退到任意 exe 附件 + exe_asset = next((a for a in assets + if str(a.get("name") or "").lower().endswith(".exe")), None) + if exe_asset is None: + raise RuntimeError("release 中没有可用的 EXE 附件") + self._download_url = exe_asset.get("browser_download_url") or "" + self._asset_size = int(exe_asset.get("size") or 0) + sha_asset = next((a for a in assets if a.get("name") == wanted + ".sha256"), None) + self._checksum_url = (sha_asset.get("browser_download_url") or "") if sha_asset else "" log(f"当前版本: {VERSION}，最新版本: {self._latest_version or '未知'}") except Exception as e: log(f"更新检查失败: {e}") @@ -2748,9 +2914,15 @@ class MainWindow(QWidget): threading.Thread(target=self._download_update, daemon=True).start() def _download_update(self): - """后台下载新版本 EXE 到 EXE 所在目录。""" + """后台下载新版本 EXE 到 EXE 所在目录。 + + 落盘前校验：① 大小与 release API 记录一致（传输完整性）； + ② 发布流程附带 .sha256 边车时校验 SHA-256（防链路被攻破/文件被替换）。 + 边车缺失（老 release）只做大小校验。 + """ try: - import urllib.request, urllib.parse, base64 + import hashlib + import urllib.request, urllib.parse url = self._download_url if not url: self._update_download_done.emit(False, "下载链接为空") @@ -2762,29 +2934,40 @@ class MainWindow(QWidget): log(f"正在下载新版本: {url}") log(f"保存到: {save_path}") - # 同更新检查：按「使用代理更新」设置选择隧道或直连 - if _update_via_proxy(): - started_here = _ensure_forwarder() - try: + def fetch(target): + # 同更新检查：按「使用代理更新」设置选择隧道或直连 + if _update_via_proxy(): opener = urllib.request.build_opener(urllib.request.ProxyHandler( {"http": f"{PROXY_HOST}:{PROXY_PORT}", "https": f"{PROXY_HOST}:{PROXY_PORT}"})) - req = urllib.request.Request(url) + else: + opener = _pac_no_proxy_opener() + with opener.open(urllib.request.Request(target), timeout=120) as resp: + return resp.read() - with opener.open(req, timeout=120) as resp: - data = resp.read() - finally: - _release_forwarder(started_here) - else: - opener = _pac_no_proxy_opener() - req = urllib.request.Request(url) - - with opener.open(req, timeout=120) as resp: - data = resp.read() + started_here = _ensure_forwarder() if _update_via_proxy() else False + try: + data = fetch(url) + checksum_text = "" + if self._checksum_url: + checksum_text = fetch(self._checksum_url).decode("ascii", "replace") + finally: + _release_forwarder(started_here) if len(data) < 1024: self._update_download_done.emit(False, "下载文件过小，可能不是有效的 EXE") return + if self._asset_size and len(data) != self._asset_size: + self._update_download_done.emit( + False, f"下载不完整：实际 {len(data)} 字节，应为 {self._asset_size} 字节") + return + if checksum_text: + want = checksum_text.strip().split()[0].lower() + got = hashlib.sha256(data).hexdigest() + if len(want) != 64 or want != got: + self._update_download_done.emit(False, "更新包校验失败（SHA-256 不匹配），已放弃") + return + log("更新包 SHA-256 校验通过") save_path.write_bytes(data) log(f"新版本已下载: {save_path}") diff --git a/lite/test/check_node_menu.py b/lite/test/check_node_menu.py index 950afb6..131fcdd 100644 --- a/lite/test/check_node_menu.py +++ b/lite/test/check_node_menu.py @@ -72,4 +72,33 @@ else: else: (pp._USER_DIR / "settings.json").write_bytes(_bak) +# ── 服务端调整节点顺序：按域名重匹配而不是死守下标 ────────────────── +print("\n服务端调序（_rematch_saved_domain）:") +if len(nodes) < 2: + print(" 节点不足 2 个，跳过") +else: + import json + dom_old = nodes[0].get("domain") + ordered = [nodes[1], nodes[0]] + nodes[2:] # 模拟服务端交换了前两个节点 + pp._NODES = ordered + # 设置里记录的是原下标 0 的域名，重匹配后下标应跟随域名（新列表里是 1） + pp._settings_save({**pp._settings_load(), "current_node": dom_old}) + switched = pp._rematch_saved_domain() + ok = switched and ordered[pp._selected_index()].get("domain") == dom_old + print(f" current_node={dom_old} 调序后跟随域名 -> 下标 {pp._selected_index()} " + f"{'PASS' if ok else 'FAIL'}") + # 已是正确域名时再跑一次：不应重复切换 + ok2 = pp._rematch_saved_domain() is False + print(f" 再次重匹配不重复切换 {'PASS' if ok2 else 'FAIL'}") + # 域名已不在列表（节点被移除）：不切换，保持现状 + pp._settings_save({**pp._settings_load(), "current_node": "gone.example.com"}) + ok3 = pp._rematch_saved_domain() is False + print(f" 节点被移除时不动 {'PASS' if ok3 else 'FAIL'}") + pp._NODES = None # 还原缓存，后续用例重读文件 + # 还原 settings.json，避免测试污染 + if _bak is None: + (pp._USER_DIR / "settings.json").unlink(missing_ok=True) + else: + (pp._USER_DIR / "settings.json").write_bytes(_bak) + print("OK") diff --git a/lite/test/test_pac_routing.py b/lite/test/test_pac_routing.py new file mode 100644 index 0000000..eb58829 --- /dev/null +++ b/lite/test/test_pac_routing.py @@ -0,0 +1,389 @@ +# -*- coding: utf-8 -*- +"""PAC（智能分流）模式路由决策测试。 + +核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？ + +判定链（proxy_lite._decide_direct）： + 0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效） + 1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定） + 2. 内网 IP 且勾选 bypass_lan -> 直连 + 3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道 + +直接 import proxy_lite 调用真实决策函数，数据用真实的 pac_ip_cn.json / +pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。 +""" + +import sys +from pathlib import Path + +TEST_DIR = Path(__file__).resolve().parent +LITE_DIR = TEST_DIR.parent + +sys.path.insert(0, str(TEST_DIR)) # 假 PyQt5 桩 +sys.path.insert(0, str(LITE_DIR)) + +import json +import socket +import proxy_lite as pp +from proxy_lite import _decide_direct, _is_china_ip, _is_force_proxy, _is_private_ip, _is_whitelisted + +# PAC 数据已不再随项目保留：从 CDN 拉取到临时用户目录（真实数据端到端） +import tempfile +pp._USER_DIR = Path(tempfile.mkdtemp()) / "RRPP" +pp._user_init_remote(base_url="https://cdn.rrrosmontis.icu/RRPP/PAC/") + +pp._load_cn_ranges() +pp._load_force_proxy_suffixes() +if not pp._CN_RANGES: + print("警告: 未能获取 pac_ip_cn.json（CDN 不可达？），中国 IP 判定将失效") +print(f"中国IP段: {len(pp._CN_RANGES or [])} 段, 强制域名表: {len(pp._FORCE_PROXY_SUFFIXES or [])} 条\n") + + +def resolved_ips(host): + try: + return sorted({a[4][0] for a in socket.getaddrinfo(host, 443, socket.AF_INET)}) + except Exception: + return [] + + +def mark(ip): + if _is_private_ip(ip): + return "内网" + if _is_china_ip(ip): + return "中国IP" + return "国外IP" + + +PATH = {True: "直连", False: "隧道"} +MODES = [ + ("PAC+bypass_lan", dict(bypass_lan=True, smart=True)), # 默认勾选状态 + ("PAC无bypass_lan", dict(bypass_lan=False, smart=True)), + ("固定代理模式", dict(bypass_lan=False, smart=False)), # 全部走隧道 +] + +# (名称, host, 期望 PAC+bypass_lan 决策: True=直连 / False=隧道) +cases = [ + ("境外域名 google.com", "google.com", False), + ("境外域名 github.com", "github.com", False), + ("境外域名 youtube.com", "youtube.com", False), + ("域名表强制 claude.ai", "claude.ai", False), + ("域名表强制 *.google 后缀", "antigravity.google", False), + ("国内域名 www.baidu.com", "www.baidu.com", True), + ("国内域名 www.qq.com", "www.qq.com", True), + ("内网 localhost", "localhost", True), + ("内网 IP 192.168.1.1", "192.168.1.1", True), + ("境外 IP 8.8.8.8", "8.8.8.8", False), + ("国内 IP 114.114.114.114", "114.114.114.114", True), +] + +failed = 0 +for name, host, expect in cases: + ips = resolved_ips(host) + force = _is_force_proxy(host) + marks = f"[{','.join(mark(ip) for ip in ips)}]" if ips else "[解析失败]" + row = f"{name:<22} {host:<18} {str(ips) if ips else '无IP':<45} {marks}" + results = [] + for mode_name, kw in MODES: + got = _decide_direct(host, **kw) + results.append(f"{mode_name}={PATH[got]}") + pac = _decide_direct(host, **MODES[0][1]) + ok = pac == expect + failed += 0 if ok else 1 + print(f"{row}\n {'域名表命中' if force else '域名表未命中':<10} | {' '.join(results)} | 期望 {PATH[expect]} -> {'PASS' if ok else 'FAIL'}") + +print(f"\n{'='*70}") +print(f"结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}") +print("=" * 70) +print("PAC 模式(智能分流)下: 域名解析结果全部为国外 IP -> 走隧道代理 ✓" + if not failed else "存在 FAIL，请检查上方明细") + +# ── 附加测试：不在域名表的被墙境外站 ───────────────────── +# 这类站点不能靠域名表兜底，必须仅凭「解析结果为国外 IP」强制走隧道， +# 否则会尝试直连 -> 被墙。 +print("\n" + "=" * 70) +print("附加: 不在 pac_domains.json 的被墙境外站（只能靠 IP 判定走隧道）") +print("=" * 70) + +extra_candidates = [ + "dailymotion.com", "vimeo.com", "notion.so", "linear.app", + "quora.com", "feedly.com", "dropbox.com", "tumblr.com", + "flickr.com", "pastebin.com", "fandom.com", "archive.org", + "godaddy.com", "imgur.com", "kickstarter.com", "shopify.com", +] + +extra_failed = 0 +skipped = 0 +for host in extra_candidates: + if _is_force_proxy(host): + print(f"{host:<20} 已在域名表中，跳过") + skipped += 1 + continue + ips = resolved_ips(host) + if not ips: + print(f"{host:<20} 解析失败，跳过") + skipped += 1 + continue + foreign = all(not _is_private_ip(ip) and not _is_china_ip(ip) for ip in ips) + if not foreign: + print(f"{host:<20} 解析到非国外 IP {ips}，跳过（不是有效的境外用例）") + skipped += 1 + continue + got = _decide_direct(host, bypass_lan=True, smart=True) # PAC 模式 + if got is not False: + # 决策函数内部会重新解析域名，CDN 轮询 DNS 可能与其上方展示的解析结果 + # 不一致：清缓存重试一次，仍不一致则按 DNS 不稳定跳过（非代码问题） + _is_china_ip.cache_clear() + _is_private_ip.cache_clear() + ips = resolved_ips(host) + got = _decide_direct(host, bypass_lan=True, smart=True) + if got is not False: + print(f"{host:<20} 解析/判定随 DNS 轮询不一致 {ips}，跳过（非代码问题）") + skipped += 1 + continue + print(f"{host:<20} 重试后走隧道（DNS 瞬时差异） PASS") + continue + extra_failed += 0 + print(f"{host:<20} {str(ips):<50} [{'，'.join(mark(ip) for ip in ips)}] -> " + f"{PATH[got]} (期望隧道) PASS") + +print(f"\n附加测试: {len(extra_candidates)} 个候选, 有效用例 {len(extra_candidates) - skipped} 个, " + f"{'全部 PASS' if not extra_failed else f'{extra_failed} 个 FAIL'}") +if extra_failed: + failed += extra_failed +print("=" * 70) + +# ── 代理白名单测试（用户可配置：命中即直连，最优先，含 IPv4/IPv6/CIDR） ───── +print("\n" + "=" * 70) +print("代理白名单: 命中规则强制直连（优先于域名表/固定代理，支持 IPv4/IPv6/CIDR）") +print("=" * 70) + +wl_cases = [ + # (规则, 测试host, 期望命中) + ("example.com", "example.com", True), + ("example.com", "www.example.com", True), + ("example.com", "sub.www.example.com", True), + ("example.com", "example.org", False), + ("claude.ai", "claude.ai", True), # 域名表强制隧道，白名单优先 + ("localhost", "localhost", True), + ("8.8.8.8", "8.8.8.8", True), # IPv4 字面量 + ("8.8.8.8", "8.8.4.4", False), + ("10.0.0.0/8", "10.1.2.3", True), # IPv4 CIDR + ("192.0.2.0/24", "192.0.2.55", True), # IPv4 CIDR + ("2001:db8::/32", "2001:db8::1", True), # IPv6 网段 + ("fe80::1", "fe80::1", True), # IPv6 单地址 +] + +# 注入白名单规则（绕过 settings.json 直接改内存缓存；版本 +1 使匹配缓存失效）， +# 另加一条"禁用"的规则验证不参与匹配 +pp._WHITELIST = [{"value": "203.0.113.0/24", "enabled": False}] + \ + [{"value": r, "enabled": True} for r, _, _ in wl_cases] +wl_ver = pp._WHITELIST_VERSION + 1 +pp._WHITELIST_VERSION = wl_ver + +wl_failed = 0 +for rule, host, expect in wl_cases: + got = _is_whitelisted(host, wl_ver) + ok = got == expect + wl_failed += 0 if ok else 1 + print(f"规则 {rule:<16} host {host:<22} 命中={got} 期望={expect} -> {'PASS' if ok else 'FAIL'}") +got = _is_whitelisted("203.0.113.9", wl_ver) # 禁用的规则不应命中 +print(f"禁用规则 203.0.113.0/24 host 203.0.113.9 -> 命中={got} 期望=False -> {'PASS' if not got else 'FAIL'}") +wl_failed += 0 if not got else 1 + +# 白名单在三种模式下都强制直连（固定代理模式原本全部走隧道） +print("\n白名单在三种模式下的决策（命中白名单应为直连）:") +for mode_name, kw in MODES: + got = _decide_direct("www.example.com", **kw) + ok = got is True + wl_failed += 0 if ok else 1 + print(f" {mode_name:<16} www.example.com -> {PATH[got]} (期望 直连) {'PASS' if ok else 'FAIL'}") +# 白名单优先于域名表：claude.ai 在 pac_domains.json 中强制隧道，白名单命中应直连 +got = _decide_direct("claude.ai", bypass_lan=True, smart=True) +ok = got is True +wl_failed += 0 if ok else 1 +print(f" {'域名表冲突':<14} claude.ai -> {PATH[got]} (期望 直连，白名单优先) {'PASS' if ok else 'FAIL'}") +# 未命中白名单的境外域名仍走隧道（白名单不影响其他路由） +got = _decide_direct("google.com", bypass_lan=True, smart=True) +ok = got is False +wl_failed += 0 if ok else 1 +print(f" {'未命中白名单':<14} google.com -> {PATH[got]} (期望 隧道) {'PASS' if ok else 'FAIL'}") + +# 恢复空白名单，避免影响后续用例 +pp._WHITELIST = None +pp._WHITELIST_VERSION += 1 +if wl_failed: + failed += wl_failed +print("=" * 70) + +# ── IPv6-only 节点标注 ───────────────────────────────────── +# IPv6 专用节点只提供 AAAA 记录，本机没有 IPv6 出口时连不上： +# 客户端据此在节点菜单标注并置灰。判定三级：ipv6_only 字段 → +# remark 里的 "(IPv6)"（服务端 rrpp_client 现网唯一的线索）→ 域名解析兜底。 +print("\n" + "=" * 70) +print("IPv6 专用节点：识别（三级判定）与可达性") +print("=" * 70) + +ipv6_failed = 0 +has_v6 = pp._local_has_ipv6() +print(f"本机是否有全局 IPv6 地址: {has_v6}") + +# 用真实 nodes.json（test 目录）；没有则跳过 +_nodes_file = TEST_DIR / "nodes.json" +real_nodes = json.loads(_nodes_file.read_text("utf-8")) if _nodes_file.is_file() else [] +if real_nodes: + pp._IPV6_OK.clear() # 清缓存，做真实探测 + for n in real_nodes: + c = n.get("country") or "?" + io = pp._node_ipv6_only(n) + reach = pp._node_reachable(n) + mark = "[IPv6]" if reach else "[不可用：本机无 IPv6]" + print(f" [{c}] {n.get('domain'):<24} ipv6_only={str(io):<6} " + f"可达={str(reach):<6} 菜单文案: {n.get('domain')} ({c})" + + (f"　{mark}" if io else "")) + # ipv6_only 节点在本机无 IPv6 时必须判定为不可达，否则必须可达 + if io: + expect = has_v6 or reach # 本机有 IPv6 -> 应可达；无 IPv6 -> 不应可达 + ok = reach == expect + ipv6_failed += 0 if ok else 1 + print(f" IPv6 专用节点可达性: {reach} (期望 {expect}) " + f"{'PASS' if ok else 'FAIL'}") + else: + ok = reach is True + ipv6_failed += 0 if ok else 1 + print(f" 普通节点恒可达: {reach} {'PASS' if ok else 'FAIL'}") + # 缓存生效：再次判定不发新探测（结果一致） + cached = list(pp._IPV6_OK.items()) + if cached: + print(f" 可达性缓存: {cached}") + +# 三级判定：字段 → remark 里的 "(IPv6)" 标记 → 域名解析兜底。 +# 服务端 subscription?method=rrpp_client 只吐旧扁平格式，ipv6_only 字段被丢掉， +# 只在 remark 里留 "(IPv6)"（后端 countries_to_rrpp_client），故第 2 级是现网主路径。 +synthetic = [ + ({"domain": "example.com"}, False, "无标记（example.com 有 A 记录）"), + ({"domain": "example.com", "ipv6_only": False}, False, "ipv6_only=false"), + ({"domain": "example.com", "ipv6_only": True}, True, "ipv6_only=true"), + ({"domain": "example.com", "remark": "GlobalVM 350G (IPv6)"}, True, "remark 含 (IPv6)"), + ({"domain": "example.com", "remark": "Azure 15G"}, False, "remark 无 IPv6 标记"), + ({"domain": "example.com", "remark": "IPv6 专用机"}, True, "remark 含 IPv6（无括号）"), + ({"domain": "example.com", "remark": "some_ipv6_thing"}, True, "remark 小写 ipv6"), +] +for node, expect, desc in synthetic: + got = pp._node_ipv6_only(node) + ok = got == expect + ipv6_failed += 0 if ok else 1 + print(f" {desc:<24} _node_ipv6_only={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") +pp._IPV6_OK.clear() # 合成用例可能污染缓存 + +if ipv6_failed: + failed += ipv6_failed +print("=" * 70) + +# ── CONNECT authority 解析：IPv6 字面量带方括号 ────────────────── +print("\n" + "=" * 70) +print("CONNECT authority 解析（IPv6 字面量剥方括号）") +print("=" * 70) + +conn_cases = [ + ("CONNECT [::1]:443 HTTP/1.1", ("::1", 443)), + ("CONNECT example.com:443 HTTP/1.1", ("example.com", 443)), + ("CONNECT iam.aeolianchi.me:443 HTTP/1.1", ("iam.aeolianchi.me", 443)), + ("CONNECT [::1] HTTP/1.1", None), # 无端口 + ("CONNECT example.com HTTP/1.1", None), # 无端口 + ("CONNECT [fe80::1%25eth0]:8443 HTTP/1.1", ("fe80::1%25eth0", 8443)), + ("CONNECT 127.0.0.1:1080 HTTP/1.1", ("127.0.0.1", 1080)), +] +conn_failed = 0 +for line, expect in conn_cases: + got = pp._parse_connect_authority(line) + ok = got == expect + conn_failed += 0 if ok else 1 + print(f" {line:<45} -> {got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") +if conn_failed: + failed += conn_failed +print("=" * 70) + +# ── 上游代理认证注入：按完整头名精确匹配 ──────────────────────── +print("\n" + "=" * 70) +print("认证注入：仅精确的 authorization 头名阻止注入") +print("=" * 70) + +def _auth_lines(data: bytes): + head = data.split(b"\r\n\r\n", 1)[0] + return [l for l in head.split(b"\r\n")[1:] if l.lower().startswith(b"authorization:")] + +auth_cases = [ + # (描述, 请求, 期望)：数字=注入后 authorization 行数；"unchanged"=原样跳过 + ("无任何认证头", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nHost: iam.aeolianchi.me\r\n\r\n", 1), + ("Proxy-Authorization", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nProxy-Authorization: Basic xyz\r\n\r\n", 1), + ("已有 Authorization", b"GET http://example.com/ HTTP/1.1\r\nAuthorization: Bearer tok\r\n\r\n", "unchanged"), + ("小写 authorization", b"GET http://example.com/ HTTP/1.1\r\nauthorization: basic abc\r\n\r\n", "unchanged"), + ("X-Authorization 前缀", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nX-Authorization-Go: 1\r\n\r\n", 1), + ("无头空块", b"GET http://example.com/ HTTP/1.1\r\n\r\n", 1), +] +auth_failed = 0 +for desc, req, expect in auth_cases: + out = pp._inject_proxy_auth(req) + if expect == "unchanged": + ok = out == req + print(f" {desc:<22} 原样跳过 {'PASS' if ok else 'FAIL'}") + else: + lines = _auth_lines(out) + ok = len(lines) == expect and out.split(b"\r\n", 2)[1] == lines[0] + # 注入位置必须紧跟请求行（上游取首个 Authorization 时读到我们的认证） + print(f" {desc:<22} authorization 行数={len(lines)} (期望 {expect}) " + f"{'PASS' if ok else 'FAIL'}") + auth_failed += 0 if ok else 1 +if auth_failed: + failed += auth_failed +print("=" * 70) + +# ── 本程序 PAC 地址判定（启动清理只认自己的残留） ──────────────── +print("\n" + "=" * 70) +print("PAC 地址归属判定（_is_our_pac_url）") +print("=" * 70) + +pac_cases = [ + ("http://127.0.0.1:39453/proxy.pac", True), # 本程序 PAC 服务 + ("http://127.0.0.1:7897/pac?auth=xyz", False), # 其他代理软件（如 Clash） + ("http://127.0.0.1:1080/other.pac", False), # 路径不符 + ("http://192.168.1.1:8080/proxy.pac", False), # 非本机 + ("", False), +] +pac_failed = 0 +for url, expect in pac_cases: + got = pp._is_our_pac_url(url) + ok = got == expect + pac_failed += 0 if ok else 1 + print(f" {url:<42} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}") +if pac_failed: + failed += pac_failed +print("=" * 70) + +# ── DNS 分流 TTL 缓存 ──────────────────────────────────────── +print("\n" + "=" * 70) +print("TTL 缓存：cache_clear 兼容 + 缓存命中") +print("=" * 70) + +ttl_failed = 0 +assert hasattr(pp._is_china_ip, "cache_clear"), "cache_clear 缺失" +assert hasattr(pp._is_private_ip, "cache_clear"), "cache_clear 缺失" +assert hasattr(pp._is_whitelisted, "cache_clear"), "cache_clear 缺失" +pp._is_china_ip.cache_clear() +pp._is_private_ip.cache_clear() +pp._is_whitelisted.cache_clear() +a1, a2 = pp._is_private_ip("10.1.2.3"), pp._is_private_ip("10.1.2.3") +ok = a1 is True and a2 is True +ttl_failed += 0 if ok else 1 +print(f" 内网 IP 判定缓存命中: {a1}/{a2} {'PASS' if ok else 'FAIL'}") +b1, b2 = pp._is_china_ip("114.114.114.114"), pp._is_china_ip("114.114.114.114") +ok = b1 is True and b2 is True +ttl_failed += 0 if ok else 1 +print(f" 中国 IP 判定缓存命中: {b1}/{b2} {'PASS' if ok else 'FAIL'}") +pp._is_china_ip.cache_clear() +pp._is_private_ip.cache_clear() +print(f" cache_clear 后重查: {pp._is_china_ip('114.114.114.114')} " + f"{'PASS' if pp._is_china_ip('114.114.114.114') is True else 'FAIL'}") +if ttl_failed: + failed += ttl_failed +print("=" * 70)