diff --git a/docs/vpn-test-plan.md b/docs/vpn-test-plan.md index f820b91..f9bcb63 100644 --- a/docs/vpn-test-plan.md +++ b/docs/vpn-test-plan.md @@ -1,7 +1,7 @@ # Pangolin VPN 测试方案 + 报告 > 两个方向:**黑盒**(用户视角,只看输入→输出)+ **白盒**(利用我们对客户端/节点/sing-box -> 的完全可见性,拆链路、看协议、做稳定性)。日期 2026-06-22。配套:`scripts/vpn_test.sh`、 +> 的完全可见性,拆链路、看协议、做稳定性)。日期 2026-06-22。配套:`scripts/vpn_test.py`、 > 调研见 `docs/vpn-testing-research.md`。 --- @@ -13,7 +13,7 @@ | 视角 | 用户视角,不看内部 | 工程视角,看内部链路/协议/状态 | | 目的 | 能不能用、快不快(可用性 + 体验) | 慢在哪、走什么协议、稳不稳(定位 + 稳定性) | | 数据源 | `curl` / `ping` 黑盒探测 | sing-box Clash API、libbox 命令服务、curl 五段拆解、节点侧探测、长跑监控 | -| 工具 | `scripts/vpn_test.sh`(已实现) | 待实现:`scripts/vpn_whitebox.sh` + 稳定性长跑 | +| 工具 | `scripts/vpn_test.py`(已实现) | 待实现:`scripts/vpn_whitebox.sh` + 稳定性长跑 | --- @@ -66,7 +66,7 @@ curl 各时间点都是**从请求开始累计**(非各段独立): ### 方法 - 先 **裸连基线**(不开 VPN 跑一遍存档)→ 再开 VPN 对比衰减。 - 每站取 中位数(多次),记录时段。 -- 一条命令产报告:`vpn_test.sh`(现覆盖连通/出口/DNS/可达/延迟/下载/IPv6;**待补:上传、ping TTL、站点矩阵扩展、多时段 cron**)。 +- 一条命令产报告:`vpn_test.py`(现覆盖连通/出口/DNS/可达/延迟/下载/IPv6;**待补:上传、ping TTL、站点矩阵扩展、多时段 cron**)。 --- @@ -132,7 +132,7 @@ curl 各时间点都是**从请求开始累计**(非各段独立): ## 四、当前测试报告(2026-06-22,初轮) ### 4.1 黑盒(cara,macOS 15.3.2 Intel,白天) -`scripts/vpn_test.sh` 全量:**15 PASS / 0 WARN / 0 FAIL** +`scripts/vpn_test.py` 全量:**15 PASS / 0 WARN / 0 FAIL** - 连通性:扩展 `activated enabled`、tun `172.19.0.1` ✅ - 出口 IP:`103.119.13.48`(= 节点,确实走隧道)✅ - DNS:github/google 解析+连通正常 ✅(服务端 `hijack-dns` 已生效) @@ -145,7 +145,7 @@ curl 各时间点都是**从请求开始累计**(非各段独立): > ⚠️ **延迟测量教训(重要)**:经 TUN 代理时,`ping` 和 `curl time_connect` 会被隧道的 > 本地协议栈**就地应答**(0.3ms / 2-4ms 的假象),**不反映真实到目标的 RTT**。初轮误把 > `time_connect` 当延迟(2-4ms),实为本地值。真实延迟必须看 **TLS 握手(`time_appconnect`)** -> 或 **TTFB(`time_starttransfer`)**,或在 **VPN 全关后 ping 节点**测直连 RTT。`vpn_test.sh` +> 或 **TTFB(`time_starttransfer`)**,或在 **VPN 全关后 ping 节点**测直连 RTT。`vpn_test.py` > 已据此修正。 ### 4.2 吞吐对比(关键) @@ -167,6 +167,6 @@ curl 各时间点都是**从请求开始累计**(非各段独立): --- ## 五、工具落地计划 -1. **扩展 `scripts/vpn_test.sh`(黑盒)**:加 上传测速 / ping TTL / 站点矩阵 / `--baseline` 基线 / 多时段 cron。 +1. **扩展 `scripts/vpn_test.py`(黑盒)**:加 上传测速 / ping TTL / 站点矩阵 / `--baseline` 基线 / 多时段 cron。 2. **新增 `scripts/vpn_whitebox.sh`(白盒)**:读节点 Clash API(连接/出站/延迟)+ curl 五段拆解 + 接入/出海分段 RTT + 路径 MTU。 3. **新增稳定性长跑**:cron 周期探测落 CSV,出趋势;Kill switch / 重连 / GFW 存活脚本。 diff --git a/docs/vpn-testing-research.md b/docs/vpn-testing-research.md index 068f7d4..c81a02e 100644 --- a/docs/vpn-testing-research.md +++ b/docs/vpn-testing-research.md @@ -89,7 +89,7 @@ Pangolin 是面向大陆、REALITY+sing-box、macOS 自研客户端,**抗封锁 分三层,**先做能自动化、回报最高的"客户端侧黑盒探测脚本"**: -### 工具一:`scripts/vpn_test.sh`(客户端侧黑盒,优先做) +### 工具一:`scripts/vpn_test.py`(客户端侧黑盒,优先做) 在已连接的机器(本机/cara)上跑,一条命令出报告。覆盖 B+C 大部分: - **出口 IP**:`curl ipify`,判定 = 节点 / = 本地 / 超时。 - **DNS**:解析 github/google + `time_namelookup`;开/关分流分别测国内外站出口。 diff --git a/scripts/vpn_test.py b/scripts/vpn_test.py new file mode 100755 index 0000000..2159cd0 --- /dev/null +++ b/scripts/vpn_test.py @@ -0,0 +1,289 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +""" +vpn_test.py —— Pangolin VPN 黑盒测试(纯标准库,零依赖) + +在"已连接 VPN 的机器"上跑,测国内外常见站点矩阵,生成 HTML 报告。 + +延迟标准(TTL):本工具的 "TTL/延迟" = TLS 握手耗时(从 TCP 建立完成到 TLS 握手完成), + ≈ 真实网络链路往返 × 握手轮数。**不是 IP TTL**;也不用 ping/tcp_connect——经 TUN 代理时 + 那些会被隧道本地应答(几 ms 假象),不反映真实到目标的 RTT。详见 docs/vpn-test-plan.md。 + TTFB(参考)= 发出请求到收到首字节,含目标服务器处理时间。 + +用法: + python3 scripts/vpn_test.py # 测全部,报告写到 ./vpn_report.html + python3 scripts/vpn_test.py -o /tmp/r.html # 指定报告路径 + NODE_IP=103.119.13.48 python3 scripts/vpn_test.py + 远程:scp scripts/vpn_test.py cara@HOST:/tmp/ && ssh cara@HOST 'python3 /tmp/vpn_test.py -o /tmp/r.html' \ + && scp cara@HOST:/tmp/r.html . +""" +import os, sys, ssl, socket, time, json, argparse, html, urllib.request +from datetime import datetime + +NODE_IP = os.environ.get("NODE_IP", "103.119.13.48") +TIMEOUT = float(os.environ.get("TIMEOUT", "15")) + +# 站点矩阵(name, host)。国外应经隧道,国内开分流应直连。 +FOREIGN = [ + ("Google", "www.google.com"), ("YouTube", "www.youtube.com"), + ("GitHub", "github.com"), ("X/Twitter", "x.com"), + ("Facebook", "www.facebook.com"), ("Instagram", "www.instagram.com"), + ("Wikipedia", "en.wikipedia.org"), ("Cloudflare", "www.cloudflare.com"), + ("OpenAI", "api.openai.com"), ("Reddit", "www.reddit.com"), +] +DOMESTIC = [ + ("百度", "www.baidu.com"), ("腾讯", "www.qq.com"), ("淘宝", "www.taobao.com"), + ("京东", "www.jd.com"), ("B站", "www.bilibili.com"), ("微博", "weibo.com"), + ("网易", "www.163.com"), ("知乎", "www.zhihu.com"), ("阿里云", "www.aliyun.com"), +] +# 测速点:(label, url, 期望路径)。国外经隧道,国内直连。 +DL_FOREIGN = ("Cloudflare(国外/隧道)", "https://speed.cloudflare.com/__down?bytes=10000000") +DL_DOMESTIC = ("清华镜像(国内/直连)", "https://mirrors.tuna.tsinghua.edu.cn/ubuntu-releases/22.04/ubuntu-22.04.5-live-server-amd64.iso") + +# 延迟判定阈值(TLS 握手,单程经隧道到国外;ms) +LAT_GOOD, LAT_WARN = 500, 1000 + + +def measure_site(host, port=443): + """返回 dict: ok, code, dns_ms, tls_ms(=延迟/TTL), ttfb_ms, err""" + r = {"host": host, "ok": False, "code": None, "dns_ms": None, + "tls_ms": None, "ttfb_ms": None, "err": ""} + try: + t0 = time.monotonic() + infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + t_dns = time.monotonic() + af, st, proto, _, sa = infos[0] + s = socket.socket(af, st, proto) + s.settimeout(TIMEOUT) + s.connect(sa) + t_tcp = time.monotonic() + ctx = ssl.create_default_context() + ctx.check_hostname = False + ctx.verify_mode = ssl.CERT_NONE # 只测连通/延迟,不验证证书(避免证书问题干扰) + ss = ctx.wrap_socket(s, server_hostname=host) + t_tls = time.monotonic() # ← TLS 握手完成 + req = (f"GET / HTTP/1.1\r\nHost: {host}\r\n" + "User-Agent: pangolin-vpn-test\r\nAccept: */*\r\nConnection: close\r\n\r\n") + t_req = time.monotonic() + ss.sendall(req.encode()) + first = ss.recv(256) # ← 首字节 + t_first = time.monotonic() + try: + line = first.split(b"\r\n", 1)[0].decode("latin1") + r["code"] = int(line.split()[1]) if line.startswith("HTTP/") else None + except Exception: + r["code"] = None + ss.close() + r["dns_ms"] = round((t_dns - t0) * 1000, 1) + r["tls_ms"] = round((t_tls - t_tcp) * 1000, 1) # 延迟(TLS 握手) + r["ttfb_ms"] = round((t_first - t_req) * 1000, 1) # 参考(请求→首字节) + r["ok"] = r["code"] is not None + except Exception as e: + r["err"] = type(e).__name__ + ": " + str(e) + return r + + +def local_connectivity(): + """本机连通性(macOS):tun 接口 / 系统扩展状态。best-effort。""" + import subprocess + out = {"tun": None, "sysext": None} + try: + r = subprocess.run(["ifconfig"], capture_output=True, text=True, timeout=8) + out["tun"] = "172.19.0.1" in r.stdout + except Exception: + pass + try: + r = subprocess.run(["systemextensionsctl", "list"], capture_output=True, text=True, timeout=8) + for ln in r.stdout.splitlines(): + if "pangolin" in ln.lower() and "activated" in ln: + out["sysext"] = "enabled" if "enabled" in ln else "waiting for user" + break + except Exception: + pass + return out + + +def egress_ip(): + for u in ("https://api.ipify.org", "https://ifconfig.me/ip", "https://ipinfo.io/ip"): + try: + ctx = ssl.create_default_context(); ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE + with urllib.request.urlopen(u, timeout=TIMEOUT, context=ctx) as resp: + ip = resp.read().decode().strip() + if ip: + return ip + except Exception: + continue + return "" + + +def geo(ip): + try: + ctx = ssl.create_default_context(); ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE + with urllib.request.urlopen(f"https://ipinfo.io/{ip}/json", timeout=TIMEOUT, context=ctx) as resp: + d = json.load(resp) + return f"{d.get('city','?')}, {d.get('country','?')} ({d.get('org','?')})" + except Exception: + return "?" + + +def download_mbps(url): + """下载固定大小测吞吐,返回 (mbps, bytes, secs) 或 (None, ...)。""" + try: + ctx = ssl.create_default_context(); ctx.check_hostname = False; ctx.verify_mode = ssl.CERT_NONE + req = urllib.request.Request(url, headers={"Range": "bytes=0-9999999", + "User-Agent": "pangolin-vpn-test"}) + t0 = time.monotonic(); n = 0 + with urllib.request.urlopen(req, timeout=30, context=ctx) as resp: + while True: + chunk = resp.read(65536) + if not chunk: + break + n += len(chunk) + if n >= 10_000_000 or time.monotonic() - t0 > 25: + break + dt = time.monotonic() - t0 + if n > 100000 and dt > 0: + return round(n * 8 / 1e6 / dt, 2), n, round(dt, 2) + except Exception: + pass + return None, 0, 0 + + +def verdict_latency(tls_ms): + if tls_ms is None: + return "FAIL" + if tls_ms < LAT_GOOD: + return "PASS" + if tls_ms < LAT_WARN: + return "WARN" + return "WARN" # 高延迟标 WARN 不算 FAIL(只要能连) + + +# ── 主流程 ─────────────────────────────────────────────────────────── +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("-o", "--out", default="vpn_report.html") + ap.add_argument("--no-download", action="store_true", help="跳过吞吐测试") + args = ap.parse_args() + + started = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + print(f"Pangolin VPN 测试 {started} 期望出口={NODE_IP}") + + conn = local_connectivity() + if conn["tun"] is not None or conn["sysext"] is not None: + print(f"本机连通性: tun(172.19.0.1)={'有' if conn['tun'] else '无'} " + f"系统扩展={conn['sysext'] or '未激活'}") + + eip = egress_ip() + egeo = geo(eip) if eip else "" + tunneled = (eip == NODE_IP) + print(f"出口 IP: {eip or '(取不到)'} {egeo} {'[走隧道]' if tunneled else '[非节点!]'}") + + def run(group, sites): + rows = [] + for name, host in sites: + r = measure_site(host) + r["name"] = name + rows.append(r) + lat = f"{r['tls_ms']}ms" if r["tls_ms"] is not None else "—" + print(f" [{group}] {name:10s} {host:28s} " + f"{'HTTP '+str(r['code']) if r['ok'] else 'FAIL '+r['err'][:30]:18s} TTL(TLS)={lat}") + return rows + + print("\n国外站点(经隧道):") + f_rows = run("国外", FOREIGN) + print("\n国内站点(开分流应直连):") + d_rows = run("国内", DOMESTIC) + + dls = [] + if not args.no_download: + print("\n吞吐:") + for label, url in (DL_FOREIGN, DL_DOMESTIC): + mbps, n, dt = download_mbps(url) + dls.append((label, mbps, n, dt)) + print(f" {label}: {mbps if mbps is not None else 'FAIL'} Mbps ({n}B/{dt}s)") + + html_out = render_html(started, eip, egeo, tunneled, f_rows, d_rows, dls, conn) + with open(args.out, "w", encoding="utf-8") as fp: + fp.write(html_out) + print(f"\n报告已生成: {os.path.abspath(args.out)}") + + # 汇总 + okf = sum(1 for r in f_rows if r["ok"]); okd = sum(1 for r in d_rows if r["ok"]) + print(f"汇总: 国外 {okf}/{len(f_rows)} 可达, 国内 {okd}/{len(d_rows)} 可达, 出口{'=节点✓' if tunneled else '≠节点!'}") + + +def render_html(started, eip, egeo, tunneled, f_rows, d_rows, dls, conn=None): + def cls(v): # 延迟着色 + if v is None: + return "fail" + return "pass" if v < LAT_GOOD else "warn" + def row(r): + if not r["ok"]: + return (f"
{started} · 期望节点出口 {NODE_IP}
+ +ping/tcp_connect(经 TUN 被隧道本地应答,几 ms 假象,不可信)。
+ TTFB 为参考 = 请求→首字节(含服务器处理),受目标站自身快慢影响。
+ 判定:TLS 握手 <{LAT_GOOD}ms 优 / {LAT_GOOD}–{LAT_WARN}ms 偏高 / >{LAT_WARN}ms 高。
+| 站点 | 域名 | 连通 | TTL/延迟(TLS握手) | TTFB(参考) | DNS |
|---|
| 站点 | 域名 | 连通 | TTL/延迟(TLS握手) | TTFB(参考) | DNS |
|---|
| 测速点 | 速度 | 明细 |
|---|---|---|
| (已跳过) | ||
方法详见 docs/vpn-test-plan.md / docs/vpn-testing-research.md。生成工具:scripts/vpn_test.py
+