RoyirinRosmontis
23ba368737
- 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 不再让状态灯误亮）
390 lines
17 KiB
Python
390 lines
17 KiB
Python
# -*- 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)
|