RoyirinRosmontis/Private-Proxy
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 21 Wiki Activity

修复：- 系统代理设置保存/恢复：覆盖前备份原 ProxyEnable/ProxyServer/ProxyOverride/AutoConfigURL（内存+settings.json 防崩溃丢失），关闭/退出时恢复用户原有代理而不是清空；启动残留清理只认本程序精确地址（127.0.0.1:1080 精确匹配 + PAC URL 正则），不再误删其他代理软件（如 Clash）在 127.0.0.1 上的 PAC

Browse Source 
- CONNECT 目标解析支持 IPv6 字面量：剥掉方括号再拆 host:port，[::1]:443 不再解析成 "[::1]"（直连/白名单/内网绕过判定失效的根因）
- 更新包校验加固：release 附件按 RRPP_Lite_<tag>.exe 名字精确匹配（不再固定取第一个）；下载后与 API 记录的 size 比对；发布流程附带 .sha256 边车时校验 SHA-256
- 上游代理认证注入按完整头名精确匹配：Proxy-Authorization 不再误判为已有 Authorization；注入位置紧跟请求行保证上游读到我们的认证
- 远程同步后节点按 settings 记录的域名重匹配：服务端调整节点顺序时不再静默连到别的节点；启动校验等待订阅同步完成（最多 15 秒）后再执行
- DNS 分流判定缓存加 300 秒 TTL：域名换 IP 后不再长期沿用旧的直连/隧道判断（_is_private_ip/_is_china_ip/_is_whitelisted 换 _ttl_cache，保留 cache_clear 兼容）
- is_proxy_on 状态判定同步收紧为本程序 PAC 地址（其他软件的 PAC 不再让状态灯误亮）
This commit is contained in:
RoyirinRosmontis
2026-09-11 01:56:04 +08:00
parent cb3552f595
commit 23ba368737
3 changed files with 681 additions and 80 deletions
Download Patch File Download Diff File Expand all files Collapse all files
lite/test/check_node_menu.py
+29
View File
@@ -72,4 +72,33 @@ else:
else:
(pp._USER_DIR / "settings.json").write_bytes(_bak)
# ── 服务端调整节点顺序：按域名重匹配而不是死守下标 ──────────────────
print("\n服务端调序（_rematch_saved_domain）:")
if len(nodes) < 2:
print(" 节点不足 2 个，跳过")
else:
import json
dom_old = nodes[0].get("domain")
ordered = [nodes[1], nodes[0]] + nodes[2:] # 模拟服务端交换了前两个节点
pp._NODES = ordered
# 设置里记录的是原下标 0 的域名，重匹配后下标应跟随域名（新列表里是 1）
pp._settings_save({**pp._settings_load(), "current_node": dom_old})
switched = pp._rematch_saved_domain()
ok = switched and ordered[pp._selected_index()].get("domain") == dom_old
print(f" current_node={dom_old} 调序后跟随域名 -> 下标 {pp._selected_index()} "
f"{'PASS' if ok else 'FAIL'}")
# 已是正确域名时再跑一次：不应重复切换
ok2 = pp._rematch_saved_domain() is False
print(f" 再次重匹配不重复切换 {'PASS' if ok2 else 'FAIL'}")
# 域名已不在列表（节点被移除）：不切换，保持现状
pp._settings_save({**pp._settings_load(), "current_node": "gone.example.com"})
ok3 = pp._rematch_saved_domain() is False
print(f" 节点被移除时不动 {'PASS' if ok3 else 'FAIL'}")
pp._NODES = None # 还原缓存，后续用例重读文件
# 还原 settings.json，避免测试污染
if _bak is None:
(pp._USER_DIR / "settings.json").unlink(missing_ok=True)
else:
(pp._USER_DIR / "settings.json").write_bytes(_bak)
print("OK")
lite/test/test_pac_routing.py
+389
View File
@@ -0,0 +1,389 @@
# -*- coding: utf-8 -*-
"""PAC（智能分流）模式路由决策测试。
核心问题：PAC 模式下，目标域名解析为国外 IP 时，是否强制走隧道代理？
判定链（proxy_lite._decide_direct）：
0. 代理白名单命中 -> 强制直连（最优先，PAC/固定模式均生效）
1. pac_domains.json 域名表命中 -> 强制隧道（优先于 IP 判定）
2. 内网 IP 且勾选 bypass_lan -> 直连
3. 智能分流：中国 IP -> 直连；其余（国外 IP）-> 走隧道
直接 import proxy_lite 调用真实决策函数，数据用真实的 pac_ip_cn.json /
pac_domains.json；PyQt5 用桩替代（决策逻辑不依赖 Qt）。
"""
import sys
from pathlib import Path
TEST_DIR = Path(__file__).resolve().parent
LITE_DIR = TEST_DIR.parent
sys.path.insert(0, str(TEST_DIR)) # 假 PyQt5 桩
sys.path.insert(0, str(LITE_DIR))
import json
import socket
import proxy_lite as pp
from proxy_lite import _decide_direct, _is_china_ip, _is_force_proxy, _is_private_ip, _is_whitelisted
# PAC 数据已不再随项目保留：从 CDN 拉取到临时用户目录（真实数据端到端）
import tempfile
pp._USER_DIR = Path(tempfile.mkdtemp()) / "RRPP"
pp._user_init_remote(base_url="https://cdn.rrrosmontis.icu/RRPP/PAC/")
pp._load_cn_ranges()
pp._load_force_proxy_suffixes()
if not pp._CN_RANGES:
print("警告: 未能获取 pac_ip_cn.json（CDN 不可达？），中国 IP 判定将失效")
print(f"中国IP段: {len(pp._CN_RANGES or [])} 段, 强制域名表: {len(pp._FORCE_PROXY_SUFFIXES or [])} 条\n")
def resolved_ips(host):
try:
return sorted({a[4][0] for a in socket.getaddrinfo(host, 443, socket.AF_INET)})
except Exception:
return []
def mark(ip):
if _is_private_ip(ip):
return "内网"
if _is_china_ip(ip):
return "中国IP"
return "国外IP"
PATH = {True: "直连", False: "隧道"}
MODES = [
("PAC+bypass_lan", dict(bypass_lan=True, smart=True)), # 默认勾选状态
("PAC无bypass_lan", dict(bypass_lan=False, smart=True)),
("固定代理模式", dict(bypass_lan=False, smart=False)), # 全部走隧道
]
# (名称, host, 期望 PAC+bypass_lan 决策: True=直连 / False=隧道)
cases = [
("境外域名 google.com", "google.com", False),
("境外域名 github.com", "github.com", False),
("境外域名 youtube.com", "youtube.com", False),
("域名表强制 claude.ai", "claude.ai", False),
("域名表强制 *.google 后缀", "antigravity.google", False),
("国内域名 www.baidu.com", "www.baidu.com", True),
("国内域名 www.qq.com", "www.qq.com", True),
("内网 localhost", "localhost", True),
("内网 IP 192.168.1.1", "192.168.1.1", True),
("境外 IP 8.8.8.8", "8.8.8.8", False),
("国内 IP 114.114.114.114", "114.114.114.114", True),
]
failed = 0
for name, host, expect in cases:
ips = resolved_ips(host)
force = _is_force_proxy(host)
marks = f"[{','.join(mark(ip) for ip in ips)}]" if ips else "[解析失败]"
row = f"{name:<22} {host:<18} {str(ips) if ips else '无IP':<45} {marks}"
results = []
for mode_name, kw in MODES:
got = _decide_direct(host, **kw)
results.append(f"{mode_name}={PATH[got]}")
pac = _decide_direct(host, **MODES[0][1])
ok = pac == expect
failed += 0 if ok else 1
print(f"{row}\n {'域名表命中' if force else '域名表未命中':<10} | {' '.join(results)} | 期望 {PATH[expect]} -> {'PASS' if ok else 'FAIL'}")
print(f"\n{'='*70}")
print(f"结论: {'全部 PASS' if not failed else f'{failed} 个 FAIL'}")
print("=" * 70)
print("PAC 模式(智能分流)下: 域名解析结果全部为国外 IP -> 走隧道代理 ✓"
if not failed else "存在 FAIL，请检查上方明细")
# ── 附加测试：不在域名表的被墙境外站 ─────────────────────
# 这类站点不能靠域名表兜底，必须仅凭「解析结果为国外 IP」强制走隧道，
# 否则会尝试直连 -> 被墙。
print("\n" + "=" * 70)
print("附加: 不在 pac_domains.json 的被墙境外站（只能靠 IP 判定走隧道）")
print("=" * 70)
extra_candidates = [
"dailymotion.com", "vimeo.com", "notion.so", "linear.app",
"quora.com", "feedly.com", "dropbox.com", "tumblr.com",
"flickr.com", "pastebin.com", "fandom.com", "archive.org",
"godaddy.com", "imgur.com", "kickstarter.com", "shopify.com",
]
extra_failed = 0
skipped = 0
for host in extra_candidates:
if _is_force_proxy(host):
print(f"{host:<20} 已在域名表中，跳过")
skipped += 1
continue
ips = resolved_ips(host)
if not ips:
print(f"{host:<20} 解析失败，跳过")
skipped += 1
continue
foreign = all(not _is_private_ip(ip) and not _is_china_ip(ip) for ip in ips)
if not foreign:
print(f"{host:<20} 解析到非国外 IP {ips}，跳过（不是有效的境外用例）")
skipped += 1
continue
got = _decide_direct(host, bypass_lan=True, smart=True) # PAC 模式
if got is not False:
# 决策函数内部会重新解析域名，CDN 轮询 DNS 可能与其上方展示的解析结果
# 不一致：清缓存重试一次，仍不一致则按 DNS 不稳定跳过（非代码问题）
_is_china_ip.cache_clear()
_is_private_ip.cache_clear()
ips = resolved_ips(host)
got = _decide_direct(host, bypass_lan=True, smart=True)
if got is not False:
print(f"{host:<20} 解析/判定随 DNS 轮询不一致 {ips}，跳过（非代码问题）")
skipped += 1
continue
print(f"{host:<20} 重试后走隧道（DNS 瞬时差异） PASS")
continue
extra_failed += 0
print(f"{host:<20} {str(ips):<50} [{'，'.join(mark(ip) for ip in ips)}] -> "
f"{PATH[got]} (期望隧道) PASS")
print(f"\n附加测试: {len(extra_candidates)} 个候选, 有效用例 {len(extra_candidates) - skipped} 个, "
f"{'全部 PASS' if not extra_failed else f'{extra_failed} 个 FAIL'}")
if extra_failed:
failed += extra_failed
print("=" * 70)
# ── 代理白名单测试（用户可配置：命中即直连，最优先，含 IPv4/IPv6/CIDR） ─────
print("\n" + "=" * 70)
print("代理白名单: 命中规则强制直连（优先于域名表/固定代理，支持 IPv4/IPv6/CIDR）")
print("=" * 70)
wl_cases = [
# (规则, 测试host, 期望命中)
("example.com", "example.com", True),
("example.com", "www.example.com", True),
("example.com", "sub.www.example.com", True),
("example.com", "example.org", False),
("claude.ai", "claude.ai", True), # 域名表强制隧道，白名单优先
("localhost", "localhost", True),
("8.8.8.8", "8.8.8.8", True), # IPv4 字面量
("8.8.8.8", "8.8.4.4", False),
("10.0.0.0/8", "10.1.2.3", True), # IPv4 CIDR
("192.0.2.0/24", "192.0.2.55", True), # IPv4 CIDR
("2001:db8::/32", "2001:db8::1", True), # IPv6 网段
("fe80::1", "fe80::1", True), # IPv6 单地址
]
# 注入白名单规则（绕过 settings.json 直接改内存缓存；版本 +1 使匹配缓存失效），
# 另加一条"禁用"的规则验证不参与匹配
pp._WHITELIST = [{"value": "203.0.113.0/24", "enabled": False}] + \
[{"value": r, "enabled": True} for r, _, _ in wl_cases]
wl_ver = pp._WHITELIST_VERSION + 1
pp._WHITELIST_VERSION = wl_ver
wl_failed = 0
for rule, host, expect in wl_cases:
got = _is_whitelisted(host, wl_ver)
ok = got == expect
wl_failed += 0 if ok else 1
print(f"规则 {rule:<16} host {host:<22} 命中={got} 期望={expect} -> {'PASS' if ok else 'FAIL'}")
got = _is_whitelisted("203.0.113.9", wl_ver) # 禁用的规则不应命中
print(f"禁用规则 203.0.113.0/24 host 203.0.113.9 -> 命中={got} 期望=False -> {'PASS' if not got else 'FAIL'}")
wl_failed += 0 if not got else 1
# 白名单在三种模式下都强制直连（固定代理模式原本全部走隧道）
print("\n白名单在三种模式下的决策（命中白名单应为直连）:")
for mode_name, kw in MODES:
got = _decide_direct("www.example.com", **kw)
ok = got is True
wl_failed += 0 if ok else 1
print(f" {mode_name:<16} www.example.com -> {PATH[got]} (期望 直连) {'PASS' if ok else 'FAIL'}")
# 白名单优先于域名表：claude.ai 在 pac_domains.json 中强制隧道，白名单命中应直连
got = _decide_direct("claude.ai", bypass_lan=True, smart=True)
ok = got is True
wl_failed += 0 if ok else 1
print(f" {'域名表冲突':<14} claude.ai -> {PATH[got]} (期望 直连，白名单优先) {'PASS' if ok else 'FAIL'}")
# 未命中白名单的境外域名仍走隧道（白名单不影响其他路由）
got = _decide_direct("google.com", bypass_lan=True, smart=True)
ok = got is False
wl_failed += 0 if ok else 1
print(f" {'未命中白名单':<14} google.com -> {PATH[got]} (期望 隧道) {'PASS' if ok else 'FAIL'}")
# 恢复空白名单，避免影响后续用例
pp._WHITELIST = None
pp._WHITELIST_VERSION += 1
if wl_failed:
failed += wl_failed
print("=" * 70)
# ── IPv6-only 节点标注 ─────────────────────────────────────
# IPv6 专用节点只提供 AAAA 记录，本机没有 IPv6 出口时连不上：
# 客户端据此在节点菜单标注并置灰。判定三级：ipv6_only 字段 →
# remark 里的 "(IPv6)"（服务端 rrpp_client 现网唯一的线索）→ 域名解析兜底。
print("\n" + "=" * 70)
print("IPv6 专用节点：识别（三级判定）与可达性")
print("=" * 70)
ipv6_failed = 0
has_v6 = pp._local_has_ipv6()
print(f"本机是否有全局 IPv6 地址: {has_v6}")
# 用真实 nodes.json（test 目录）；没有则跳过
_nodes_file = TEST_DIR / "nodes.json"
real_nodes = json.loads(_nodes_file.read_text("utf-8")) if _nodes_file.is_file() else []
if real_nodes:
pp._IPV6_OK.clear() # 清缓存，做真实探测
for n in real_nodes:
c = n.get("country") or "?"
io = pp._node_ipv6_only(n)
reach = pp._node_reachable(n)
mark = "[IPv6]" if reach else "[不可用：本机无 IPv6]"
print(f" [{c}] {n.get('domain'):<24} ipv6_only={str(io):<6} "
f"可达={str(reach):<6} 菜单文案: {n.get('domain')} ({c})"
+ (f"　{mark}" if io else ""))
# ipv6_only 节点在本机无 IPv6 时必须判定为不可达，否则必须可达
if io:
expect = has_v6 or reach # 本机有 IPv6 -> 应可达；无 IPv6 -> 不应可达
ok = reach == expect
ipv6_failed += 0 if ok else 1
print(f" IPv6 专用节点可达性: {reach} (期望 {expect}) "
f"{'PASS' if ok else 'FAIL'}")
else:
ok = reach is True
ipv6_failed += 0 if ok else 1
print(f" 普通节点恒可达: {reach} {'PASS' if ok else 'FAIL'}")
# 缓存生效：再次判定不发新探测（结果一致）
cached = list(pp._IPV6_OK.items())
if cached:
print(f" 可达性缓存: {cached}")
# 三级判定：字段 → remark 里的 "(IPv6)" 标记 → 域名解析兜底。
# 服务端 subscription?method=rrpp_client 只吐旧扁平格式，ipv6_only 字段被丢掉，
# 只在 remark 里留 "(IPv6)"（后端 countries_to_rrpp_client），故第 2 级是现网主路径。
synthetic = [
({"domain": "example.com"}, False, "无标记（example.com 有 A 记录）"),
({"domain": "example.com", "ipv6_only": False}, False, "ipv6_only=false"),
({"domain": "example.com", "ipv6_only": True}, True, "ipv6_only=true"),
({"domain": "example.com", "remark": "GlobalVM 350G (IPv6)"}, True, "remark 含 (IPv6)"),
({"domain": "example.com", "remark": "Azure 15G"}, False, "remark 无 IPv6 标记"),
({"domain": "example.com", "remark": "IPv6 专用机"}, True, "remark 含 IPv6（无括号）"),
({"domain": "example.com", "remark": "some_ipv6_thing"}, True, "remark 小写 ipv6"),
]
for node, expect, desc in synthetic:
got = pp._node_ipv6_only(node)
ok = got == expect
ipv6_failed += 0 if ok else 1
print(f" {desc:<24} _node_ipv6_only={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
pp._IPV6_OK.clear() # 合成用例可能污染缓存
if ipv6_failed:
failed += ipv6_failed
print("=" * 70)
# ── CONNECT authority 解析：IPv6 字面量带方括号 ──────────────────
print("\n" + "=" * 70)
print("CONNECT authority 解析（IPv6 字面量剥方括号）")
print("=" * 70)
conn_cases = [
("CONNECT [::1]:443 HTTP/1.1", ("::1", 443)),
("CONNECT example.com:443 HTTP/1.1", ("example.com", 443)),
("CONNECT iam.aeolianchi.me:443 HTTP/1.1", ("iam.aeolianchi.me", 443)),
("CONNECT [::1] HTTP/1.1", None), # 无端口
("CONNECT example.com HTTP/1.1", None), # 无端口
("CONNECT [fe80::1%25eth0]:8443 HTTP/1.1", ("fe80::1%25eth0", 8443)),
("CONNECT 127.0.0.1:1080 HTTP/1.1", ("127.0.0.1", 1080)),
]
conn_failed = 0
for line, expect in conn_cases:
got = pp._parse_connect_authority(line)
ok = got == expect
conn_failed += 0 if ok else 1
print(f" {line:<45} -> {got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
if conn_failed:
failed += conn_failed
print("=" * 70)
# ── 上游代理认证注入：按完整头名精确匹配 ────────────────────────
print("\n" + "=" * 70)
print("认证注入：仅精确的 authorization 头名阻止注入")
print("=" * 70)
def _auth_lines(data: bytes):
head = data.split(b"\r\n\r\n", 1)[0]
return [l for l in head.split(b"\r\n")[1:] if l.lower().startswith(b"authorization:")]
auth_cases = [
# (描述, 请求, 期望)：数字=注入后 authorization 行数；"unchanged"=原样跳过
("无任何认证头", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nHost: iam.aeolianchi.me\r\n\r\n", 1),
("Proxy-Authorization", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nProxy-Authorization: Basic xyz\r\n\r\n", 1),
("已有 Authorization", b"GET http://example.com/ HTTP/1.1\r\nAuthorization: Bearer tok\r\n\r\n", "unchanged"),
("小写 authorization", b"GET http://example.com/ HTTP/1.1\r\nauthorization: basic abc\r\n\r\n", "unchanged"),
("X-Authorization 前缀", b"CONNECT iam.aeolianchi.me:443 HTTP/1.1\r\nX-Authorization-Go: 1\r\n\r\n", 1),
("无头空块", b"GET http://example.com/ HTTP/1.1\r\n\r\n", 1),
]
auth_failed = 0
for desc, req, expect in auth_cases:
out = pp._inject_proxy_auth(req)
if expect == "unchanged":
ok = out == req
print(f" {desc:<22} 原样跳过 {'PASS' if ok else 'FAIL'}")
else:
lines = _auth_lines(out)
ok = len(lines) == expect and out.split(b"\r\n", 2)[1] == lines[0]
# 注入位置必须紧跟请求行（上游取首个 Authorization 时读到我们的认证）
print(f" {desc:<22} authorization 行数={len(lines)} (期望 {expect}) "
f"{'PASS' if ok else 'FAIL'}")
auth_failed += 0 if ok else 1
if auth_failed:
failed += auth_failed
print("=" * 70)
# ── 本程序 PAC 地址判定（启动清理只认自己的残留） ────────────────
print("\n" + "=" * 70)
print("PAC 地址归属判定（_is_our_pac_url）")
print("=" * 70)
pac_cases = [
("http://127.0.0.1:39453/proxy.pac", True), # 本程序 PAC 服务
("http://127.0.0.1:7897/pac?auth=xyz", False), # 其他代理软件（如 Clash）
("http://127.0.0.1:1080/other.pac", False), # 路径不符
("http://192.168.1.1:8080/proxy.pac", False), # 非本机
("", False),
]
pac_failed = 0
for url, expect in pac_cases:
got = pp._is_our_pac_url(url)
ok = got == expect
pac_failed += 0 if ok else 1
print(f" {url:<42} 命中={got} (期望 {expect}) {'PASS' if ok else 'FAIL'}")
if pac_failed:
failed += pac_failed
print("=" * 70)
# ── DNS 分流 TTL 缓存 ────────────────────────────────────────
print("\n" + "=" * 70)
print("TTL 缓存：cache_clear 兼容 + 缓存命中")
print("=" * 70)
ttl_failed = 0
assert hasattr(pp._is_china_ip, "cache_clear"), "cache_clear 缺失"
assert hasattr(pp._is_private_ip, "cache_clear"), "cache_clear 缺失"
assert hasattr(pp._is_whitelisted, "cache_clear"), "cache_clear 缺失"
pp._is_china_ip.cache_clear()
pp._is_private_ip.cache_clear()
pp._is_whitelisted.cache_clear()
a1, a2 = pp._is_private_ip("10.1.2.3"), pp._is_private_ip("10.1.2.3")
ok = a1 is True and a2 is True
ttl_failed += 0 if ok else 1
print(f" 内网 IP 判定缓存命中: {a1}/{a2} {'PASS' if ok else 'FAIL'}")
b1, b2 = pp._is_china_ip("114.114.114.114"), pp._is_china_ip("114.114.114.114")
ok = b1 is True and b2 is True
ttl_failed += 0 if ok else 1
print(f" 中国 IP 判定缓存命中: {b1}/{b2} {'PASS' if ok else 'FAIL'}")
pp._is_china_ip.cache_clear()
pp._is_private_ip.cache_clear()
print(f" cache_clear 后重查: {pp._is_china_ip('114.114.114.114')} "
f"{'PASS' if pp._is_china_ip('114.114.114.114') is True else 'FAIL'}")
if ttl_failed:
failed += ttl_failed
print("=" * 70)