# -*- coding: utf-8 -*- import sys, os, subprocess, time, ctypes, threading, functools, base64, bisect, re, json, uuid from pathlib import Path from PyQt5.QtWidgets import ( QApplication, QWidget, QLabel, QPushButton, QVBoxLayout, QHBoxLayout, QFrame, QMessageBox, QTextEdit, QCheckBox, QMenu, QDialog, QProgressDialog, ) from PyQt5.QtCore import Qt, QTimer, pyqtSignal from PyQt5.QtGui import QFont, QIcon, QPixmap, QImage, QColor from rrpp_ui.theme import C_TITLE, C_GREEN, C_YELLOW, C_RED, C_BG, C_CARD, C_BORDER, C_TEXT, C_SUBTEXT from rrpp_ui.widgets import NodeLabel, StatusDot, ToggleSwitch from rrpp_ui.dialogs import ( LoginDialog, LoginServices, SettingsDialog, SettingsServices, ThirdPartyDialog, ThirdPartyServices, WhitelistDialog, WhitelistServices, JumpDialog, JumpServices, ) from rrpp_core.routing_primitives import ( build_suffix_trie as _build_suffix_trie, merge_ip_ranges as _merge_ip_ranges, parse_domain_rules as _parse_domain_rules, suffix_trie_matches as _suffix_trie_matches, ) from rrpp_core.settings_store import JsonSettingsStore from rrpp_core.pac_data import ( atomic_write as _pac_atomic_write, fetch_json as _pac_fetch_json, no_proxy_opener as _pac_no_proxy_opener, read_version as _pac_read_version, validate_document as _pac_validate, ) from rrpp_core.forwarder_primitives import ( force_http_close as _force_http_close, inject_proxy_auth as _inject_proxy_auth_value, parse_connect_authority as _parse_connect_authority, parse_http_absolute as _parse_http_absolute, starts_with_http_method as _starts_with_method, ) from rrpp_core.node_secrets import NodeSecretsStore from rrpp_core.traffic import TrafficStats, format_bytes as _fmt_bytes from rrpp_core.node_api import ( NodeAuthError as _AuthError, b64url as _b64url, fetch_nodes_auth as _fetch_nodes_auth, ) from rrpp_core.third_party_subscriptions import ( fetch_nodes as _fetch_third_party_nodes, mask_url as _mask_subscription_url, ) from rrpp_core.network_primitives import ( is_private_host as _is_private_host, is_private_ipv4_int as _is_private_int, resolve_host_ips as _resolve_host_ips, ttl_cache as _ttl_cache, ) # GUI 窗口在 96 DPI / 100% 缩放下的逻辑尺寸；启用 Qt 高分屏缩放后， # Qt 会自动按 devicePixelRatio 放大/缩小，因此代码里保持逻辑尺寸不变即可。 _BASE_WINDOW_WIDTH = 340 _BASE_WINDOW_HEIGHT = 470 # 内容区 + 加大日志区 + 底部两行提示小字 # ── 路径 ────────────────────────────────────────────── # 已被打包为 EXE（PyInstaller 的 sys.frozen / Nuitka 的 __compiled__）： # BASE_DIR -> 内置数据目录：PyInstaller 用 _MEIPASS，Nuitka onefile 用 # __file__ 所在解压临时目录（无 _MEIPASS 时回退） # PERSIST_DIR -> EXE 所在目录，服务安装等需要持久化的操作都放这里。 # Nuitka onefile 的 sys.executable 指向 TEMP\onefile_* 解包子进程，不能用它 # 定位用户启动的原始 EXE；优先使用 __compiled__.containing_dir，旧版 Nuitka # 再回退到 sys.argv[0]（官方保证它指向原始 onefile 路径）。 def _frozen_executable_dir(): if "__compiled__" in globals(): compiled_dir = getattr(globals()["__compiled__"], "containing_dir", None) if compiled_dir: return Path(compiled_dir).resolve() return Path(sys.argv[0]).resolve().parent return Path(sys.executable).resolve().parent if getattr(sys, 'frozen', False) or "__compiled__" in globals(): BASE_DIR = Path(getattr(sys, "_MEIPASS", None) or Path(__file__).resolve().parent) PERSIST_DIR = _frozen_executable_dir() else: BASE_DIR = Path(__file__).resolve().parent PERSIST_DIR = BASE_DIR # ── 普通设置（config.json）──────────────────────────── # 版本号、更新链接等设置，缺失时用默认值兜底 _CONFIG = None def _load_config(): """加载 config.json；失败或缺失返回空字典（各设置用默认值兜底）。""" global _CONFIG if _CONFIG is not None: return _CONFIG _CONFIG = {} try: import json path = _data_path("config.json") if path is None: log("config.json 缺失，使用默认设置") return _CONFIG data = json.loads(Path(path).read_text("utf-8")) if not isinstance(data, dict): raise ValueError("config.json 顶层应为对象") _CONFIG = data log(f"已加载普通设置: {len(_CONFIG)} 项 (config.json)") except Exception as e: log_red(f"config.json 加载失败: {e}") return _CONFIG def _config_val(key, default=""): return _load_config().get(key, default) # ── 节点配置（官方 nodes.json + 第三方订阅）──────────────── # 官方节点和第三方节点分别保存；加载时才合并，避免登录更新覆盖第三方缓存。 _NODES = None _THIRD_PARTY_FILE = "third_party_subscriptions.json" _THIRD_PARTY_LOCK = threading.RLock() def _third_party_path(): return _USER_DIR / _THIRD_PARTY_FILE def _load_third_party_subscriptions(): """读取本机 DPAPI 加密的第三方订阅与最近一次成功缓存。""" with _THIRD_PARTY_LOCK: try: path = _third_party_path() if not path.is_file(): return [] data = json.loads(_nodes_decrypt(path.read_bytes())) records = data.get("subscriptions", []) if isinstance(data, dict) else [] return [record for record in records if isinstance(record, dict) and record.get("id")] except Exception as error: log_red(f"第三方订阅配置读取失败: {error}") return [] def _save_third_party_subscriptions(records): """原子保存加密订阅配置；URL 与节点认证信息不进入明文 settings.json。""" with _THIRD_PARTY_LOCK: _USER_DIR.mkdir(parents=True, exist_ok=True) text = json.dumps({"version": 1, "subscriptions": records}, ensure_ascii=False, indent=2) + "\n" path = _third_party_path() temporary = path.with_suffix(path.suffix + ".tmp") temporary.write_bytes(_nodes_encrypt(text)) os.replace(temporary, path) def _subscription_nodes(records): result = [] for record in records: source_id = str(record.get("id") or "") source_name = str(record.get("name") or "第三方订阅") nodes = record.get("nodes") or [] if not isinstance(nodes, list): continue for node in nodes: if not isinstance(node, dict) or not str(node.get("domain") or "").strip(): continue item = dict(node) item["_rrpp_source_id"] = source_id item["_rrpp_source_name"] = source_name item["_rrpp_source_kind"] = "third_party" result.append(item) return result def _tag_official_nodes(nodes): result = [] for node in nodes: if not isinstance(node, dict) or not str(node.get("domain") or "").strip(): continue item = dict(node) item["_rrpp_source_id"] = "official" item["_rrpp_source_name"] = "官方订阅" item["_rrpp_source_kind"] = "official" result.append(item) return result def _node_reference(node): """选中节点的稳定标识，避免不同订阅的同域名互相覆盖。""" if isinstance(node, dict) and node.get("_rrpp_source_kind") == "jump": return "jump|" + str(node.get("_jump_profile_id") or "") method = (node.get("methods") or {}).get("http") or {} port = method.get("port", node.get("port", "")) return "|".join(( str(node.get("_rrpp_source_id") or "official"), str(node.get("domain") or ""), str(port or ""), str(node.get("country") or ""), str(node.get("remark") or ""), )) def _load_nodes(): """加载并合并官方节点与每个第三方订阅的上次成功缓存。""" global _NODES if _NODES is not None: return _NODES official = [] try: path = _user_path("nodes.json") if path is not None: data = json.loads(_nodes_decrypt(Path(path).read_bytes())) if not isinstance(data, list): raise ValueError("nodes.json 顶层应为节点数组") official = _tag_official_nodes(data) except Exception as error: log_red(f"官方节点配置读取失败: {error}") third_party = _subscription_nodes(_load_third_party_subscriptions()) base_nodes = official + third_party _NODES = base_nodes + _jump_virtual_nodes(base_nodes) if not _SELECTED_RESTORED: _restore_selected_node(_NODES) if _NODES: log(f"已加载节点配置: 官方 {len(official)} 个，第三方 {len(third_party)} 个") return _NODES def _third_party_summaries(): """供 UI 展示，不向界面或日志暴露完整订阅 URL。""" return [{ "id": str(record.get("id")), "name": str(record.get("name") or "第三方订阅"), "url": _mask_subscription_url(record.get("url") or ""), "node_count": len(record.get("nodes") or []), "updated_at": str(record.get("updated_at") or "未更新"), "last_error": str(record.get("last_error") or ""), } for record in _load_third_party_subscriptions()] def _subscription_error_text(error): """保留格式错误提示，网络异常不回显可能含令牌的请求地址。""" if isinstance(error, ValueError): return str(error) return "下载失败，请检查网络和订阅地址" def _add_third_party_subscription(name, url): name, url = str(name or "").strip(), str(url or "").strip() if not name: raise ValueError("请填写订阅名称") if not url.lower().startswith(("https://", "http://")): raise ValueError("订阅链接必须以 http:// 或 https:// 开头") with _THIRD_PARTY_LOCK: records = _load_third_party_subscriptions() if any(str(record.get("url") or "") == url for record in records): raise ValueError("该订阅链接已添加") try: nodes = _fetch_third_party_nodes(url) except Exception as error: raise ValueError(_subscription_error_text(error)) from None records.append({ "id": uuid.uuid4().hex, "name": name, "url": url, "nodes": nodes, "updated_at": time.strftime("%Y-%m-%d %H:%M:%S"), "last_error": "", }) _save_third_party_subscriptions(records) _reload_nodes() log(f"第三方订阅已添加: {name}（{len(nodes)} 个节点）") def _refresh_third_party_subscription(subscription_id): with _THIRD_PARTY_LOCK: records = _load_third_party_subscriptions() record = next((item for item in records if str(item.get("id")) == str(subscription_id)), None) if record is None: raise ValueError("第三方订阅不存在") try: nodes = _fetch_third_party_nodes(str(record.get("url") or "")) record["nodes"] = nodes record["updated_at"] = time.strftime("%Y-%m-%d %H:%M:%S") record["last_error"] = "" _save_third_party_subscriptions(records) _reload_nodes() log(f"第三方订阅已更新: {record.get('name')}（{len(nodes)} 个节点）") except Exception as error: record["last_error"] = _subscription_error_text(error) _save_third_party_subscriptions(records) raise ValueError(record["last_error"]) from None def _remove_third_party_subscription(subscription_id): with _THIRD_PARTY_LOCK: records = _load_third_party_subscriptions() selected = [item for item in records if str(item.get("id")) == str(subscription_id)] if not selected: return _save_third_party_subscriptions([item for item in records if item not in selected]) _reload_nodes() log(f"第三方订阅已删除: {selected[0].get('name')}") def _sync_third_party_subscriptions(): """启动后更新第三方订阅；失败时保留上次成功的节点缓存。""" try: for record in _load_third_party_subscriptions(): try: _refresh_third_party_subscription(record.get("id")) except Exception as error: log(f"第三方订阅更新失败，继续使用本地缓存: {record.get('name')} ({_subscription_error_text(error)})") finally: _THIRD_PARTY_SYNC_DONE.set() _SELECTED_NODE = 0 _SELECTED_RESTORED = False _NODES_SYNC_DONE = threading.Event() # 官方订阅同步完成信号（启动校验等它） _THIRD_PARTY_SYNC_DONE = threading.Event() def _current_node(): """当前使用的节点：默认取列表第一个；无节点时返回 None。""" nodes = _load_nodes() if not nodes: return None i = _SELECTED_NODE return nodes[i] if 0 <= i < len(nodes) else nodes[0] def _selected_index(): """当前选中节点下标。""" return _SELECTED_NODE def _switch_node(index): """切换当前节点下标；相同或越界返回 False。""" global _SELECTED_NODE nodes = _load_nodes() if not nodes or not (0 <= index < len(nodes)): log_red("节点切换失败: 列表为空或下标越界") return False if index == _SELECTED_NODE: log("已在当前节点") return False _SELECTED_NODE = index _save_current_node(nodes[index]) return True def _restore_selected_node(nodes): """首次加载时按稳定节点标识恢复；兼容旧版仅按域名保存的设置。""" global _SELECTED_NODE, _SELECTED_RESTORED _SELECTED_RESTORED = True saved_ref = str(_settings_load().get("current_node_ref") or "") saved = str(_settings_load().get("current_node") or "") for i, n in enumerate(nodes): if saved_ref and _node_reference(n) == saved_ref: _SELECTED_NODE = i return if saved_ref.startswith("jump|"): # 跳板已失效时不能按其出口域名回退，否则可能无提示地改为直连出口节点。 _SELECTED_NODE = _first_reachable_index(nodes) if nodes: _save_current_node(nodes[_SELECTED_NODE]) return for i, n in enumerate(nodes): if saved and str(n.get("domain") or "") == saved: _SELECTED_NODE = i return _SELECTED_NODE = 0 def _save_current_node(node): """保存来源与节点标识；保留旧域名字段以兼容已有 settings.json。""" data = _settings_load() data["current_node"] = str((node or {}).get("domain") or "") data["current_node_ref"] = _node_reference(node or {}) _settings_save(data) _JUMP_MAX_HOPS = 3 def _roman_number(value): """将跳板序号转换为主页面使用的罗马数字。""" try: value = int(value) except (TypeError, ValueError): return str(value) if value <= 0: return str(value) result = [] for number, numeral in ( (1000, "M"), (900, "CM"), (500, "D"), (400, "CD"), (100, "C"), (90, "XC"), (50, "L"), (40, "XL"), (10, "X"), (9, "IX"), (5, "V"), (4, "IV"), (1, "I"), ): count, value = divmod(value, number) result.append(numeral * count) return "".join(result) def _jump_node_reference(node): """跳板专用稳定标识；备注、国家和凭据更新不应改变节点身份。""" source = str(node.get("_rrpp_source_id") or "official") domain = str(node.get("domain") or "").strip().rstrip(".").lower() return f"jump-node-v2|{source}|{domain}|{_node_http_port(node)}" def _jump_reference_key(ref): """兼容旧版 source|domain|port|country|remark 引用。""" parts = str(ref or "").split("|", 4) if parts[0] == "jump-node-v2": if len(parts) != 4: return None source, domain, port = parts[1:] elif len(parts) >= 3: source, domain, port = parts[:3] else: return None try: port = int(port) if port else 443 except ValueError: return None if not source or not domain or not 1 <= port <= 65535: return None return f"jump-node-v2|{source}|{domain.strip().rstrip('.').lower()}|{port}" def _jump_profile_records(): """读取跳板配置；只保存节点稳定标识，不保存节点凭据。""" raw = _settings_load().get("jump_profiles") if not isinstance(raw, list): return [] result = [] for profile in raw: if not isinstance(profile, dict): continue profile_id = str(profile.get("id") or "").strip() refs = profile.get("node_refs") if profile_id and isinstance(refs, list): result.append({"id": profile_id, "node_refs": [str(ref) for ref in refs if str(ref).strip()]}) return result def _jump_profile_save(records): data = _settings_load() data["jump_profiles"] = [ {"id": str(record["id"]), "node_refs": list(record["node_refs"])} for record in records ] _settings_save(data) def _jump_chain(node): """返回虚拟跳板节点对应的实际节点链，普通节点返回空列表。""" chain = (node or {}).get("_jump_chain") if isinstance(node, dict) else None return list(chain) if isinstance(chain, list) else [] def _terminal_node(node): chain = _jump_chain(node) return chain[-1] if chain else (node or {}) def _jump_route_text(node): chain = _jump_chain(node) return "→".join(str(item.get("country") or "?") for item in chain) if chain else "" def _jump_resolve_chain(refs, base_nodes): """优先精确匹配旧引用；备注变更后仅在稳定标识唯一时回退匹配。""" if not 2 <= len(refs) <= _JUMP_MAX_HOPS: return None by_old, by_stable = {}, {} for node in base_nodes: by_old.setdefault(_node_reference(node), []).append(node) by_stable.setdefault(_jump_node_reference(node), []).append(node) chain = [] for ref in refs: exact = by_old.get(ref, []) candidates = exact if len(exact) == 1 else by_stable.get(_jump_reference_key(ref), []) if len(candidates) != 1: return None chain.append(candidates[0]) if len({_jump_node_reference(node) for node in chain}) != len(chain): return None return chain def _jump_virtual_nodes(base_nodes): """将设置中的有效跳板配置映射为可切换的虚拟节点。""" result = [] for index, profile in enumerate(_jump_profile_records(), 1): chain = _jump_resolve_chain(profile["node_refs"], base_nodes) if chain is None: continue final = chain[-1] result.append({ "_rrpp_source_id": f"jump:{profile['id']}", "_rrpp_source_name": "代理跳板", "_rrpp_source_kind": "jump", "_jump_profile_id": profile["id"], "_jump_name": f"自定义跳板 {_roman_number(index)}", "_jump_chain": chain, "domain": str(final.get("domain") or ""), "port": final.get("port"), "country": _jump_route_text({"_jump_chain": chain}), "remark": "多跳链路", }) return result def _jump_summaries(): """供跳板管理 UI 使用，不暴露节点用户名或密码。""" base_nodes = [node for node in _load_nodes() if node.get("_rrpp_source_kind") != "jump"] result = [] for index, profile in enumerate(_jump_profile_records(), 1): chain = _jump_resolve_chain(profile["node_refs"], base_nodes) available = chain is not None and all(_node_reachable(node) for node in chain) route = "→".join(str(node.get("country") or "?") for node in chain) if chain else "" result.append({ "id": profile["id"], "name": f"自定义跳板 {_roman_number(index)}", "route": route or "节点已失效", "available": available, }) return result def _jump_available_nodes(): """返回可供配置的真实节点摘要和稳定标识。""" base_nodes = [node for node in _load_nodes() if node.get("_rrpp_source_kind") != "jump"] counts = {} for node in base_nodes: ref = _jump_node_reference(node) counts[ref] = counts.get(ref, 0) + 1 result = [] for node in base_nodes: ref = _jump_node_reference(node) if counts[ref] != 1 or not _node_reachable(node): continue result.append({ "ref": ref, "label": f"{node.get('country') or '?'} {node.get('domain') or '???'}:{_node_http_port(node)}", }) return result def _add_jump_profile(node_refs): refs = [str(ref).strip() for ref in node_refs if str(ref).strip()] if not 2 <= len(refs) <= _JUMP_MAX_HOPS: raise ValueError(f"跳板链路必须包含 2-{_JUMP_MAX_HOPS} 个节点") refs = [_jump_reference_key(ref) for ref in refs] if len(set(refs)) != len(refs): raise ValueError("同一条链路不能重复使用节点") available = {node["ref"] for node in _jump_available_nodes()} if any(ref not in available for ref in refs): raise ValueError("链路中包含当前不可用的节点") records = _jump_profile_records() if any([_jump_reference_key(ref) for ref in record["node_refs"]] == refs for record in records): raise ValueError("该跳板链路已存在") records.append({"id": uuid.uuid4().hex, "node_refs": refs}) _jump_profile_save(records) _reload_nodes() log(f"代理跳板已添加: {'→'.join(refs)}") def _remove_jump_profile(profile_id): profile_id = str(profile_id or "") if (_tls_forwarder and _node_reference(_current_node() or {}) == f"jump|{profile_id}"): raise ValueError("代理运行中，请先关闭代理再删除当前跳板") records = _jump_profile_records() remaining = [record for record in records if record["id"] != profile_id] if len(remaining) != len(records): _jump_profile_save(remaining) _reload_nodes() log("代理跳板已删除") # ── IPv6-only 节点可达性（客户端标注用）───────────────────── # IPv6-only 节点的域名只有 AAAA 记录（无 A 记录），本机没有 IPv6 出口时连不上。 # 判定三级：ipv6_only 字段 → 服务端 remark 里的 "(IPv6)" 标记 → 域名解析兜底。 # 服务端 subscription?method=rrpp_client 只吐旧扁平格式，ipv6_only 字段被丢掉， # 只留 remark 里的 "(IPv6)"（见后端 countries_to_rrpp_client），故第 2 级是现网主路径。 _IPV6_OK = {} # domain -> 本机能否连上该节点的 IPv6 地址 _IPV6_ONLY = {} # domain -> 该域名是否只有 IPv6 地址（解析兜底） _IPV6_PROBE_TIMEOUT = 4 # 单次 TCP 探测超时（秒） def _local_has_ipv6(): """本机是否有全局 IPv6 地址（排除回环/链路本地/唯一本地地址）。 没有则不可能连上 IPv6-only 节点，直接短路，省去探测。""" try: import socket import ipaddress for info in socket.getaddrinfo(socket.gethostname(), None, socket.AF_INET6): addr = info[4][0].split("%")[0] # 去掉 zone id（fe80::1%eth0） try: ip = ipaddress.ip_address(addr) except ValueError: continue if ip.is_loopback or ip.is_link_local or ip.is_private: continue return True except Exception: return False return False def _domain_ipv6_only(domain: str) -> bool: """域名是否「只有 IPv6 地址」：解析结果里有 IPv6 且没有 IPv4 才算。 解析失败按 False（当普通节点处理，不误伤）。结果缓存。""" if domain in _IPV6_ONLY: return _IPV6_ONLY[domain] import ipaddress only = False try: addrs = _resolve_host_ips(domain) if addrs: has6 = any(ip.version == 6 for ip in addrs) has4 = any(ip.version == 4 for ip in addrs) only = has6 and not has4 except Exception: only = False _IPV6_ONLY[domain] = only if only: log(f"节点 {domain} 解析结果只有 IPv6 地址，按 IPv6 专用节点处理") return only def _node_ipv6_only(node) -> bool: """节点是否 IPv6 专用。 三级判定： 1. ipv6_only 字段（服务端新格式；rrpp_client 输出里没有） 2. remark 里的 "(IPv6)" 标记（rrpp_client 现网唯一的权威线索） 3. 域名解析兜底：解析结果只有 IPv6 地址、没有 IPv4 """ if not isinstance(node, dict): return False if node.get("ipv6_only"): return True if "ipv6" in str(node.get("remark") or "").lower(): return True return _domain_ipv6_only(str(node.get("domain") or "").strip()) def _node_http_port(node) -> int: """节点的 HTTP 代理端口（不依赖当前选中节点，供可达性探测用）。""" m = (node.get("methods") or {}).get("http") or {} for v in (m.get("port"), node.get("port")): if v not in (None, ""): try: return int(v) except (TypeError, ValueError): pass return 443 def _node_reachable(node): """节点在本机是否可达：普通节点恒为 True；IPv6 专用节点需真的能建立 IPv6 连接 （按该节点自己的 HTTP 端口探测）。结果按域名缓存（探测一次）。""" if not isinstance(node, dict): return True if node.get("_rrpp_source_kind") == "jump": return all(_node_reachable(item) for item in _jump_chain(node)) domain = str(node.get("domain") or "").strip() if not _node_ipv6_only(node): return True if not domain: return False if domain in _IPV6_OK: return _IPV6_OK[domain] port = _node_http_port(node) ok = False addrs = sorted((ip for ip in _resolve_host_ips(domain) if ip.version == 6), key=str) if addrs and _local_has_ipv6(): import socket for ip in addrs: try: s = socket.socket(socket.AF_INET6, socket.SOCK_STREAM) s.settimeout(_IPV6_PROBE_TIMEOUT) s.connect((str(ip), port)) s.close() ok = True break except Exception: try: s.close() except Exception: pass _IPV6_OK[domain] = ok return ok def _reload_nodes(): """强制重新读取 nodes.json（菜单打开时调用，外部改动即时生效）。""" global _NODES, _SELECTED_NODE _NODES = None _IPV6_OK.clear() # 节点列表变了，IPv6 可达性/专用判定缓存作废 _IPV6_ONLY.clear() nodes = _load_nodes() if not nodes: _SELECTED_NODE = 0 return nodes # 跳板增删会改变虚拟节点下标；优先按稳定标识恢复，避免删除当前跳板后 # 误把同一个下标解释成另一条链路。 settings = _settings_load() saved_ref = str(settings.get("current_node_ref") or "") if saved_ref: match = next((i for i, node in enumerate(nodes) if _node_reference(node) == saved_ref), None) if match is not None: _SELECTED_NODE = match return nodes # 原节点（包括被删除的跳板）已不存在，旧下标不可再用于新列表。 _SELECTED_NODE = _first_reachable_index(nodes) _save_current_node(nodes[_SELECTED_NODE]) return nodes if not 0 <= _SELECTED_NODE < len(nodes): _SELECTED_NODE = 0 _save_current_node(nodes[0]) return nodes def _first_reachable_index(nodes): """第一个本机可达的节点下标；都不可达返回 0。供选中不可达节点时回退。""" for i, n in enumerate(nodes): if _node_reachable(n): return i return 0 def _rematch_saved_domain(): """按保存的来源与节点标识重匹配，兼容旧版域名选择记录。""" global _SELECTED_NODE nodes = _load_nodes() if not nodes: return False saved_ref = str(_settings_load().get("current_node_ref") or "") saved = str(_settings_load().get("current_node") or "") cur = nodes[_SELECTED_NODE] if 0 <= _SELECTED_NODE < len(nodes) else None if saved_ref and (not cur or _node_reference(cur) != saved_ref): match = next((i for i, n in enumerate(nodes) if _node_reference(n) == saved_ref), None) if match is not None: _SELECTED_NODE = match log(f"节点列表已更新，继续使用上次选择的节点 {nodes[match].get('domain')}") return True 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 switched cur = nodes[_SELECTED_NODE] if _node_reachable(cur): return switched idx = _first_reachable_index(nodes) old = str(cur.get("domain") or "") new = str(nodes[idx].get("domain") or "") _SELECTED_NODE = idx _save_current_node(nodes[idx]) if new and new != old: log_red(f"节点 {old} 在本机不可用，已自动切换到 {new}") return True def _node_val(key, default=""): node = _current_node() if not node: return default v = node.get(key, default) return v if v not in (None, "") else default # 空字符串视为未填写 def _endpoint_domain(): return str(_terminal_node(_current_node()).get("domain") or "???") def _http_method(): """节点 HTTP 代理方法参数（新版 nodes.json 的 methods.http）。 返回 (node, get)：get(key, default) 优先取 methods.http 下的键， 缺失时回退节点顶层——兼容旧扁平结构（port/username/password 直接挂在节点上）。 """ node = _terminal_node(_current_node()) m = (node.get("methods") or {}).get("http") or {} def get(key, default=""): v = m.get(key) if v in (None, ""): v = node.get(key) return default if v in (None, "") else v return node, get def _endpoint_port(): """HTTP 代理端口：methods.http.port（新格式）或节点顶层 port（旧格式）。""" _, get = _http_method() try: return int(get("port", 443)) except (TypeError, ValueError): return 443 def _endpoint_country(): node = _current_node() or {} return _jump_route_text(node) or str(node.get("country") or "") def _endpoint_label(): """主界面节点副标题；无节点时给出可操作的提示。""" node = _current_node() or {} if node.get("_rrpp_source_kind") == "jump": return f"{node.get('_jump_name') or '自定义跳板'} ({_endpoint_country()})" domain = _endpoint_domain() if domain == "???": return "无可用节点(请登录)" return f"{domain} ({_endpoint_country()})" def _proxy_auth_basic(): """正向代理 Basic 认证凭证：methods.http.{username,password}（新格式） 或节点顶层同名字段（旧格式）；用于转发器向上游代理注入 Authorization。""" _, get = _http_method() return base64.b64encode(f"{get('username')}:{get('password')}".encode()).decode() _tls_forwarder = None _pac_server = None PROXY_HOST = "127.0.0.1" PROXY_REG = r"HKCU\Software\Microsoft\Windows\CurrentVersion\Internet Settings" # _TLS_LOCAL_PORT / PROXY_PORT 定义在 _data_path 之后（取自 config.json local_port） # PAC 模式使用全量转发 PAC：所有流量送进本地转发器按 IP 智能分流 # （中国 IP 直连、其余走加密隧道）；域名表在 pac_domains.json。 _BUILT_IN_PAC = r'''function FindProxyForURL(url, host) { var proxy = "PROXY 127.0.0.1:1080"; return proxy; } ''' def _build_pac(proxy_spec: str) -> str: """生成全量转发 PAC，替换为当前模式对应的代理地址。""" return _BUILT_IN_PAC.replace( 'var proxy = "PROXY 127.0.0.1:1080";', f'var proxy = "PROXY {proxy_spec}";' ) _LAN_BYPASS_PATTERNS = ["10.*", "192.168.*", "169.254.*", "127.*"] _LAN_BYPASS_PATTERNS += [f"172.{i}.*" for i in range(16, 32)] _LAN_BYPASS_JS = ( " // LAN bypass: RFC1918 / link-local / loopback (resolves domain first)\n" " var _lanIp = dnsResolve(host);\n" " if (_lanIp && (" + " ||\n".join(f' shExpMatch(_lanIp, "{p}")' for p in _LAN_BYPASS_PATTERNS) + ")) {\n" " return \"DIRECT\";\n" " }\n" ) def _apply_lan_bypass(pac_content: str) -> str: """Inject LAN-direct rules into PAC so private IPs skip the local forwarder.""" marker = "function FindProxyForURL(url, host) {" if marker not in pac_content: return pac_content return pac_content.replace(marker, marker + "\n" + _LAN_BYPASS_JS, 1) def _proxy_override(bypass_lan: bool = False) -> str: """ProxyOverride for fixed-proxy mode; append common LAN ranges when enabled.""" if not bypass_lan: return "" parts = ["", "10.*"] parts += [f"172.{i}.*" for i in range(16, 32)] parts += ["192.168.*", "169.254.*", "127.*"] return ";".join(parts) def _start_pac_server(pac_content: str) -> int: """在 127.0.0.1 启动内存 PAC HTTP 服务，返回实际监听的端口。""" global _pac_server _stop_pac_server() import http.server import socketserver class _PACHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): if self.path == "/proxy.pac": body = pac_content.encode("utf-8") self.send_response(200) self.send_header("Content-Type", "application/x-ns-proxy-autoconfig") self.send_header("Content-Length", str(len(body))) self.end_headers() self.wfile.write(body) else: self.send_response(404) self.end_headers() def log_message(self, format, *args): pass # 避免在日志区输出大量 HTTP 访问日志 class _ReuseAddrTCPServer(socketserver.TCPServer): allow_reuse_address = True srv = _ReuseAddrTCPServer(("127.0.0.1", 0), _PACHandler) port = srv.server_address[1] threading.Thread(target=srv.serve_forever, daemon=True).start() _pac_server = {"server": srv, "port": port} return port def _stop_pac_server(): """停止本地 PAC HTTP 服务；shutdown() 唤醒 serve_forever 后优雅退出。""" global _pac_server if _pac_server: srv = _pac_server["server"] _pac_server = None # 不要直接 close socket：serve_forever 的 select 仍注册着它，Windows 上 # 对已关闭 socket 做 select 会抛 WinError 10038（非套接字操作）。 # shutdown() 让 select 在 poll_interval（0.5s）内返回并正常退出。 srv.daemon_threads = True # 请求处理线程不阻塞 serve_forever 退出 def _shutdown(): try: srv.shutdown() except Exception: pass try: srv.server_close() except Exception: pass t = threading.Thread(target=_shutdown, daemon=True) t.start() t.join(timeout=1.0) # ── 日志输出 ────────────────────────────────────────── # GUI 初始化后会把 _log_func 设为 MainWindow.log_signal.emit _log_func = None def log(msg: str): """输出一行日志到 GUI 日志区。""" if _log_func: _log_func(msg) def log_red(msg: str): """输出一行红色警告日志到 GUI 日志区。""" if _log_func: _log_func(f'{msg}') # ── 通用工具函数 ────────────────────────────────────── def _run(cmd, warn=True): """无窗口执行子进程，失败时记录警告日志。""" r = subprocess.run(cmd, capture_output=True, text=True, creationflags=getattr(subprocess, "CREATE_NO_WINDOW", 0)) if warn and r.returncode != 0: err = r.stderr.strip() if r.stderr else "" log(f"[警告] {' '.join(str(c) for c in cmd)} -> {r.returncode} {err}") return r def _notify_proxy_change(): """通知 WinINET 和顶层窗口代理设置已变更，让已运行程序立即重读。""" try: import ctypes ctypes.windll.Wininet.InternetSetOptionW(None, 39, None, 0) ctypes.windll.Wininet.InternetSetOptionW(None, 37, None, 0) result = ctypes.c_ulong() ctypes.windll.user32.SendMessageTimeoutW( 0xFFFF, 0x001A, 0, ctypes.c_wchar_p("InternetSettings"), 0x0002, 5000, ctypes.byref(result)) except Exception: 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， # 避免程序读到"固定代理已启用但服务器为空"的混乱状态 _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "0", "/f"]) # PAC 模式：清理固定代理遗留项，避免两种设置冲突 _run(["reg", "delete", PROXY_REG, "/v", "ProxyServer", "/f"], warn=False) _run(["reg", "delete", PROXY_REG, "/v", "ProxyOverride", "/f"], warn=False) _run(["reg", "add", PROXY_REG, "/v", "AutoConfigURL", "/t", "REG_SZ", "/d", pac_url, "/f"]) log("系统代理已设置 (PAC)") else: hp = f"{PROXY_HOST}:{PROXY_PORT}" _run(["reg", "add", PROXY_REG, "/v", "ProxyEnable", "/t", "REG_DWORD", "/d", "1", "/f"]) _run(["reg", "add", PROXY_REG, "/v", "ProxyServer", "/t", "REG_SZ", "/d", hp, "/f"]) _run(["reg", "add", PROXY_REG, "/v", "ProxyOverride", "/t", "REG_SZ", "/d", _proxy_override(bypass_lan), "/f"]) log(f"系统代理已设置: {hp}") _notify_proxy_change() def proxy_off(): """关闭系统代理并停止 TLS 转发器 / PAC 本地服务。""" log("正在关闭系统代理...") _restore_proxy_state() # 恢复用户原有代理设置，而不是简单清空 log("系统代理已关闭") _notify_proxy_change() _tls_stop() _stop_pac_server() # 打印本次转发流量统计 up, down = _traffic_snapshot() 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 # ── 本地认证转发器 ───────────────────────────────────── # ── 流量统计 ────────────────────────────────────────── # 所有代理流量都经过本地转发器（直连 / PAC 均如此），转发器里计数即可覆盖全部模式； # 每次"开启代理"时清零，即"本次连接"流量。 _TRAFFIC = TrafficStats() def _traffic_add(up=0, down=0): _TRAFFIC.add(up=up, down=down) def _traffic_reset(): _TRAFFIC.reset() def _traffic_rate(): return _TRAFFIC.rate() def _traffic_snapshot(): return _TRAFFIC.snapshot() def _tls_start(remote_host, remote_port=443, use_tls=True, bypass_lan=False, smart=False, quiet=False, jump_chain=None): """启动本地认证转发器：127.0.0.1: -> remote_host:remote_port（可选 TLS 包裹）。 smart=True 时启用 IP 智能分流（域名表/非中国 IP 走隧道，中国 IP 直连）。 quiet=True 时不输出启动日志（临时转发器用）。 """ global _tls_forwarder _tls_stop() import socket, ssl # 服务器证书为 Let's Encrypt 正式证书，默认上下文完整校验 ctx = ssl.create_default_context() # Windows 上 Hyper-V/WSL 的端口保留区间会动态变动，bind 可能瞬时被拒 # (WinError 10013/10048)；重试 3 次，间隔 1 秒，每次用全新 socket last_err = None svr = None for attempt in range(3): svr = socket.socket(socket.AF_INET, socket.SOCK_STREAM) try: svr.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) svr.bind(("127.0.0.1", _TLS_LOCAL_PORT)) break except OSError as e: last_err = e svr.close() svr = None if attempt < 2: log(f"端口 {_TLS_LOCAL_PORT} 绑定失败: {e}，1 秒后重试 ({attempt + 1}/3)") time.sleep(1) if svr is None: raise last_err svr.listen(32) svr.settimeout(1) _tls_forwarder = {"sock": svr, "running": True} proto = "TLS" if use_tls else "TCP" def _run(): while _tls_forwarder and _tls_forwarder.get("running"): try: client, addr = svr.accept() threading.Thread( target=_tls_relay, args=(client, remote_host, remote_port, ctx, use_tls, bypass_lan, smart, jump_chain), daemon=True, ).start() except socket.timeout: continue except Exception: break t = threading.Thread(target=_run, daemon=True) t.start() if not quiet: route = "→".join(str(node.get("country") or "?") for node in (jump_chain or [])) suffix = f" [{route}]" if route else "" log(f"本地转发器已启动 ({proto} 127.0.0.1:{_TLS_LOCAL_PORT} -> {remote_host}:{remote_port}{suffix})") def _inject_proxy_auth(data: bytes) -> bytes: """在 HTTP 代理请求头中注入 Authorization（如果尚未存在）。 nginx 的 auth_basic 校验 Authorization 而非 Proxy-Authorization， 因此转发器向上游发送 Basic 认证时使用 Authorization 头。 只按完整头名精确匹配（行首到冒号）：Proxy-Authorization 等含 "authorization" 字样的行不算已有认证，不能阻止注入；目标站自己的 Authorization（明文 HTTP 请求）才跳过。注入位置紧跟请求行， 保证上游（取首个 Authorization）读到的是我们的认证。 """ node, _ = _http_method() if not any(_node_proxy_credentials(node)): return data return _inject_proxy_auth_value(data, _proxy_auth_basic()) def _make_request_injector(): """为明文 HTTP 代理连接生成逐条注入认证的流式处理器。 keep-alive 连接上的后续请求同样需要 Authorization，不能只在首包注入。 请求头跨包时先缓冲，凑齐 \r\n\r\n（或超过 64KB）后统一处理； 请求体数据不以方法开头，原样透传。 注意：仅用于非 CONNECT 连接；CONNECT 隧道的后续流量是加密的，不能触碰。 """ pending = b"" def feed(data: bytes) -> bytes: nonlocal pending if pending: data, pending = pending + data, b"" if not _starts_with_method(data): return data if b"\r\n\r\n" in data or len(data) >= 65536: return _inject_proxy_auth(data) pending = data return b"" return feed @_ttl_cache(300, maxsize=2048) def _is_private_ip(host: str) -> bool: """判断目标是否命中常用内网、回环或链路本地地址（IPv4/IPv6）。""" return _is_private_host(host) # ── 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 而直连。 _CORE_FORCE_PROXY_SUFFIXES = frozenset(( "anthropic.com", "chatgpt.com", "claude.ai", "googlecdn.com", "oaistatic.com", "oaiusercontent.com", "openai.com", )) def _data_path(name: str): """数据文件路径：只取打包内置目录（config.json 随 exe 内置）。""" p = Path(BASE_DIR) / name return p if p.is_file() else None # 用户数据目录：Windows 惯例 %LOCALAPPDATA%（C:\Users\<用户>\AppData\Local）， # 非 Windows 环境回退用户主目录；测试可覆盖 _USER_DIR = (Path(os.environ["LOCALAPPDATA"]) if os.name == "nt" and os.environ.get("LOCALAPPDATA") else Path.home()) / "RRPP" _NODE_SECRETS = NodeSecretsStore(lambda: _USER_DIR) def _user_path(name: str): """用户数据文件路径：dev 模式（未打包）源码目录优先，方便本地测试； 打包后用户数据目录优先（首启模板/在线更新写在这里）。""" if not getattr(sys, 'frozen', False): p = _data_path(name) if p is not None: return p p = _USER_DIR / name if p.is_file(): return p return _data_path(name) # ── 本地设置（settings.json，用户数据目录，明文） ───────────── def _settings_path(): return _USER_DIR / "settings.json" _SETTINGS_STORE = JsonSettingsStore(_settings_path) def _settings_load(): """读取本地设置；缺失或损坏返回空字典。""" return _SETTINGS_STORE.load() def _settings_save(data): """本地设置原子写回。""" _SETTINGS_STORE.save(data) def _parse_port(text): """解析端口号；非法（非数字或超出 1-65535）返回 None。""" try: port = int(str(text).strip()) except (TypeError, ValueError): return None return port if 1 <= port <= 65535 else None def _port_val(): """代理转发端口：本地设置 > 内置 config.json > 默认 1080。""" port = _parse_port(_settings_load().get("local_port")) if port is not None: return port port = _parse_port(_config_val("local_port", 1080)) return port if port is not None else 1080 def _set_local_port(port): """更新代理转发端口并写回设置文件（下次开启代理生效）。""" global _TLS_LOCAL_PORT, PROXY_PORT _TLS_LOCAL_PORT = port PROXY_PORT = port data = _settings_load() data["local_port"] = port _settings_save(data) log(f"代理转发端口已保存: {port}（下次开启代理生效）") def _update_via_proxy(): """更新检查/下载是否走节点隧道（CONNECT）；设置项，默认关=直连仓库。""" return bool(_settings_load().get("update_via_proxy", False)) def _set_update_via_proxy(enabled): """写回「使用代理更新」设置（立即生效）。""" data = _settings_load() data["update_via_proxy"] = bool(enabled) _settings_save(data) log(f"更新方式已设为: {'走代理隧道' if enabled else '直连仓库'}") def _skip_update_check(): """是否跳过 PAC 数据与程序版本检查（设置项，默认关）。 不影响节点配置更新与客户端认证。""" return bool(_settings_load().get("skip_update_check", False)) def _set_skip_update_check(enabled): """写回「跳过检查更新」设置（立即生效）。""" data = _settings_load() data["skip_update_check"] = bool(enabled) _settings_save(data) log(f"检查更新已{'跳过' if enabled else '恢复'}（PAC 与程序版本；节点更新不受影响）") # ── 代理白名单（用户可配置，强制直连，settings.json 的 whitelist 键） ───────── # 每条规则 {"value": 域名/IP/CIDR, "enabled": 是否启用}，命中即直连（不经过代理）。 # 规则保存在内存缓存 _WHITELIST，连接建立时直接读取，不落盘回读； # _WHITELIST_VERSION 在保存时 +1，使按 (host, version) 键控的匹配缓存失效。 _WHITELIST = None # 内存缓存：规范化后的规则列表 _WHITELIST_VERSION = 0 # 规则版本号（保存时 +1，用于匹配缓存失效） def _whitelist_rules(): """白名单规则列表（内存缓存，首次读取 settings.json；保存时同步更新）。""" global _WHITELIST if _WHITELIST is None: raw = _settings_load().get("whitelist") if not isinstance(raw, list): raw = [] _WHITELIST = [{"value": str(r.get("value") or "").strip().lower(), "enabled": bool(r.get("enabled", True))} for r in raw if isinstance(r, dict) and str(r.get("value") or "").strip()] return _WHITELIST def _whitelist_save(rules): """保存白名单规则（规范化 + 去空白条目），更新内存缓存并记录日志。""" global _WHITELIST, _WHITELIST_VERSION clean = [{"value": _whitelist_normalize(r.get("value") or ""), "enabled": bool(r.get("enabled", True))} for r in rules if _whitelist_normalize(r.get("value") or "")] data = _settings_load() data["whitelist"] = clean _settings_save(data) _WHITELIST = clean _WHITELIST_VERSION += 1 log(f"代理白名单已保存: {len(clean)} 条（立即生效）") def _whitelist_normalize(entry: str) -> str: """条目规范化：去协议头/通配符/末尾点，统一小写。""" s = str(entry).strip().lower() if "://" in s: s = s.split("://", 1)[1] return s.rstrip("/").rstrip(".").lstrip("*.") def _whitelist_validate_entry(entry: str): """校验并规范化白名单条目；返回 (规范值, 错误)，错误非空表示非法。""" s = _whitelist_normalize(entry) if not s: return "", "内容不能为空" if any(ch.isspace() for ch in s): return "", "不能包含空格" import ipaddress if "/" in s: # CIDR（IPv4/IPv6 网段或单地址） try: ipaddress.ip_network(s, strict=False) except ValueError: return "", "IP/CIDR 格式不正确（如 10.0.0.0/8）" elif s.count(".") == 3 and all(p.isdigit() for p in s.split(".")): # 形如 x.x.x.x：必须是合法 IPv4，防止 300.1.2.3 这类误当域名 try: ipaddress.ip_address(s) except ValueError: return "", "IP 格式不正确" elif ":" in s: # IPv6 字面量（域名不含冒号） try: ipaddress.ip_address(s) except ValueError: return "", "IP 格式不正确" return s, "" def _ver_tuple(v): """版本号转数字元组：'v0.1.0' -> (0, 1, 0)，无法解析的段按 0 处理。""" parts = [] for p in str(v).strip().lstrip("vV").split("."): try: parts.append(int(p)) except ValueError: parts.append(0) return tuple(parts) def _version_newer(latest, current): """latest 是否比 current 新（逐段数字比较，位数不同补 0）。""" lt, ct = _ver_tuple(latest), _ver_tuple(current) n = max(len(lt), len(ct)) lt += (0,) * (n - len(lt)) ct += (0,) * (n - len(ct)) return lt > ct # 以下模块级取值需在 _data_path 定义之后执行 VERSION = _config_val("version", "999908162319") _TLS_LOCAL_PORT = _port_val() # 本地转发器监听端口（设置 > 内置 config > 1080） PROXY_PORT = _TLS_LOCAL_PORT # ── PAC 数据在线更新（CDN 直连，无需认证） ───────────── # 更新源地址取 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_direct_domains", "pac_ip_cn") def _pac_write(name, data): """PAC 数据原子写入用户数据目录（tmp+replace，写不坏旧文件）。""" _pac_atomic_write(_USER_DIR, name, data) 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/测试）不加密，读写明文；读取对密文/明文都兼容。 def _dpapi_protect(data: bytes) -> bytes: return _NODE_SECRETS.protect_bytes(data) def _dpapi_unprotect(data: bytes) -> bytes: return _NODE_SECRETS.unprotect_bytes(data) def _nodes_encrypt(text: str) -> bytes: return _NODE_SECRETS.encrypt_text(text) def _nodes_decrypt(blob: bytes) -> str: return _NODE_SECRETS.decrypt_bytes(blob) def _auth_save(username, password): _NODE_SECRETS.save_auth(username, password) def _auth_load(): return _NODE_SECRETS.load_auth() def _nodes_sync_remote(): """启动时用已保存凭据拉取最新节点配置（内容比对，变化才覆盖）。""" try: 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("节点配置已是最新") 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() # 无论成败都要放行启动校验 def _user_init_local(): """初始化用户数据目录；无节点时允许用户跳过登录后再添加第三方订阅。""" _USER_DIR.mkdir(parents=True, exist_ok=True) def _user_init_remote(base_url=None): """检查并更新 PAC 数据（后台线程）： 直接拉取两个数据文件（内含 version/date），按版本比对后更新；失败用本地副本。 节点配置的拉取走登录框（nodes_url 需 SSO 认证，见 LoginDialog）。 """ if _skip_update_check(): return # 设置跳过：不拉取也不打日志 base = (base_url or str(_config_val("pac_base_url", "") or "")).strip().rstrip("/") if not base: log("未配置 PAC 更新源（config.json 的 pac_base_url），使用本地数据") else: import json updated = failed = 0 for name in _PAC_FILES: try: data = _pac_fetch_json(f"{base}/{name}.json") if not _pac_validate(name, data): raise ValueError(f"{name}.json 数据格式不正确") rver = str(data.get("version") or "") rdate = str(data.get("date") or "") # 版本只与用户数据目录本地文件比对（回退文件仅用于读取，不参与更新判定） local = _USER_DIR / f"{name}.json" lver = _pac_read_version(local)[0] if local.is_file() else "" if rver: if rver > lver: _pac_write(name, data) _invalidate_pac_routing_cache(name) if name != "pac_direct_domains": log(f"PAC 数据已更新: {name}.json {rver} ({rdate})") updated += 1 # 否则已是最新，跳过 else: # 远端缺版本号：本地没有 -> 兜底下载；本地有 -> 按内容比对 new_bytes = json.dumps(data, ensure_ascii=False).encode("utf-8") if local.is_file() and local.read_bytes() == new_bytes: continue _pac_write(name, data) _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}") failed += 1 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/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 的后缀、精确和正则规则。 domain-list-community 的三种规则语义：普通域名匹配自身及子域； ``full:`` 仅匹配该主机；``regexp:`` 按正则匹配。旧逻辑把后两类 当成带冒号的普通后缀，导致它们永远不命中。 """ global _FORCE_PROXY_SUFFIXES, _FORCE_PROXY_EXACT, _FORCE_PROXY_REGEXES, _FORCE_PROXY_LOADED 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, exact, regexes, invalid_regexes = _parse_domain_rules( data.get("domains"), _CORE_FORCE_PROXY_SUFFIXES ) _FORCE_PROXY_SUFFIXES = suffixes _FORCE_PROXY_EXACT = exact _FORCE_PROXY_REGEXES = 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 = () finally: _FORCE_PROXY_LOADED = True return _FORCE_PROXY_SUFFIXES 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, invalid_regexes = _parse_domain_rules(data.get("domains")) _DIRECT_SUFFIX_TRIE = _build_suffix_trie(suffixes) _DIRECT_EXACT = exact _DIRECT_REGEXES = 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 @_ttl_cache(300, maxsize=16384) def _resolve_china_direct_ip(host: str): """返回可安全直连的中国 IPv4/IPv6，或 None。 同一地址族的所有解析结果都必须在对应中国 IP 段内，才允许直连；混合 CDN 记录不会因其中一个中国 IP 而误判。优先 IPv4 保持原有兼容性； 数据源提供 IPv6 CIDR 后，IPv6-only 目标同样可直连。返回本次解析得到 的地址，让转发器直接连接它，避免判定后再次 DNS 解析到不同的地址。 """ import ipaddress try: ip = ipaddress.ip_address(host) addresses = [ip] except ValueError: try: import socket addresses = [] for addr in socket.getaddrinfo(host, None, socket.AF_UNSPEC, socket.SOCK_STREAM): try: addresses.append(ipaddress.ip_address(addr[4][0])) except ValueError: pass except Exception: return None if not addresses: return None _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: """PAC 强制规则命中的域名走隧道（后缀 / full / regexp）。""" suffixes = _load_force_proxy_suffixes() exact = _FORCE_PROXY_EXACT or () regexes = _FORCE_PROXY_REGEXES or () if not suffixes and not exact and not regexes: return False h = host.rstrip(".").lower() return (any(h == s or h.endswith("." + s) for s in suffixes) or 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 _whitelist_entry_match(host: str, entry: str) -> bool: """单条白名单规则匹配：IP/CIDR 条目按解析地址比对，域名条目按后缀匹配。""" h = host.rstrip(".").lower() entry = entry.rstrip(".").lower() import ipaddress net = None # 只有含点/冒号才按 IP 解析：纯数字串（如 8080）会被 ip_network 误解析为 # 整型地址，应视为域名条目 if "." in entry or ":" in entry: try: net = ipaddress.ip_network(entry, strict=False) except ValueError: net = None if net is not None: return any(addr in net for addr in _resolve_host_ips(h)) # 域名条目：精确匹配或子域后缀匹配（与 pac_domains 同一规则） return h == entry or h.endswith("." + entry) @_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")] if not entries: return False return any(_whitelist_entry_match(host, e) for e in entries) def _decide_direct(host: str, bypass_lan: bool, smart: bool) -> bool: """判断连接是否直连。判定顺序（白名单最优先）： 白名单命中 -> 直连；强制代理域名 -> 隧道；明确直连域名 -> 直连； 内网 + 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 _node_proxy_credentials(node): method = (node.get("methods") or {}).get("http") or {} username = method.get("username") password = method.get("password") if username in (None, ""): username = node.get("username", "") if password in (None, ""): password = node.get("password", "") return str(username or ""), str(password or "") def _resolve_chain_address(host, port): """为跳板 CONNECT 解析 IP；CONNECT 行只使用 IP，不暴露节点域名。""" import socket infos = socket.getaddrinfo(host, port, socket.AF_UNSPEC, socket.SOCK_STREAM) infos.sort(key=lambda item: 0 if item[0] == socket.AF_INET else 1) if not infos: raise OSError("跳板地址解析失败") return infos[0][4] def _http_connect_over(transport, proxy_node, target_host, target_port): """在已有 TLS/分层 TLS 传输上建立 CONNECT，目标行仅含解析后的 IP。""" import base64 username, password = _node_proxy_credentials(proxy_node) auth_header = "" if username or password: token = base64.b64encode(f"{username}:{password}".encode()).decode("ascii") auth_header = f"Authorization: Basic {token}\r\n" target = _resolve_chain_address(target_host, target_port) target_authority = f"[{target[0]}]:{target[1]}" if ":" in target[0] else f"{target[0]}:{target[1]}" request = ( f"CONNECT {target_authority} HTTP/1.1\r\n" f"Host: {target_authority}\r\n" f"{auth_header}" "Connection: Keep-Alive\r\n\r\n" ).encode("ascii") transport.sendall(request) response = b"" while b"\r\n\r\n" not in response and len(response) < 65536: chunk = transport.recv(4096) if not chunk: break response += chunk marker = b"\r\n\r\n" end = response.find(marker) first_line = response.split(b"\r\n", 1)[0].decode("latin-1", "replace") if not (first_line.startswith("HTTP/1.1 200") or first_line.startswith("HTTP/1.0 200")): raise RuntimeError(f"跳板 CONNECT 失败: {first_line[:80]}") return response[end + len(marker):] if end >= 0 else b"" class _LayeredTlsTransport: """在已有 TLS 字节流上承载下一层 TLS，保留 socket-like 转发接口。""" def __init__(self, transport, ctx, server_hostname, initial=b""): import ssl self._transport = transport self._incoming = ssl.MemoryBIO() self._outgoing = ssl.MemoryBIO() self._pending = initial self._ssl = ctx.wrap_bio( self._incoming, self._outgoing, server_side=False, server_hostname=server_hostname) self._handshake() def _flush(self): data = self._outgoing.read() if data: self._transport.sendall(data) def _feed(self): if self._pending: data, self._pending = self._pending, b"" else: data = self._transport.recv(16384) if not data: raise ConnectionError("分层 TLS 下层连接已关闭") self._incoming.write(data) def _handshake(self): import ssl while True: try: self._ssl.do_handshake() self._flush() return except ssl.SSLWantReadError: self._flush() self._feed() except ssl.SSLWantWriteError: self._flush() def sendall(self, data): import ssl view = memoryview(data) while view: try: written = self._ssl.write(view) view = view[written:] self._flush() except ssl.SSLWantReadError: self._flush() self._feed() except ssl.SSLWantWriteError: self._flush() def recv(self, size): import ssl while True: try: return self._ssl.read(size) except ssl.SSLWantReadError: self._flush() self._feed() except ssl.SSLWantWriteError: self._flush() def recv_buffered(self, size): """只提取已在 TLS 各层缓冲的数据；不足一条记录时不等待网络。""" import ssl while True: try: return self._ssl.read(size) except ssl.SSLWantReadError: self._flush() if self._pending: data, self._pending = self._pending, b"" elif isinstance(self._transport, _LayeredTlsTransport): data = self._transport.recv_buffered(16384) elif hasattr(self._transport, "pending") and self._transport.pending(): data = self._transport.recv(16384) else: return None if data is None: return None if not data: return b"" self._incoming.write(data) except ssl.SSLWantWriteError: self._flush() def settimeout(self, value): self._transport.settimeout(value) def gettimeout(self): return self._transport.gettimeout() def wait_socket(self): return self._transport.wait_socket() if hasattr(self._transport, "wait_socket") else self._transport def close(self): self._transport.close() def _connect_via_jump(jump_chain, ctx): """最多三跳建立最终节点 TLS；中间 CONNECT 目标只发送解析后的 IP。""" import socket chain = list(jump_chain or []) if not 2 <= len(chain) <= _JUMP_MAX_HOPS: raise RuntimeError(f"跳板链路必须包含 2-{_JUMP_MAX_HOPS} 个节点") first = chain[0] first_host = str(first.get("domain") or "") first_port = _node_http_port(first) first_address = _resolve_chain_address(first_host, first_port) transport = socket.create_connection(first_address, 10) transport = ctx.wrap_socket(transport, server_hostname=first_host) for index, next_node in enumerate(chain[1:], 1): next_host = str(next_node.get("domain") or "") next_port = _node_http_port(next_node) leftover = _http_connect_over(transport, chain[index - 1], next_host, next_port) transport = _LayeredTlsTransport(transport, ctx, next_host, leftover) return transport def _connect_route(direct_target, allow_direct_fallback, remote_host, remote_port, ctx, use_tls, create_connection=None, jump_chain=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}") if jump_chain: if not use_tls: raise RuntimeError("跳板模式仅支持 TLS 上游") return _connect_via_jump(jump_chain, ctx), True 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, jump_chain=None): """处理一条客户端连接：内网(bypass_lan)/中国IP(智能分流)目标直接连接；其余走远端隧道。""" import socket, select client.settimeout(2.0) buf = b"" is_http = None is_connect = False injector = None direct_target = None allow_direct_fallback = False max_headers = 65536 try: while len(buf) < max_headers: try: chunk = client.recv(4096) except socket.timeout: break if not chunk: break buf += chunk if is_http is None: first_crlf = buf.find(b"\r\n") if first_crlf != -1: is_http = _starts_with_method(buf[:first_crlf] + b" ") if not is_http: break if b"\r\n\r\n" in buf: break if is_http and b"\r\n\r\n" in buf: is_connect = buf.startswith(b"CONNECT ") first_line = buf.split(b"\r\n", 1)[0].decode("latin-1", "replace") if is_connect: target = _parse_connect_authority(first_line) else: target = _parse_http_absolute(first_line) 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 remote = None try: remote, via_tunnel = _connect_route( direct_target, allow_direct_fallback, remote_host, remote_port, ctx, use_tls, jump_chain=jump_chain) 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: 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 via_tunnel: _traffic_add(up=len(buf)) except Exception as e: log(f"[转发器] 连接目标失败: {e}") try: client.close() except Exception: pass if remote is not None: try: remote.close() except Exception: pass return client.settimeout(None) remote.settimeout(None) # CONNECT 隧道承载 WebSocket / 推送等长连接，空闲属正常，不设超时； # 明文 HTTP 连接空闲 120 秒回收 idle_timeout = None if is_connect else 120 try: while True: if isinstance(remote, _LayeredTlsTransport): while True: data = remote.recv_buffered(65536) if data is None: break if not data: return client.sendall(data) if count_traffic: _traffic_add(down=len(data)) remote_wait = remote.wait_socket() if hasattr(remote, "wait_socket") else remote r, _, _ = select.select([client, remote_wait], [], [], idle_timeout) if not r: break if client in r: data = client.recv(65536) if not data: break if injector is not None: data = injector(data) if data: remote.sendall(data) if count_traffic: _traffic_add(up=len(data)) if remote_wait in r: data = remote.recv(65536) if not data: break client.sendall(data) if count_traffic: _traffic_add(down=len(data)) except Exception: pass finally: try: client.close() except Exception: pass try: remote.close() except Exception: pass def _tls_stop(): """停止 TLS 转发器监听线程。""" global _tls_forwarder if _tls_forwarder: _tls_forwarder["running"] = False try: _tls_forwarder["sock"].close() except Exception: pass _tls_forwarder = None def _ensure_forwarder(): """确保本地转发器在运行；未运行时启动临时全隧道转发器（静默），返回是否由本次启动。""" if _tls_forwarder: return False _tls_start(_endpoint_domain(), _endpoint_port(), use_tls=True, bypass_lan=False, smart=False, quiet=True, jump_chain=_jump_chain(_current_node())) return True def _release_forwarder(started_here): """停止由 _ensure_forwarder 启动的临时转发器；非本次启动则保持不动。""" if started_here: _tls_stop() def is_proxy_on(): """系统代理是否生效：固定代理已开启，或 AutoConfigURL 指向本地 PAC 服务。""" v = _reg_query_value("ProxyEnable") if v and "0x1" in (v[1] or ""): return True 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: 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 proxy_off() # 内部从备份恢复用户原有设置（无备份则清空） except Exception: pass # ── 图标 ────────────────────────────────────────────── # 程序图标 base64 编码（原 ico.ico），避免打包后依赖外部资源文件 _ICO_B64 = "iVBORw0KGgoAAAANSUhEUgAAAEAAAABACAYAAACqaXHeAAAfqElEQVR4nMV7CZClV3Xed++/v33p193Ts7UGabQOICSDAWGBCImR4zgmjFhjIsAhKYeUK46xUpgaERuIVRRVxCmnAGODkyrioYxEMJjFILEYKO0aZkaj2VozPTO9v37rv94lde7/3kz3aEYMRCRX9dTT/d77/3vOPct3vnN+hhdoaa0ZAIsxJi76+7VS4hbLwi0AbmyvdncsL3dbUZiVm62yu3VbE4yxjHM+ALAKYB7A0wAeB/AogMOMMbnhejYAxRhTL8S+2QskOD+/Sa2ZBl4F4J8BuAPADQAKaaqxcHYVGhqNRgmVauFKLp8CeAbAQwC+AuB7jLFkdF9rpAiN/19L55vI/93vT2qt/8Nw5eyTy8ePaiW0WVJIfeCpE/LggTnR6w2l1lqNXlop82O8xn+nz4jRz4vXM1rrfVrrnRv3MDqEn2uxn1NwTj9I+72zZyccW7zfn5h5H7g9BTFE++hBnQgtF9bBU6/GZq+aYtNT9fF3zW3Zld1Zj15k7nz0otUF8DkAn2CMnR7v6YVyiys/dS3+tRbZvFo4rpPuitY6zbKoJ+PTB/WpR36kH3v4kO4PY32J0/55F1lFdv4XKdpaiHu0PuhuiA+/uKUffNDcQMfxtVqKb403Mlw6nZ07/LgylqtifeyRR/TBQyd1OhL6BRJ+41KbFCGyR3WavuKFcIlLLrrgyOwhkvAdUmYdc+cszGQ6VFrG+uzBJ7RMh/rU8Tl9+Oi8Xu2FZA/6F7mklEqmibmJzNJUp+nvbtjvFSmB/bQPjC9E/q61/hiAewABZKmEZRl30GDoL5zDmcV1OM1pVKtlZKkA40DRd1EIPDj2ec95QZdWCkwJCa04HJ8hy/4SBw68j916a3YlcYE/78W1Zvfeey8pgA07i38FiHvC9bNShEMNy7GSfhdLzxxGe/4UMm1D2AGq9SqyjIQ3X0NvmKA3CE3wi5IUUr2wcYpxDs25RaKoOBRwnLux56avaq1LJPzYci/7fTyP8OP3hxn2Fx38i/X+MMNwzSkGPtzqBFQSIpMSdP8TJ5dQn56GlNmmy5Lgk80SskxhtTuEbXG4tgXXseA5NhzbxqDfQ6lcBufPu9fLL62hlQQj5aosg1d0kCY/WO3175xotQaETS5nCfxywj+UozrVT/B5Er4dIYNfdKRTJI0j6nWguQWvUMXyYhtuqQQhBZQ2+zFLKY2C78KxbPTCCDYJqIEklegNYrQHMY4dncN6LzQWE8Up4lRASGm+e8WLMTBuQYJBgDtIogy2e5vL+P0YZYbLxQT7Mpe0XseY6Cb6Y2UX7yThGYND2Zi5JXSjHrjFsHr6GbJBhE4VW7ZVjeSZkJBSQSoNMsxqyccwjpGmZCn8vG1Yto2o1zObrzXqRuD1fgg6RIod3LIgohCO66BaKSHw3JELaVicgTMO0qdtWRSfzHU6gwhSCEzWSo6IB1ml2Xh9f3nxLytTW945ymDipypAa23wfCfUbyu7uKeTIGOAk78JWK4H22sBFNPiIRaOHEHzxTtRKfgbr2I2SgohoS3GUS0FUGQaOrcMpQSYTFCfnETg2lBKEZ4Gt5lBPqQqEiwc9FGrVcxVkySDIAVYDDa3YFvMxOGxUsvFAForAuawLMdBGmfleu0dJ48cOcCuu+6+Bx980H7d6163SQnsUggvjrFL2XhCKhQzaZS7yVVIEN8GuEhx4uizuP7Fu8lfcGUrB3XD3jpsx4UXFMf3Hut4YwTBoL2KcnPyCq89+hYpVyhACyo95MqZU1xDvmbqmj0/1Pv3W+yuu+TlLIAxzlQn0p8t2Sh3Qkg6wItvYEGj6HCcW+2jNdU0wtP+SQckiDHJSyyRpYgGPZTrE1AigVuubbhx/p3N32TG1aQUsKwrB3kUDwxelRZjFrgm4ypU/0JrfTMZEsWDcRHFLzJ9mWT6PVUPt3ciiEsJT4KWXCBWQJoK1GulkQBAe3UVUp5X7uYvAYh6awhKFSiRwrZHvmvO/FJLn09zSlBm+VkWAwUICtbgFld+SbS2bb8WwAdH2eC83HwkPO1EdTq6zhk+Qv/mGvzirZFbFxxtjHiYaFhMw/U9816/2zW+bdv2eXOmlYQ9KK0IQcKh+OG4kCKG7XpXhMUoyOVu8zOqgHGjPDqZQuBZiys9usjvaR29KPfiHB/wsSLIJKwAv2tzTEWZUhVfGygzFoVk8iwNhwOR4JCZMEFoLEA0DNGYaG5yAcrNWRIanJAlQzj+yN8pZ1t5XN3ot5ddGxT6s2khV4DtuiyJyGbhD2J738j8zSb5yB+oympo4Hf6GfQwg5VKoOIp4wN08jbTKNgag3RUn4oMnpsLIQT5qGUEC4e98/eXxtRz3yXL4yM/5lBGKeOVhF1kabhB2AsCp1n2c9M2+UFwMNuBkMrqSSgw/lat9e6RzJwswOxkkOFdZRfNTEBanLNIMKQZQ8lTcLhG4AKJZJDauBcUBaYRvhdZBse1kcQxtLhwkpTQbDs3dU1aHAlCKY8EVVIiSxMkw3W4jg+dJXQxaPMSUFIhSRODCf5vFqViMkwloUoep1P7d+O3bDooCoCdGO+NKbyP3IKUFxmBGSpefjADkQulx5hgDF21hOPYEEoZkLIxXWbRGoRMEPXXYNsc2ioijAWsbADGPcikC5/cISMoS8ojUJNH8oSUSdZltnlxirzyRZiEYgIdttmT1G/TWv8hY6xnfD/K8PLAxQ1hSizPhsxgNKGNIPSfa9G/R2+alDfWcG7+BP3G78s0RGd10QAXZVfMywmIFXIh20cRd1cgBm2ouAeXFBkPgCQ2L5amgEiR9tZgiT7EYAXZYBlRZwEiIe70ymLCeH9SKFiuS9iFUQXt2nwikvhVs3f6n5D4Tc9AoAvhlr5Lfh84Gv0E6CcMgQMENhUedEIsN2X6nOsZBRAa7y2cMH/LpEKmHfDlo8aKdLgEnQ7BZIzC4ml44Sp00ocrpQEtOkkAcgHyeRFDRTHSKISlhrDdGmy7CM5dRL0VyCzaJKi46PcNKsj3kpEVMZCMUcYMHhUCbzIKoECggH9kqNYN+ZHiZMnViFMGoRkou/cSBs8mJZCL0Knn/m7grm1BZyHSNMKg0wYLSrDiLoLlNuRgGSzsQdOJDtfgNq+D4wMs7sAiy0kTsDSBphcJk2aI4xR22oc1fwQy6oDBMi4jlAVrFFfGK4uHY8e85CJs4tvaxK9UwYoUmNR4jda6zHsJXqSB6+M82BoXIxUVHQ2hGBKy7NEbaqQEm4gOR4FZY31puEEZSW8Bbm0n+oMuev0IttTgW29CMSjAp9jfWYaibDB7HRyvCjsLoYU0J08BkBTAsgwqSUxWCNrnEPQciBNPATIB18zUAzLubzphY5LPkyoNGaM1YpKFM5Yk0LaNmVjgVl708EsFH77IKEXA2IdnUxEDhBmVmRfCzhgXDIkQylJ4ruEizb0pxRVqU2BZF9pvQKyfgSpVoRtNOMxBYeZG6GIDen3R3MRyJ+CkFPhICTE5J5jMINOBMVm7vwQvToCX3gZn8mpkZw4bJcAuIuqvG6FNZtmoiOesfO9knVG2wb8BRbWM1HgV4eRbSAzKFCbf87zQGVK+v0TIpewgFRBHCdwgGP0t/6Bf3QqvVIHn+ghkH97y4dw0lQAvBHAmd8KubTOpDoUSrPoWqPVjkEnH7FVkPWRJZD5vRz30pnejx10khRrsqV1Iu2fA4i7iMDbutnFPl7KAcUlCCiClblykDKlxqz3IcGPZA8qOZFHGTdCLszz9PbemoRqfIez2DdFhov+m4odKVRuuz+BOzEDMnwSYg0xrnHrsAPrnFrBtzw2Y2NmgQgLW9CxcJaDWz0JTppCS7gDZO4vjT57C6dOPYBD14DZbePVdb8b07hfDXjiO9fkDiIsegomrck7wpyRH3/cg0zXDNYwWz/LK/FoK9DsjUp5grOwzEAJMBJn+pS9GKUUMeqju2DKqAC8IT8dIeMCBBHdKYJUmrEYN3/6zz+JrH7kPTZFi++w2vPlzn0Rp9sXQSQR7x40Qrg/ZOWe+I9I16FIBh77yNRz6wZM4S1km7uK7n/wkbn//+/HP7/kDTEQhonOnoSeuMmeZH/6l3YAOiICU51jodwaoT5QQR4qJjNActtlgmJACcCiv08044FgaqSLWZfOFHJdh+dQCtrSqeaGx6Z6jclZLWKRpZsNp1PA399yLB/7k45j8lTfia9e/G3GqcPWKh9uu1lAJYCVD2K0dEEEAuXQMKg3hzezCbXe/By/992V84PA0njy4gNcc/St8ed+9WD4xh/d96r/BOv440kEPXrFsrGAjREqiIbjtwHHIuXPBpmdaOHhoDra9DdVGAXFk3L5sK6XKNuPwHMZ6sTZCF12AZdpAX/qdhPcDhjMnFlD0LRTKJWOuF0NUYnmIDnfsCJZnY+HUGr73iT/FLb/1b/CKj92HuR9IfPtYigeXl3CbTEfFiknksEstUFtVL83BSTKsBi3MT12LO8pFPBTuRPjKm/D6ygA/+ev78eirX42Xv/VOdDpr8IiLFGITBxGGAwTFChznQhzwPAfXXLMND//4KIqVEm665WqILJfXJb+PUgoKDJliGKTMlL0mdyoNz2M4duAU5p6ZQ7FUxJmTc8gIrV3sHsnQBBsRd8G0DW17+JU33I6XvenN+Mz313F9KcK79lB5akOG2QUobSJrBqfQgjf7CrDGdqSugwcOS8ytCvzWDQo7mw6ueeWrMVMpY/mJpwCvYEjYOBpR7UTKjA+CgqzxYY1Op5tnrjBCd0AVaxmNiYpxG00WT+QGBUgy+bG2hCbkx1F2lQFEmWDwi77x3x6lB9s1NcEG/zAnSaansxhuZYsJcpM7ZrDj9tvQuOV6vKVr4bqdDbRKwNOHu0ilQuCOdjF2IaKyKBX5LuqzW/E7u8vYMU1sMzC/0kfQvA3HPvd5vOQ33mjAk6Uz9Lp9+GNDHBULBHyoloiTFEvnltFZHyBMCQ062H3NDNxSgH5oMAFsoZAlkrkbIz790yC/lKHsagObtsxOmb/T5warqxdsa3RnQoXxYA2l8gQ4d6DJt0tFVGbqqAzamGQe+ssp2qcitHSEwPZM1UcbveC/I9whJLbWXBw8t4hVVkeBZWhawHBpEa967zux/dYboZeX4UIgSkPwYEPHWUsQHUju2e8PUKiUUW5NoUZY0hoVcUohcBhiwWCnig+4hYYQphA6LxWBFbpsLx1ZgtQGG1g2AyfcP6rz8zTIkfY7sLSCX6oDyQCMORBRB7OvuBEOMtTSDjoLIQqBj2KtAt0fAEEB2iW8wUH/XQgmGqXAxcu2ltFtrwFhHzWPQ+9swt3zBqgFyhgeuI7BpYJb3TKyQgaZpaOsQOW8MIowdYvQyEwaZBCSG5hvGdqLYW0z35ibs7YZtUbMr1QIEeoteQw+1zBvjRQwBiDh8kkUSk2zeUKmhhoXAkmokIYpVJSi5QB+FkN0egTgjZuQyWtBm97MJWoSzAZaHtCyCcAPDC8hu30oNzBNGTbswUracPyysSYTh6SAtiyzX7omKWA0kmACJVkJbS8UDJ7Fhhwa8yTQOJESEcZEBre9AmvYz+kSxjDIODybwdbCNEMNM0Cnz3mOzS0XXrmV+3EeiUyQ0vDAOguwVGYqxLw3mMcWc0ut8uKGbc4oZIxaaohMIEkERCKRDWPTLFFL58DoJynFoozrE0NjvpclMSzLHZOd4/7tZrJdQZNyAZwj2Z+2aJaHWu4Wh9PvoPjsATAqTamAmJxFtG03HJlAa8+0tKhbkwfv3OSyNEbQ3G4+L0UEizmQcggpNLjvwCrZsL0qtIxzytrxoMns6MXoDEiledS+2BKtwDdluYqoSjQFPTExgO0YziCob4WkapAp2F5gWCnbzRspeVv7uWwSeYTPYWVSHSNe8zE6E804s6RE8dQh8LAPbTvQlgN/ZR5OdwW60cJw+24MhhGmp2rm9BfnnjZ+VK5PGuori/tQdAKBhzQZQGkXzHGQqHlYrg1mN3LGmDbveQAVUwZQEdGSgY+Yn5F2wUhBfiGPEY4DnQVk46AEr4dDA7udUgvxYBV+bQZaJIZFKpbzIo0O/+IymZJMYJAaR5Txx4ilfjiMVcZd7tjtdbA0QjS7x+TlZGIHyscfhSUSDKZmkSQSNiRc10O/s4ZCpYFKcxK95bNwygHiqI+g0IAWQySphuYENBVkdRfCbA2F2i4wyzO8CwlnKEnK1+Qq5BOkADXuLjPj59qS4EEALT1j5hRw6XuDhZ/A23G1MXlYDrjnI1pfgGLe+SrVcxzEm/pMgMsB1wKPyJBs/JC7wDMMOGpyKXU1tUZarCLc8SI4YQc8GiAr1qECF8O1NsrVMsJhH47nozoxbfCA0swQIUoyWK6PeDiAAPn12JQ5UiEQdlYgdAxG9LjlglG5SXyd7UJ1l0zWMEogK7FsqP4qspWTIPaEmF0WlKEcjkF7Hqpah12ZMX2HQnUSKuqjs76SWwXheQI5tr2pp0eZzbeVlpyzOMNK0cbDNjVCe4l+MABu6AZlxTjj5aOPQBaqsCmdyQy6UEUaKhOtHaeIcDhEY6JlGpHnsUGvj8AvEBmImCoqq2hOk6o7aAHba0ChgGRlFaI4gFtpgTu5SRu3CAcGe1heMa/utDINFBZUciuxLKThGuL+EEo5KLd2Ix6so1ibhkxjtNeXoIMWAkuYJqvxdcoASV4GE69ZdgjZQtkWuM3xI8ZYO2eAFR4gwlf4BR5P7wInWNpbBUtjiEIN6dQOrM6fxfR0A912B9V640JkZZZpdcXDnhFIiQiS2fnkxqhtTSavuAvqxWi7ADEUiJbmkbRPQVFgdFzwyR3gQRVZ3DXkqhl68AMwt4AsWsdg9QzCzhBCe/ACB1Ik4F7ecW6vnEPGKyhVavB9CsA5THccB0qlxu+JDyTkN0w1NURJpQ8YK6HpiRLwg26CucDRVw2ndinmBdwerCKFD7V9FguLHbTqJSRxAtcP8iYIpcCRGqguiNcXobftRhp1QO0UU0mO+IK8V5cjPuP/FGDhQ547CSs6DNmchioUDDCyggaSwQIsr2SKJKklkv4Q2i6BgAGXISzbQao5ZNjL71vZZcy9VPBM5kmGPRQqE7BsF5yoep33NQaJ1rbLrV6KXsXF3xorIW+j8VMOfD4gbAyl7OktGM7uQXLVNehFGnLQx8RkDWtrXdSb9QskyMjUXLoZWYoEev0BFA1C0MkzDknkKQUu83n66UASXqdURkxPqmAvLMI5dhT82EHq6+cg6vFvw4picNeH5oGJDYb6IFdIIyTDEOH6PHgwAS1TBAXfnLDjFgyIEpTGGYNL/IQIkVGhpyErDiPvup8xRoONhtU0uK3s4c9TqYZFF1aUSC1SBS41Fk/M45prZnDuzBJq9SoutUrVBpxiBdnScfDuoml6KIsiMUV2EpwMLTOpjvrtpAQCSqnoQhmgQhZCkbyOrL8Kr7ELbHK7yUTUQ6ROrbQsKG5DZn3EUc+0rtnEjZCWDxpTLhRyes5E/1IVYW/d/LtYLqHXWUckDHvNIwEV+PjkBUzAzCQVWcFZz+J/YVOEFEy6AceZuWXMtMjkGZ565MT5SY1N/X8iSmyOYq0Jp30GXmcZeu0MhIigyNTp1G3yRQHJ6HdKbS4y0UVqcbAshSpWoK+/GdZLXgPtOMj6C7B3vQyssdXMIujes1AUa5CiGEYodTqwRQT4ZShuwXH4qCOVd2tsxwPjtqlObdeH7zpI+h1Z98HTTH3VY+yJ0QidPM9rU5M0BP6LVKpX8TjvrEVa9jqY2rEFj3z/EFaWaC6Im37dpjVSRrHWQtutoD11DezZl8CWoTFZwuWSTlGGRhHSsiF1BuHXkG15ObpX7Qa78WZYrRmI/hJ0dyUfjNAMllcB4y6KxTI4YiirgAgCSToETwY5JLf8EXurN++nUkevvWo6ILVaRVtRj2KVLAT8Q+eBBs73AfOhgSJj5+JEf5RiTfv0vNxF5KUS+PEPj2H7bAvPtyQVI0EdhUoTjaktqE9tBQ1OS3AI5kMkkYG/dPqKZrkKTXDLRb+6HSIZmsIoW30WVmUqb6Fpkb9oEm36OjjEFskYcVBCdPUrMSxM0NiqwQux1CYTbbRKSoGOV0BnZZkAgWxt22p126uf8Rh7anz6uGhMTu3fv98qBtYnjs8tH9jWqtrFRlGunV3DzLYmduwkBRDz8lzy0UR6i6PWrJu0ZCZAgiKKgWto9kajjEJzC0Q6gFZDVJoNtFoTmKz5qBaLyLrr5hAt5sMu1qA3zhoazsSGpuKH8H6phdbUdlgEzihVkmtoDpGOelsGeOWFWrXZxOkTcwpCWJBqoTU988HRYMR5M+YXLJnpvXv30rBAVuDi7karIAikdLtDfeOereBmXmQjg3NBeFNmihTf+dFxdFYWwGzbbKpQbeLMs8/i0/f/BH/z3TPwHGBiagp+ZdJ8hyixSrkM8m596iSSpVOIw3Xjv9R3NB11j+r+AfzwHCDJpz3D8we+b1gsY30SiKIoV8i4WTTiKfygoA4/epDBdt5HwCcX9cIp8s3unAfErbNbH4cQ/xFe2UqTTARFCirs0ifPGNZW2/i3H/suPvSlFF94cBGIu2bzllvA/GqCD3z+LP7ukEZ96zXmb+fvPiJdFAlz+inwm15lWnEijcEDD9xTWFlcwDcenMN/+qaPL3zrDGrFvIdf8oB6ANRKNtysjaKdj9nl03lEdppmXnbtLTfb7fUePVfwFRqT2/j4Da3njF6ZyQn6YLnxybX5szdPTjbfNUijTArtbJ7gIFDDwGSKP/7cQRxot7Dnhib+fn4Bv/LDZ3DHHb9sPrtzdiumG12UnRRZew7exNXn2RtzP0JszRnEN70M5YltpmbI1o7j2eOr+NJjAg8dA3rZNoSJQHD8EP7VrwtMFIB/eGIeX3pwDpVmCyXdxfVXx3jJ9Ro7d0zn52oGETnBwr+/7dde//uj5xyeM8HFcan12tdKmqf7r39+9L1O0f7WRLPhKEVFwahtTDUeTYIy4MmnF/H4UhmzM3WIJEalPolPfTvEuTPzRsgtEwVMlQ3IR38QorNyLk+jG5RAJGxCSI/OLO3j89+LcO9fL+HLBz2UfA+1ACjaAmsDhU4vAnSGx59exP94ooBvHq3gy3O7cN83HPz2n57ARz/zMDrttqDaWil1AJa1d2Tyl3y+yL6U/KPReH3vXhMs3rS6uPp35WL5NmiVMc2cnHLLU893nuqD03iLysysrmspnA1n8On7j+AD76kZNnnbhGOmypzGVWZaLOy3USg30Cfc4FfQXls3Pr++dBof/+Icvn58C7bXA+z05lGgAopNouYO4JYTPHbgFFpBil+7fTe+cfgZJH4BBY/IVBcWL+HLh4cisZbtD749eEao4M5ikXWeb2yeX9ICLqRGChiD1Q7uBNPfJlaCGDytlBkhowrwqbkUgcPPT4YIqWkqC8+ei/DsgScMHp8ocThMwLcUmq0pM2HW77ZRqE5jfW0Njl9CtVrFA986iP/1iI1f3hkisDJwu4w47KNiD8xzB2n5Btz7vwW+/sgCdu+exa//UhWdHsUbbpojWSayrVsm7a4oHHj29Pzri0V2dv9+A/IuO2dnX+6NsRJG2utrre9UQnyW2847hZSUS9RPjnX5woCjWMojMW2EGikElnhxAm65gpW5o2hYA7gFz3STyfLLjUl01pYw98whNKe2wg98ZGmESr0J21ZYiSsosGVq7CO1prA8LJprFwuAt7OCT/94DTdcfxJCSGRZRtFIxRnT1XrTubqVfPO+u3e8lTG2TsLfddfmoPczKWCshH379pESKDn/Sy3EQW5ZHyXr+YdD60Jxh0o+RrOWQuc9dmL5HBnCq78I6elDOLUwgD9Zz9PbaFis1pyCTlPUmxNmooRy/T957Y14cu4JHD97HPOxZ3I+NWodLmEzSuc0daaReg3cd38fbDhAubRNCNh2s1bEnu3iE/vetvX3Nxzc8wpP64qeUPjwhz9MF8zHaW37TzjwumEvPHh8vWhzy2JCKrIJw2pp7qLEeuiHQyTDGLMvvRl+rWlmDOl2G9v49S3bzc98ilyjVCzhQ++4ClvsJSwlVcN9OjwnNMy3R7OrJSfFUlqS0qnp2WpmT9Xt0//0pfZv3vv2Lb+XY6DLPyBx8eK4wkWBkTQ6yqXfO/Ldp18xOxN8ZFvTHzK3bsWCM0/0hKNjLbmPpiF7Ulh+GRkLUPSey86Oh6xozqC73sbK2VNw69vwj197PYosonHKDRsgYEN2piU9hVYsBlatXlUv3cH++5+9p3Lr3b/aemDvXnpiLN/rlcpl42dcNG8/qh5pROMPHz3S+5/7fzz4g7kV/+1xL3a5TNFjJVl0HdQDOkPO1taHuGHHeOT9wrTfuKpUSsINiiiViia9vurlV+MNR4/hoTlFg9n0sCDJTCSp5RdKVtUPsW0C999x3dTH3vDy1iP3vBvYu19bX7yLySue2v95FUBrNGbK7voi+K3XsSMA7v7+seTj33yo/9srfestZ9LydLExg8mduzDsD7DQSWWzaia7Rj6weZtkCK4XIChXyCp0sVjSb7+jpZ/4XJcPVY15nmcVXAslL17f2eL3375n4lN33lx7mB5hw979lt6/V12Jv/9C1r59mtMmxr9rret/9IUzb/nOj098QWt9+sjxRb3rbV/V3/rh3HOffhw9ULm+uqoXl1af8/4Xvn5M/8YfnVx8/6cW7v/4l1bfPRhognnjO/F9P+WJsP+na5/W/PZ9+ZOl46W1rh05vnLHu/74Rx88eHThS1rrw1LK7sUK6LbX9eKiUUBfSnlUa/23Wuv/HMfxG7XutTZJuVdbL6Tg/wcpnXblom8x+wAAAABJRU5ErkJggg==" def load_icon(): """加载程序图标，base64 解码失败时返回纯色兜底图标。""" try: import base64 data = base64.b64decode(_ICO_B64) pixmap = QPixmap() pixmap.loadFromData(data) if not pixmap.isNull(): return QIcon(pixmap) except Exception: pass # 兜底：生成一个纯色方形图标 img = QImage(64, 64, QImage.Format_ARGB32) img.fill(QColor(C_TITLE)) return QIcon(QPixmap.fromImage(img)) def _login_services(): """为登录框提供明确的节点与凭据服务，UI 不直接依赖本模块全局变量。""" return LoginServices( load_icon=load_icon, load_auth=_auth_load, get_nodes_url=lambda: _config_val("nodes_url", ""), fetch_nodes=_fetch_nodes_auth, user_dir=lambda: _USER_DIR, encrypt_nodes=_nodes_encrypt, save_auth=_auth_save, auth_error=_AuthError, log=log, ) def _whitelist_services(): return WhitelistServices( load_icon=load_icon, load_rules=_whitelist_rules, validate_rule=_whitelist_validate_entry, save_rules=_whitelist_save, ) def _third_party_services(): return ThirdPartyServices( load_icon=load_icon, load_subscriptions=_third_party_summaries, add_subscription=_add_third_party_subscription, refresh_subscription=_refresh_third_party_subscription, remove_subscription=_remove_third_party_subscription, ) def _open_login_dialog(parent=None): """从启动流程或设置页登录官方订阅；成功后立即重载合并节点列表。""" dlg = LoginDialog(_login_services(), parent) if dlg.exec_() == QDialog.Accepted: _reload_nodes() return True return False def _logout_user_data(): """清除官方登录与节点缓存；第三方订阅保持不变。""" global _NODES for name in ("auth.json", "nodes.json"): try: (_USER_DIR / name).unlink(missing_ok=True) except Exception: pass _NODES = None log("已退出登录：官方凭据与节点配置已清除，第三方订阅保持不变") def _settings_services(): return SettingsServices( load_icon=load_icon, version=VERSION, get_port=lambda: _TLS_LOCAL_PORT, get_update_via_proxy=_update_via_proxy, get_skip_update_check=_skip_update_check, parse_port=_parse_port, set_port=_set_local_port, set_update_via_proxy=_set_update_via_proxy, set_skip_update_check=_set_skip_update_check, open_whitelist=lambda parent: WhitelistDialog(_whitelist_services(), parent).exec_(), open_jump=lambda parent: JumpDialog(_jump_services(), parent).exec_(), open_third_party=lambda parent: ThirdPartyDialog(_third_party_services(), parent).exec_(), has_login=lambda: _auth_load() is not None, open_login=_open_login_dialog, logout=_logout_user_data, ) def _jump_services(): return JumpServices( load_icon=load_icon, load_profiles=_jump_summaries, available_nodes=_jump_available_nodes, add_profile=_add_jump_profile, remove_profile=_remove_jump_profile, ) # ── 代理白名单弹窗 ───────────────────────────────────── # ── 主窗口 ──────────────────────────────────────────── class MainWindow(QWidget): log_signal = pyqtSignal(str) _op_done = pyqtSignal(str, bool) _update_done = pyqtSignal() _update_download_done = pyqtSignal(bool, str) _update_download_progress = pyqtSignal(int, int, int) _status_ready = pyqtSignal(object) _node_switched = pyqtSignal() # 启动校验时代替用户切换了节点（刷新标题栏） 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 # 防止状态轮询线程堆叠 self._update_progress_dialog = None # 把全局 log() 函数接到 GUI 日志区 global _log_func _log_func = self.log_signal.emit self.log_signal.connect(self._append_log) self._op_done.connect(self._after_op) self._update_done.connect(self._on_update_done) self._update_download_done.connect(self._on_update_download_done) self._update_download_progress.connect(self._on_update_download_progress) self._status_ready.connect(self._apply_status) self._node_switched.connect(self._on_node_switched) _user_init_local() # 首启初始化用户数据目录（节点模板），须在 init_ui 读节点之前 self.init_ui() # 每 2 秒刷新一次状态指示灯 self.timer = QTimer() self.timer.timeout.connect(self.refresh_status) self.timer.start(2000) self.refresh_status() # 启动时后台检查更新，完成前禁用开启按钮 threading.Thread(target=self._check_update, daemon=True).start() # 后台检查 PAC 数据更新（首次启动自动拉取到用户数据目录） threading.Thread(target=_user_init_remote, daemon=True).start() # 后台用已保存凭据同步最新节点配置（服务端改动下次启动生效） threading.Thread(target=_nodes_sync_remote, daemon=True).start() threading.Thread(target=_sync_third_party_subscriptions, daemon=True).start() # 后台校验上次选中的节点在本机是否还可用（IPv6 专用节点本机无 IPv6 时回退） threading.Thread(target=self._verify_node_thread, daemon=True).start() # 节点未配置时延迟弹出登录框（等窗口显示后） QTimer.singleShot(600, self._maybe_show_login) def _verify_node_thread(self): """等节点配置在线同步完成后，校验当前选中节点是否可用，必要时自动回退。 探测要解析 DNS 并尝试 TCP 连接，放在后台线程，避免卡住启动。 订阅同步可能较慢：等同步完成（最多 15 秒），同步后才校验， 否则服务端换过节点顺序时校验的是旧列表。""" _NODES_SYNC_DONE.wait(15) _THIRD_PARTY_SYNC_DONE.wait(15) try: if _verify_selected_node(): self._node_switched.emit() except Exception as e: log(f"节点可用性校验失败: {e}") def _on_node_switched(self): """节点被自动切换后刷新标题栏（在 UI 线程执行）。""" self.subtitle.setText(_endpoint_label()) def init_ui(self): """初始化主界面布局与样式。""" self.setWindowTitle("R.R.P.P.") self.setWindowFlags(self.windowFlags() & ~Qt.WindowContextHelpButtonHint) self.setWindowIcon(load_icon()) self.setFixedSize(_BASE_WINDOW_WIDTH, _BASE_WINDOW_HEIGHT) self.setStyleSheet(f""" QWidget {{ color: {C_TEXT}; font-family: "Microsoft YaHei", "Segoe UI", sans-serif; font-size: 13px; }} """) layout = QVBoxLayout(self) layout.setSpacing(14) layout.setContentsMargins(24, 20, 24, 20) # 标题 title = QLabel("Royirin Rosmontis Private Proxy") title.setFont(QFont("Microsoft YaHei", 16, QFont.Bold)) title.setStyleSheet(f"color: {C_TITLE};") title.setAlignment(Qt.AlignCenter) layout.addWidget(title) # 节点副标题（域名可点击切换：hover 下划线+手型提示，点击弹节点列表） node_row = QHBoxLayout() node_row.addStretch() prefix = QLabel("节点: ") prefix.setStyleSheet(f"color: {C_TITLE}; font-size: 12px;") node_row.addWidget(prefix) self.subtitle = NodeLabel(_endpoint_label()) self.subtitle.clicked.connect(self._show_node_menu) node_row.addWidget(self.subtitle) node_row.addStretch() layout.addLayout(node_row) # 连接方式选择区 pac_layout = QHBoxLayout() pac_layout.setSpacing(8) pac_layout.addStretch() self.cb_bypass_lan = QCheckBox("不代理内网 IP") self.cb_bypass_lan.setChecked(True) # 默认勾选：内网 IP 直连，不经过本地转发器 self.cb_bypass_lan.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") pac_layout.addWidget(self.cb_bypass_lan) self.cb_pac = QCheckBox("PAC 规则模式") self.cb_pac.setChecked(True) # 默认开启智能分流（域名表/中国IP判定已足够完善） self.cb_pac.setStyleSheet(f"color: {C_SUBTEXT}; font-size: 12px;") self.cb_pac.setToolTip("只代理海外服务，使用域名和解析 IP 匹配。") pac_layout.addWidget(self.cb_pac) pac_layout.addStretch() layout.addLayout(pac_layout) # 分隔线 line = QFrame() line.setFrameShape(QFrame.HLine) line.setStyleSheet("background-color: rgba(136, 153, 170, 80); max-height: 1px;") layout.addWidget(line) # 系统代理状态卡片 px_frame = QFrame() px_frame.setStyleSheet(f""" QFrame {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 8px; }} """) px_layout = QHBoxLayout(px_frame) px_layout.setContentsMargins(16, 12, 16, 12) px_label = QLabel("IE 代理") px_label.setStyleSheet(f"color: {C_TEXT}; font-weight: bold; background: transparent;") self.px_dot = StatusDot() self.px_status = QLabel("检测中...") self.px_status.setStyleSheet(f"color: {C_SUBTEXT}; background: transparent;") px_layout.addWidget(px_label) px_layout.addStretch() px_layout.addWidget(self.px_status) px_layout.addWidget(self.px_dot) layout.addWidget(px_frame) traffic_cards = QHBoxLayout() traffic_cards.setSpacing(8) self.traffic_up = QLabel("↑0 B/s") self.traffic_down = QLabel("↓0 B/s") self.traffic_total = QLabel("0 B") for card in (self.traffic_up, self.traffic_down, self.traffic_total): card.setStyleSheet(f""" QLabel {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 8px; color: {C_SUBTEXT}; font-size: 12px; padding: 10px 4px; }} """) card.setAlignment(Qt.AlignCenter) traffic_cards.addWidget(card, 1) layout.addLayout(traffic_cards) layout.addStretch() # 开启 / 关闭 代理按钮 self.btn_toggle = QPushButton("开启代理") self.btn_toggle.setMinimumHeight(40) self._set_toggle_button_style(is_start=True) self.btn_toggle.clicked.connect(self.on_toggle) self.btn_toggle.setEnabled(False) # 更新检查完成前禁用，避免旧版本被使用 layout.addWidget(self.btn_toggle) # 日志输出区 self.log_text = QTextEdit() self.log_text.setReadOnly(True) self.log_text.setMaximumHeight(140) self.log_text.setStyleSheet(f""" QTextEdit {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 6px; color: {C_SUBTEXT}; font-size: 11px; padding: 4px 6px; }} QScrollBar:vertical {{ width: 6px; background: transparent; }} QScrollBar::handle:vertical {{ background: {C_BORDER}; border-radius: 3px; min-height: 20px; }} """) layout.addWidget(self.log_text) hints = QWidget() hints_layout = QHBoxLayout(hints) hints_layout.setContentsMargins(0, 0, 0, 0) hints_layout.setSpacing(8) # 左侧两行小字（左对齐），右侧完全留给设置按钮 left_col = QVBoxLayout() left_col.setContentsMargins(0, 0, 0, 0) left_col.setSpacing(1) warn_label = QLabel("请勿二次分发此软件！") warn_label.setStyleSheet(f"color: {C_RED}; font-size: 11px;") left_col.addWidget(warn_label) privacy_label = QLabel("使用此软件将会收集并上报您的 IP 地址及部分设备信息") privacy_label.setStyleSheet("color: #c8ccd4; font-size: 9px;") left_col.addWidget(privacy_label) hints_layout.addLayout(left_col, 1) self.btn_settings = QPushButton("设置") self.btn_settings.setCursor(Qt.PointingHandCursor) self.btn_settings.setStyleSheet(f""" QPushButton {{ background-color: {C_CARD}; color: {C_SUBTEXT}; border: 1px solid {C_BORDER}; border-radius: 6px; padding: 3px 10px; font-size: 11px; }} """) self.btn_settings.clicked.connect(self._show_settings_dialog) hints_layout.addWidget(self.btn_settings, 0, Qt.AlignVCenter) layout.addWidget(hints) def closeEvent(self, event): """关闭窗口时自动清理代理和隧道。""" try: self.quit_app() finally: event.accept() # 即使清理异常窗口也必须关闭 # ── 开关操作 ── 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 {{ background-color: {C_GREEN}; color: #fff; border: none; border-radius: 8px; padding: 10px 24px; font-size: 14px; font-weight: bold; }} QPushButton:hover {{ background-color: #5ed978; }} QPushButton:pressed {{ background-color: #45b85a; }} QPushButton:disabled {{ background-color: #2a3a2e; color: #666; }} """) else: self.btn_toggle.setText("关闭代理") self.btn_toggle.setStyleSheet(f""" QPushButton {{ background-color: {C_RED}; color: #fff; border: none; border-radius: 8px; padding: 10px 24px; font-size: 14px; font-weight: bold; }} QPushButton:hover {{ background-color: #ff8787; }} QPushButton:pressed {{ background-color: #e05555; }} QPushButton:disabled {{ background-color: #3a1e1e; color: #666; }} """) def on_toggle(self): """根据按钮当前文字判断是开启还是关闭。""" 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() == "???": message = "未配置可用节点。请在设置中登录，或在「第三方」中添加订阅。" log_red(message) QMessageBox.information(self, "无法开启代理", message) return 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() 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, 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 _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() == "???": # 未配置节点时禁止开启：否则系统代理会指向连不通任何目标的转发器，导致断网 raise RuntimeError("未配置节点：请先通过 nodes_url 在线更新或 nodes_tool 导入节点配置") if not _node_reachable(_current_node()): # 当前选中节点是 IPv6 专用而本机没有 IPv6 出口：转发的流量全部连不上 raise RuntimeError("当前节点仅提供 IPv6 地址，本机没有 IPv6 出口，请在节点菜单切换其他节点") _traffic_reset() current_node = _current_node() jump_chain = _jump_chain(current_node) log("TLS 多跳模式" if jump_chain else "TLS 直连模式") ep = _endpoint_domain() log(f" 目标域名: {ep}") if jump_chain: log(f" 跳板路径: {_jump_route_text(current_node)}") # PAC 模式：全部流量进转发器，按域名表/解析 IP 智能分流 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, jump_chain=jump_chain) 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: log("不代理内网 IP：内网地址将直连") log("验证代理连通性...") _verify_proxy_connectivity(cancel_event=cancel_event) except Exception: try: # 连通性未通过（或用户取消）绝不能保留指向本地转发器的系统代理。 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) threading.Thread(target=self._stop_thread, daemon=True).start() def _stop_thread(self): """后台关闭代理：清理系统代理、停止 TLS/PAC 服务。""" try: log("正在关闭代理...") proxy_off() self._proxy_set_by_us = False log("已关闭") self._op_done.emit("stop", True) except Exception as e: log(f"[错误] 关闭失败: {e}") self._op_done.emit("stop", False) # ── 设置 ── def _show_settings_dialog(self): """打开设置弹窗；第三方订阅与登录状态可独立管理。""" SettingsDialog(_settings_services(), self, port_locked=self._proxy_set_by_us).exec_() self._refresh_subtitle() # ── 节点登录 ── def _maybe_show_login(self): """首次无节点时提示登录；用户可跳过并在主界面添加第三方订阅。""" if _endpoint_domain() == "???": self._show_login_dialog() def _show_login_dialog(self): """弹出登录框；成功后重载节点并刷新标题。""" if _open_login_dialog(self): self._refresh_subtitle() # ── 节点切换 ── def _show_node_menu(self): """弹出节点切换菜单（每次打开重新读取 nodes.json，外部改动即时生效）。""" if self._proxy_set_by_us: log("代理运行中，请先关闭代理再切换节点") return nodes = _reload_nodes() if not nodes: QMessageBox.information(self, "提示", "当前没有可用节点。\n" "请在右下角「设置」中登录，或在「第三方」中添加订阅。") return menu = self._build_node_menu(nodes) menu.exec_(self.subtitle.mapToGlobal(self.subtitle.rect().bottomLeft())) def _build_node_menu(self, nodes): """构建节点切换菜单：IPv6 专用节点加标注，本机无 IPv6 出口时置灰不可选。""" menu = QMenu(self) menu.setStyleSheet(f""" QMenu {{ background-color: {C_CARD}; border: 1px solid {C_BORDER}; border-radius: 8px; padding: 4px; }} QMenu::item {{ padding: 6px 18px; border-radius: 6px; color: {C_TEXT}; font-size: 12px; }} QMenu::item:selected {{ background-color: #e3e8ee; }} """) cur = _selected_index() for i, n in enumerate(nodes): is_jump = n.get("_rrpp_source_kind") == "jump" domain = n.get("_jump_name") if is_jump else (n.get("domain") or "???") country = _jump_route_text(n) if is_jump else (n.get("country") or "") remark = n.get("remark") or "" text = f"{domain} ({country})" if remark and remark != domain and not is_jump: text += f"　{remark}" source_name = n.get("_rrpp_source_name") or "官方订阅" text += f"　[{source_name}]" ipv6_only = _node_ipv6_only(n) reachable = _node_reachable(n) if ipv6_only: # 不可达时的文案由 tooltip 改为直接写在菜单项上，用户一眼可见 text += "　[IPv6]" if reachable else "　[不可用：本机无 IPv6]" if i == cur: text = "✓ " + text act = menu.addAction(text) if not reachable: # 本机无 IPv6 出口：置灰不可选，避免选中后连不上 act.setEnabled(False) act.setToolTip("不可用：本机无 IPv6") else: if ipv6_only: act.setToolTip("该节点仅提供 IPv6 地址（本机 IPv6 可用）") act.triggered.connect( lambda _=False, idx=i: self._on_node_selected(idx)) return menu def _on_node_selected(self, idx): """切换节点：更新标题。""" if not _switch_node(idx): return node = _current_node() self._refresh_subtitle() if node.get("_rrpp_source_kind") == "jump": log(f"已切换节点: {node.get('_jump_name')} ({_jump_route_text(node)})") else: log(f"已切换节点: {node.get('domain')} ({node.get('country')})") def _refresh_subtitle(self): """刷新节点副标题文字。""" self.subtitle.setText(_endpoint_label()) def _update_node_label(self): """代理运行中时提示禁止切换节点。""" if self._proxy_set_by_us: self.subtitle.setToolTip("代理运行中，请先关闭代理再切换节点") else: self.subtitle.setToolTip("点击切换节点") 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 == "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 创建前）到达的日志缓存 def _append_log(self, msg: str): """追加日志并自动滚动，最多保留 200 行；log_text 未创建时先缓存。""" if not hasattr(self, "log_text"): self._early_logs.append(msg) return for m in self._early_logs + [msg]: self._append_log_line(m) self._early_logs.clear() def _append_log_line(self, msg: str): self.log_text.append(msg) scrollbar = self.log_text.verticalScrollBar() scrollbar.setValue(scrollbar.maximum()) if self.log_text.document().blockCount() > 200: cursor = self.log_text.textCursor() cursor.movePosition(cursor.Start) cursor.select(cursor.BlockUnderCursor) cursor.removeSelectedText() cursor.deleteChar() # ── 状态刷新 ── def refresh_status(self): """触发一次后台状态轮询（不在 GUI 线程起子进程，避免周期性卡顿）。""" if self._status_busy: return self._status_busy = True # GUI 线程读取模式选择，传给后台线程 threading.Thread(target=self._poll_status, daemon=True).start() def _poll_status(self): try: px = is_proxy_on() except Exception: px = False self._status_ready.emit(px) def _apply_status(self, px_on_): """GUI 线程：根据轮询结果刷新状态指示灯、IE代理、流量卡片。""" self._status_busy = False # IE 代理状态 if px_on_: self.px_dot.set_on() self.px_status.setText("已开启") self.px_status.setStyleSheet(f"color: {C_GREEN}; background: transparent;") 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;") if not self._start_in_progress: self._set_toggle_button_style(is_start=True) # 流量卡片（始终更新，未连接时显示 0） px_on = is_proxy_on() if px_on: up_rate, down_rate, up_total, down_total = _traffic_rate() else: up_rate, down_rate, up_total, down_total = 0, 0, 0, 0 self.traffic_up.setText(f"↑{_fmt_bytes(up_rate)}/s") self.traffic_down.setText(f"↓{_fmt_bytes(down_rate)}/s") self.traffic_total.setText(_fmt_bytes(up_total + down_total)) def _check_update(self, manual=False): """后台检查服务器是否有新版本；manual=True 时无视「跳过检查更新」设置。""" # 先初始化结果，任何路径（含跳过）完成后 _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 # 设置跳过（仅自动检查）：不检查也不打日志 import json, urllib.request # 基于 Gitea 仓库 releases API 检查更新；地址只取 config.json 的 update_url url = str(_config_val("update_url", "") or "").strip() if not url: raise RuntimeError("config.json 未配置 update_url") log("正在检查更新...") if _update_via_proxy(): # 走节点隧道（CONNECT，设置「使用代理更新」开启时）；显式代理避免残留干扰 started_here = _ensure_forwarder() try: 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) data = opener.open(req, timeout=20).read().decode("utf-8-sig") finally: _release_forwarder(started_here) else: # 默认直连仓库；显式空代理避免系统/环境残留代理干扰 opener = _pac_no_proxy_opener() 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 附件链接。 # 附件可能有多个（发布流程附带 .sha256 边车），按名字精确匹配而不是 # 固定取第一个，避免附件顺序变化时下载到错文件。 self._latest_version = str(info.get("tag_name") or "") assets = info.get("assets") or [] 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}") self._latest_version = "" self._download_url = "" finally: self._update_done.emit() def _on_update_done(self): """更新检查完成后恢复按钮；发现新版本时弹窗询问是否下载。""" self.btn_toggle.setEnabled(True) if self._latest_version and _version_newer(self._latest_version, VERSION): reply = QMessageBox.question( self, "发现新版本", f"当前版本: {VERSION}\n最新版本: {self._latest_version}\n\n是否下载更新？", QMessageBox.Yes | QMessageBox.No, QMessageBox.Yes, ) if reply == QMessageBox.Yes: self._show_update_download_progress() threading.Thread(target=self._download_update, daemon=True).start() def _show_update_download_progress(self): """在 GUI 线程创建下载进度窗口；网络线程只通过 signal 回传状态。""" dialog = QProgressDialog("正在准备下载更新…", None, 0, 100, self) dialog.setWindowTitle("正在下载更新") dialog.setWindowModality(Qt.WindowModal) dialog.setMinimumDuration(0) dialog.setAutoClose(False) dialog.setAutoReset(False) dialog.setValue(0) dialog.show() self._update_progress_dialog = dialog def _on_update_download_progress(self, percent, downloaded, total): """把后台下载进度渲染到进度窗口。""" dialog = self._update_progress_dialog if dialog is None: return if total > 0: dialog.setRange(0, 100) dialog.setValue(max(0, min(100, percent))) dialog.setLabelText( f"正在下载更新… {percent}%（{_fmt_bytes(downloaded)} / {_fmt_bytes(total)}）") else: dialog.setRange(0, 0) dialog.setLabelText(f"正在下载更新… 已下载 {_fmt_bytes(downloaded)}") def _download_update(self): """后台下载新版本 EXE 到 EXE 所在目录。 落盘前校验：① 大小与 release API 记录一致（传输完整性）； ② 发布流程附带 .sha256 边车时校验 SHA-256（防链路被攻破/文件被替换）。 边车缺失（老 release）只做大小校验。 """ try: import hashlib import urllib.request, urllib.parse url = self._download_url if not url: self._update_download_done.emit(False, "下载链接为空") return # 对 URL 进行安全编码，兼容空格或已编码空格 url = urllib.parse.quote(url, safe=':/?#[]@!$&\'()*+,;=%') # 冻结程序必须放到当前 EXE 的目录，便于用户关闭旧版后就地替换； # 源码调试则仍写入源码目录，避免误把测试文件写到 Python 安装目录。 if getattr(sys, 'frozen', False) or "__compiled__" in globals(): download_dir = _frozen_executable_dir() else: download_dir = PERSIST_DIR save_path = download_dir / f"Royirin Rosmontis Private Proxy {self._latest_version}.exe" part_path = save_path.with_suffix(save_path.suffix + ".part") log(f"正在下载新版本: {url}") log(f"保存到: {save_path}") use_proxy = _update_via_proxy() started_here = _ensure_forwarder() if use_proxy else False try: if use_proxy: opener = urllib.request.build_opener(urllib.request.ProxyHandler( {"http": f"{PROXY_HOST}:{PROXY_PORT}", "https": f"{PROXY_HOST}:{PROXY_PORT}"})) else: opener = _pac_no_proxy_opener() if part_path.exists(): part_path.unlink() downloaded = 0 hasher = hashlib.sha256() last_percent = -1 with opener.open(urllib.request.Request(url), timeout=120) as resp, part_path.open("wb") as fp: try: response_size = int(resp.headers.get("Content-Length") or 0) except (TypeError, ValueError): response_size = 0 total = self._asset_size or response_size self._update_download_progress.emit(0, 0, total) log(f"更新下载进度: 0%（0 / {_fmt_bytes(total)}）" if total else "更新下载进度: 已开始") while True: chunk = resp.read(256 * 1024) if not chunk: break fp.write(chunk) hasher.update(chunk) downloaded += len(chunk) if total: percent = min(100, downloaded * 100 // total) if percent != last_percent: last_percent = percent log(f"更新下载进度: {percent}%（{_fmt_bytes(downloaded)} / {_fmt_bytes(total)}）") self._update_download_progress.emit(percent, downloaded, total) checksum_text = "" if self._checksum_url: checksum_url = urllib.parse.quote( self._checksum_url, safe=':/?#[]@!$&\'()*+,;=%') with opener.open(urllib.request.Request(checksum_url), timeout=120) as resp: checksum_text = resp.read().decode("ascii", "replace") finally: _release_forwarder(started_here) if downloaded < 1024: raise RuntimeError("下载文件过小，可能不是有效的 EXE") if self._asset_size and downloaded != self._asset_size: raise RuntimeError( f"下载不完整：实际 {downloaded} 字节，应为 {self._asset_size} 字节") if checksum_text: want = checksum_text.strip().split()[0].lower() got = hasher.hexdigest() if len(want) != 64 or want != got: raise RuntimeError("更新包校验失败（SHA-256 不匹配），已放弃") log("更新包 SHA-256 校验通过") part_path.replace(save_path) log(f"新版本已下载: {save_path}") self._update_download_done.emit(True, str(save_path)) except Exception as e: try: if 'part_path' in locals(): part_path.unlink(missing_ok=True) except Exception: pass log(f"下载更新失败: {e}") self._update_download_done.emit(False, str(e)) def _on_update_download_done(self, success, message): """下载完成后提示用户手动替换 EXE。""" dialog = self._update_progress_dialog self._update_progress_dialog = None if dialog is not None: dialog.close() dialog.deleteLater() if success: QMessageBox.information( self, "下载完成", f"新版本已下载到当前程序所在文件夹:\n{message}\n\n请关闭本程序后，用该文件替换当前运行的程序，然后重新启动。" ) else: QMessageBox.warning(self, "下载失败", f"无法下载新版本:\n{message}") def quit_app(self): """只清理自己设置的系统代理：用户或其他软件设置的代理不受影响。 本程序异常退出残留的代理设置由下次启动时 _cleanup_stale_proxy() 处理。 """ 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: pass # 清理失败也不能阻止退出，残留交给下次启动清理 QApplication.quit() # ── 程序入口 ────────────────────────────────────────── def _set_per_monitor_dpi_aware(): """在创建 QApplication 前尽量设置为每显示器 DPI 感知， 使程序在 2K/4K 高分屏以及多显示器不同缩放比例之间移动时， 由系统通知 WM_DPICHANGED 而不是被位图拉伸。 """ try: # Windows 10 1607+：Per Monitor V2（最优） ctypes.windll.user32.SetThreadDpiAwarenessContext(ctypes.c_ssize_t(-4)) return except Exception: pass try: # Windows 8.1+：Per Monitor ctypes.windll.shcore.SetProcessDpiAwareness(2) return except Exception: pass try: # 最低兜底：System DPI Aware ctypes.windll.user32.SetProcessDPIAware() except Exception: pass _SINGLE_INSTANCE_MUTEX = None def _acquire_single_instance(): """通过命名互斥体保证单实例运行；已有实例时返回 False。""" global _SINGLE_INSTANCE_MUTEX try: # 必须用 use_last_error 捕获真实错误码：直接调 GetLastError 可能被 # ctypes 内部调用重置（曾因此漏检已运行实例，导致双开互抢端口） kernel32 = ctypes.WinDLL("kernel32", use_last_error=True) kernel32.CreateMutexW.restype = ctypes.c_void_p _SINGLE_INSTANCE_MUTEX = kernel32.CreateMutexW(None, True, r"Local\RRPP_SingleInstance") return ctypes.get_last_error() != 183 # ERROR_ALREADY_EXISTS except Exception: return True # 互斥体创建失败时放行，避免把用户锁在门外 def main(): """主入口：检查管理员权限、依赖文件，然后启动 GUI。""" _set_per_monitor_dpi_aware() # 单实例：第二个实例直接退出（且不能执行下面的残留代理清理， # 否则会把第一个实例正在使用的系统代理清掉） if not _acquire_single_instance(): app = QApplication(sys.argv) QMessageBox.information(None, "提示", "程序已在运行中。") sys.exit(0) # 上次异常退出可能残留指向本地转发端口的系统代理（转发器已随进程消失）， # 不清理会导致全网不通，启动时先清掉 _cleanup_stale_proxy() # 任务栏图标：显式设置 AppUserModelID， # 避免 Windows 依赖 exe 图标缓存推断（冷缓存下任务栏会显示空白/默认图标） try: ctypes.windll.shell32.SetCurrentProcessExplicitAppUserModelID( "RRPP.RoyirinRosmontisPrivateProxy") except Exception: pass # Qt5 默认把缩放因子四舍五入到整数倍：2K 屏常见的 150% 会被取整为 2x， # 界面直接翻倍过大。改为 PassThrough，按实际比例（1.5x）缩放。 os.environ.setdefault("QT_SCALE_FACTOR_ROUNDING_POLICY", "PassThrough") try: QApplication.setHighDpiScaleFactorRoundingPolicy( Qt.HighDpiScaleFactorRoundingPolicy.PassThrough) except Exception: pass QApplication.setAttribute(Qt.AA_EnableHighDpiScaling, True) QApplication.setAttribute(Qt.AA_UseHighDpiPixmaps, True) app = QApplication(sys.argv) icon = load_icon() app.setWindowIcon(icon) app.setQuitOnLastWindowClosed(True) window = MainWindow() window.show() # 开机自启等场景下任务栏/Explorer 可能尚未就绪， # 延迟重设一次图标兜底，避免首次启动任务栏图标丢失 QTimer.singleShot(2000, lambda: window.setWindowIcon(icon)) # 退出必须彻底：此前点 X 后出现过进程残留（解释器收尾被残留线程/信号 # 竞态挂住），os._exit 直接终止进程——所有清理已在 quit_app 里完成 ret = app.exec_() os._exit(ret) if __name__ == "__main__": main()