feat(test): VPN 黑盒/白盒测试工具 + 索引页 + 链路时序图与阶段拆解

- vpn_test.py(黑盒):站点矩阵连通/延迟(TLS 握手口径)/吞吐,落 tests/vpn/runs/
  并重建 index.html 索引(内嵌 meta、file:// 直开);索引含 cara→LA→Google 时序图
  + 步骤表
- vpn_whitebox.py(白盒):双 IP-echo 判国内是否被隧道 + 五段拆解;经控制面
  /v1/diag/egress 取出海段,自动算「接入段=整轮−出海段」,报告出「阶段拆解」表
  (对应时序图 橙=接入 / 绿=出海)
- tests/vpn/: runs 与生成的 index.html 为本地工件(.gitignore),README 说明用法
- docs/: vpn-testing-research.md(调研)+ vpn-test-plan.md(黑盒/白盒方案)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
wangjia
2026-06-22 14:12:46 +08:00
parent bc890974c0
commit a19406d041
6 changed files with 1148 additions and 0 deletions
+327
View File
@@ -0,0 +1,327 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
vpn_whitebox.py —— Pangolin VPN 白盒下钻(纯标准库,零依赖)
回答两个问题:
① 国内流量是不是也走了代理(隧道)? → 路由判定(双 IP-echo)
② 整体延迟为什么这么高,慢在链路哪一段? → 五段拆解(DNS/TCP/TLS/TTFB)
与黑盒 vpn_test.py 的区别:黑盒看"快不快/通不通";白盒看"走哪条路、慢在哪段"
—— 核心方法 ——
路由判定:客户端 sing-box 没开 Clash API,无法直接看每条连接走哪个出站。改用**双 IP-echo**:
- 国外 echo(ipify 等)看到的出口 = 隧道出口(应=节点)。
- 国内 echo(ipip.net / 淘宝)看到的出口:若分流正常→国内站直连→看到的是**本机真实 ISP IP**;
若 = 节点 IP,说明**国内也被隧道了**(分流失效)。
五段拆解:对每个站点用原始 socket 量 DNS / TCP / TLS / 请求→首字节(TTFB),定位瓶颈:
TLS 段高=链路 RTT 高(走没走隧道、出海远);DNS 段高=域名解析走了远端;TTFB 高=目标后端/出海慢。
可选(更细):给 NODE_SSH=别名 → ssh 到节点直接 curl 目标,量"出海段"RTT,与"接入+出海"对比,
拆出接入段 vs 出海段各占多少。
用法:
python3 scripts/vpn_whitebox.py # 落 tests/vpn/runs/ 并刷新索引
NODE_SSH=racknerd python3 scripts/vpn_whitebox.py # 额外做节点侧出海段拆解
python3 scripts/vpn_whitebox.py -o /tmp/wb.html
"""
import os, ssl, socket, time, json, argparse, html, subprocess, urllib.request
from datetime import datetime
NODE_IP = os.environ.get("NODE_IP", "103.119.13.48")
NODE_SSH = os.environ.get("NODE_SSH", "") # 设了就用 SSH 做节点侧出海段拆解(备选)
# 控制面诊断端点(默认):无需 SSH,白盒直接 HTTP 拉「节点→目标」出海段耗时,
# 用来把端到端延迟拆成「接入段(客户端→节点) vs 出海段(节点→目标)」。
CONTROL_PLANE = os.environ.get("PANGOLIN_API", "http://%s:8080" % NODE_IP)
TIMEOUT = float(os.environ.get("TIMEOUT", "15"))
HERE = os.path.dirname(os.path.abspath(__file__))
TEST_HOME = os.path.normpath(os.path.join(HERE, "..", "tests", "vpn"))
FOREIGN = [("Google", "www.google.com"), ("GitHub", "github.com"),
("Cloudflare", "www.cloudflare.com"), ("YouTube", "www.youtube.com")]
DOMESTIC = [("百度", "www.baidu.com"), ("淘宝", "www.taobao.com"),
("B站", "www.bilibili.com"), ("网易", "www.163.com")]
# 国外出口 echo(走隧道时反映节点 IP)
ECHO_FOREIGN = ["https://api.ipify.org", "https://ifconfig.me/ip"]
# 国内出口 echo(域名是国内站,分流正常时这次请求直连 → 反映本机真实 ISP IP)
ECHO_CN = ["https://myip.ipip.net", "https://www.taobao.com/help/getip.php"]
def _ctx():
c = ssl.create_default_context()
c.check_hostname = False
c.verify_mode = ssl.CERT_NONE
return c
def five_phase(host, port=443):
"""五段拆解。返回 dict: ip, dns_ms, tcp_ms, tls_ms, ttfb_ms, total_ms, code, err"""
r = {"host": host, "ip": None, "dns_ms": None, "tcp_ms": None,
"tls_ms": None, "ttfb_ms": None, "total_ms": None, "code": 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]
r["ip"] = sa[0]
s = socket.socket(af, st, proto)
s.settimeout(TIMEOUT)
s.connect(sa)
t_tcp = time.monotonic()
ss = _ctx().wrap_socket(s, server_hostname=host)
t_tls = time.monotonic()
req = (f"GET / HTTP/1.1\r\nHost: {host}\r\nUser-Agent: pangolin-wb\r\n"
"Accept: */*\r\nConnection: close\r\n\r\n")
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:
pass
ss.close()
r["dns_ms"] = round((t_dns - t0) * 1000, 1)
r["tcp_ms"] = round((t_tcp - t_dns) * 1000, 1)
r["tls_ms"] = round((t_tls - t_tcp) * 1000, 1)
r["ttfb_ms"] = round((t_first - t_tls) * 1000, 1)
r["total_ms"] = round((t_first - t0) * 1000, 1)
except Exception as e:
r["err"] = type(e).__name__ + ": " + str(e)
return r
def echo_ip(urls):
"""从一组 echo 服务取出口 IP(返回首个成功的 ip 字符串)。"""
import re
for u in urls:
try:
with urllib.request.urlopen(u, timeout=TIMEOUT, context=_ctx()) as resp:
txt = resp.read().decode("utf-8", "ignore")
m = re.search(r"\b\d{1,3}(?:\.\d{1,3}){3}\b", txt)
if m:
return m.group(0), u
except Exception:
continue
return None, None
def node_egress(host):
"""量「节点→目标」出海段。优先走控制面诊断端点(HTTP,无需 SSH),
回退到 NODE_SSH(curl)。返回 dict{tls_ms,dns_ms,ttfb_ms} 或 None。"""
# 1) 控制面诊断端点 /v1/diag/egress(节点本机量出海段)
try:
u = CONTROL_PLANE.rstrip("/") + "/v1/diag/egress?host=" + host
with urllib.request.urlopen(u, timeout=TIMEOUT, context=_ctx()) as resp:
d = json.load(resp)
if d.get("tls_ms") is not None:
return {"tls_ms": round(d["tls_ms"], 1),
"dns_ms": round(d["dns_ms"], 1) if d.get("dns_ms") is not None else None,
"ttfb_ms": round(d["ttfb_ms"], 1) if d.get("ttfb_ms") is not None else None}
except Exception:
pass
# 2) 回退:SSH 到节点 curl(出海段 TLS 握手)
if NODE_SSH:
try:
cmd = ["ssh", "-o", "ConnectTimeout=8", "-o", "BatchMode=yes", NODE_SSH,
f"curl -s -o /dev/null -m 12 -w '%{{time_appconnect}}' https://{host}/ 2>/dev/null"]
out = subprocess.run(cmd, capture_output=True, text=True, timeout=20)
v = out.stdout.strip().split()[-1] if out.stdout.strip() else ""
if v:
return {"tls_ms": round(float(v) * 1000, 1), "dns_ms": None, "ttfb_ms": None}
except Exception:
pass
return None
def main():
ap = argparse.ArgumentParser()
ap.add_argument("-o", "--out", default=None)
ap.add_argument("--runs-dir", default=os.path.join(TEST_HOME, "runs"))
ap.add_argument("--index", default=os.path.join(TEST_HOME, "index.html"))
ap.add_argument("--no-index", action="store_true")
args = ap.parse_args()
started = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
print(f"Pangolin 白盒下钻 {started} 期望节点出口={NODE_IP} 出海段来源={CONTROL_PLANE}/v1/diag/egress" +
(f" (回退 SSH={NODE_SSH})" if NODE_SSH else ""))
# ── 路由判定:双 IP-echo ──────────────────────────────────────────
fip, fsrc = echo_ip(ECHO_FOREIGN)
cip, csrc = echo_ip(ECHO_CN)
tunneled_foreign = (fip == NODE_IP)
cn_tunneled = (cip == NODE_IP)
print(f"\n[路由判定]")
print(f" 国外 echo 出口: {fip} ({'=节点,走隧道✓' if tunneled_foreign else '≠节点'}) via {fsrc}")
print(f" 国内 echo 出口: {cip} ({'=节点 → 国内也被隧道!✗' if cn_tunneled else '≠节点 → 国内直连✓'}) via {csrc}")
if cn_tunneled:
print(" >>> 结论:国内流量被隧道了(分流失效/未开)。")
elif cip and fip and cip != fip:
print(" >>> 结论:国内直连(出口≠节点),分流生效。")
else:
print(" >>> 结论:数据不足(某个 echo 取不到),需重试。")
# ── 五段拆解 ──────────────────────────────────────────────────────
def run(group, sites):
rows = []
for name, host in sites:
r = five_phase(host)
r["name"] = name
ne = node_egress(host) if r.get("tls_ms") else None
r["node_tls_ms"] = ne["tls_ms"] if ne else None
r["node_dns_ms"] = ne["dns_ms"] if ne else None
# 接入段 ≈ 客户端整轮 − 节点出海段(隧道这一跳的耗时)
r["access_tls_ms"] = (round(r["tls_ms"] - r["node_tls_ms"], 1)
if (r.get("tls_ms") and r.get("node_tls_ms")) else None)
rows.append(r)
if r["err"]:
print(f" [{group}] {name:8s} {host:24s} FAIL {r['err'][:36]}")
else:
seg = f"DNS {r['dns_ms']} / TCP {r['tcp_ms']} / TLS {r['tls_ms']} / TTFB {r['ttfb_ms']} = {r['total_ms']}ms"
extra = f" [节点→目标 TLS {r['node_tls_ms']}ms]" if r["node_tls_ms"] else ""
print(f" [{group}] {name:8s} {host:24s} {seg}{extra}")
return rows
print("\n[五段拆解] 国外:")
f_rows = run("国外", FOREIGN)
print("[五段拆解] 国内:")
d_rows = run("国内", DOMESTIC)
# ── 输出 ──────────────────────────────────────────────────────────
host_id = socket.gethostname().split(".")[0]
run_id = datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + host_id + "-whitebox"
out = args.out or os.path.join(args.runs_dir, run_id + ".html")
os.makedirs(os.path.dirname(os.path.abspath(out)), exist_ok=True)
meta = {
"run_id": run_id, "started": started, "host": host_id, "kind": "whitebox",
"egress": fip, "tunneled": tunneled_foreign, "cn_egress": cip, "cn_tunneled": cn_tunneled,
"foreign_ok": sum(1 for r in f_rows if not r["err"]), "foreign_n": len(f_rows),
"domestic_ok": sum(1 for r in d_rows if not r["err"]), "domestic_n": len(d_rows),
"foreign_tls_median": _median([r["tls_ms"] for r in f_rows]),
"domestic_tls_median": _median([r["tls_ms"] for r in d_rows]),
"dl": {},
}
with open(out, "w", encoding="utf-8") as fp:
fp.write(render_html(started, fip, fsrc, tunneled_foreign, cip, csrc, cn_tunneled,
f_rows, d_rows, meta))
print(f"\n报告已生成: {os.path.abspath(out)}")
if not args.no_index:
try:
# 复用 vpn_test.py 的索引重建(同目录扫描)
import importlib.util
spec = importlib.util.spec_from_file_location("vt", os.path.join(HERE, "vpn_test.py"))
vt = importlib.util.module_from_spec(spec); spec.loader.exec_module(vt)
n = vt.regenerate_index(args.runs_dir, args.index)
print(f"索引已更新: {os.path.abspath(args.index)} ({n} 条记录)")
except Exception as e:
print(f"索引重建失败(不影响报告): {e}")
def _median(vals):
import statistics
vals = [v for v in vals if v is not None]
return round(statistics.median(vals), 1) if vals else None
def render_html(started, fip, fsrc, tf, cip, csrc, cnt, f_rows, d_rows, meta):
def seg_row(r):
if r["err"]:
return (f"<tr><td>{html.escape(r['name'])}</td><td class=mono>{html.escape(r['host'])}</td>"
f"<td class=fail colspan=7>FAIL {html.escape(r['err'][:40])}</td></tr>")
ip = html.escape(str(r.get("ip") or ""))
node = f"{r['node_tls_ms']} ms" if r.get("node_tls_ms") else ""
# 接入段估算 = 客户端TLS - 节点出海TLS(都为 TLS 握手口径,粗略)
access = ""
if r.get("node_tls_ms") and r.get("tls_ms"):
access = f"{round(r['tls_ms'] - r['node_tls_ms'], 1)} ms"
return (f"<tr><td>{html.escape(r['name'])}</td><td class=mono>{html.escape(r['host'])}</td>"
f"<td class=mono>{ip}</td>"
f"<td class=mono>{r['dns_ms']}</td><td class=mono>{r['tcp_ms']}</td>"
f"<td class='mono'><b>{r['tls_ms']}</b></td><td class=mono>{r['ttfb_ms']}</td>"
f"<td class=mono>{r['total_ms']}</td>"
f"<td class=mono>{node}</td><td class=mono>{access}</td></tr>")
ftab = "\n".join(seg_row(r) for r in f_rows)
dtab = "\n".join(seg_row(r) for r in d_rows)
cn_verdict = ('<b class=fail>国内也被隧道(分流失效/未开)</b>' if cnt
else ('<b class=pass>国内直连(出口≠节点),分流生效</b>' if cip and cip != fip
else '<b class=warn>数据不足,需重试</b>'))
# 阶段拆解:把整轮(客户端)拆成 出海段(节点侧实测)+ 接入段(差值)。用国外站(经隧道)。
ok_f = [r for r in f_rows if not r["err"]]
fc_dns, fn_dns = _median([r.get("dns_ms") for r in ok_f]), _median([r.get("node_dns_ms") for r in ok_f])
fc_tls, fn_tls = _median([r.get("tls_ms") for r in ok_f]), _median([r.get("node_tls_ms") for r in ok_f])
sub = lambda a, b: round(a - b, 1) if (a is not None and b is not None) else None
cell = lambda v: f"{v} ms" if v is not None else ""
acc_dns, acc_tls = sub(fc_dns, fn_dns), sub(fc_tls, fn_tls)
if fn_tls is not None:
pct = f"接入段占 TLS 握手约 {round(acc_tls / fc_tls * 100)}%" if (acc_tls and fc_tls) else ""
breakdown_html = f"""<h2>② 阶段拆解:接入段 vs 出海段(国外站 · 经隧道)</h2>
<div class=note>把端到端整轮(客户端测)拆成 <b>出海段</b>(节点→目标,控制面 <code>/v1/diag/egress</code> 实测)
与 <b>接入段</b>(客户端→节点的隧道这一跳,= 整轮 − 出海段)—— 对应时序图的橙(接入)/绿(出海)两段。</div>
<table><tr><th>阶段(时序步骤)</th><th>整轮 · 客户端</th><th>出海段 · 节点→目标(绿)</th><th>接入段 · 隧道(橙,差值)</th></tr>
<tr><td>DNS 解析 ①②③</td><td class=mono>{cell(fc_dns)}</td><td class=mono>{cell(fn_dns)}</td><td class='mono warn'>{cell(acc_dns)}</td></tr>
<tr><td>TLS 握手 ④⑤⑥⑦</td><td class=mono>{cell(fc_tls)}</td><td class=mono>{cell(fn_tls)}</td><td class='mono fail'>{cell(acc_tls)}</td></tr>
</table>
<div class=note style="background:#fff1f1;border-color:#f3c7c7;color:#9b2c2c">📌 <b>{pct}</b> —— 延迟几乎全在「中国 → {NODE_IP}(LA)」这一跳,节点出海很快。治本 = 换近节点。</div>"""
else:
breakdown_html = ("""<h2>② 阶段拆解:接入段 vs 出海段</h2>
<div class=note>未取到节点侧出海段数据(控制面 <code>/v1/diag/egress</code> 不可达或未部署)。"""
"""部署诊断端点 / 设 <code>PANGOLIN_API</code> 或 <code>NODE_SSH</code> 后重跑即可拆出接入/出海。</div>""")
metajson = json.dumps(meta, ensure_ascii=False)
return f"""<!DOCTYPE html><html lang=zh-CN><head><meta charset=UTF-8>
<meta name=viewport content="width=device-width,initial-scale=1">
<title>Pangolin 白盒下钻 {started}</title>
<script type="application/json" id="pangolin-run-meta">{metajson}</script>
<style>
body{{font-family:-apple-system,"PingFang SC",Arial,sans-serif;margin:0;background:#f6f7f9;color:#1c2330;line-height:1.6}}
.wrap{{max-width:1100px;margin:0 auto;padding:32px 20px 80px}}
h1{{font-size:24px;margin:0 0 4px}} .sub{{color:#6b7280;margin:0 0 20px}}
h2{{font-size:18px;margin:30px 0 10px;border-bottom:2px solid #e5e7eb;padding-bottom:6px}}
table{{width:100%;border-collapse:collapse;background:#fff;border-radius:10px;overflow:hidden;box-shadow:0 1px 3px rgba(0,0,0,.06);font-size:13px}}
th,td{{text-align:left;padding:8px 10px;border-bottom:1px solid #eef0f3}}
th{{background:#fafbfc;color:#6b7280;font-weight:600;font-size:12px}}
.mono{{font-family:"SF Mono",Menlo,monospace;font-size:12px}}
.pass{{color:#16a34a;font-weight:600}} .warn{{color:#d97706;font-weight:600}} .fail{{color:#dc2626;font-weight:600}}
.card{{background:#fff;border-radius:10px;padding:14px 18px;margin:12px 0;box-shadow:0 1px 3px rgba(0,0,0,.06)}}
.note{{background:#fff8ee;border:1px solid #f5e3c4;border-radius:10px;padding:12px 16px;margin:16px 0;font-size:13px;color:#7a5b25}}
code{{background:#eef0f3;padding:1px 5px;border-radius:4px;font-family:"SF Mono",monospace}}
a{{color:#2563eb;text-decoration:none}}
</style></head><body><div class=wrap>
<h1>Pangolin VPN 白盒下钻</h1>
<p class=sub>{started} · 期望节点 {NODE_IP} · <a href="../index.html">← 返回索引</a></p>
<h2>① 路由判定:国内是不是也走了代理?</h2>
<div class=card>
国外 echo 出口:<b class="mono {'pass' if tf else 'warn'}">{html.escape(str(fip or '取不到'))}</b>
{'(=节点,走隧道 ✓)' if tf else '(≠节点)'} <span class=sub>via {html.escape(str(fsrc or ''))}</span><br>
国内 echo 出口:<b class="mono {'fail' if cnt else 'pass'}">{html.escape(str(cip or '取不到'))}</b>
<span class=sub>via {html.escape(str(csrc or ''))}</span><br>
<div style=margin-top:8px>结论:{cn_verdict}</div>
</div>
<div class=note>原理:国外 echo 反映隧道出口(应=节点);国内 echo 域名是国内站,分流正常时这次请求**直连**,
反映本机真实 ISP 出口。若国内 echo 也 = 节点 IP,说明国内流量被隧道了(<code>split_cn</code> 没开或
<code>geosite-cn</code> 规则未命中)。</div>
{breakdown_html}
<h2>③ 逐站五段拆解:慢在链路哪一段?</h2>
<div class=note>各段=该阶段独立耗时(非累计)。<b>TLS</b> 段高 = 链路 RTT 高(走隧道/出海远);
<b>DNS</b> 高 = 域名解析走了远端(经隧道到 8.8.8.8);<b>TTFB</b> 高 = 目标后端/出海段慢。
末两列(设 <code>NODE_SSH</code> 才有):节点→目标 TLS = <b>出海段</b>;接入段 ≈ 客户端TLS 出海段。</div>
<h3 style=font-size:15px>国外(经隧道)</h3>
<table><tr><th>站点</th><th>域名</th><th>解析IP</th><th>DNS</th><th>TCP</th><th>TLS(延迟)</th><th>TTFB</th><th>总</th><th>节点→目标</th><th>接入段≈</th></tr>
{ftab}</table>
<h3 style=font-size:15px>国内(开分流应直连)</h3>
<table><tr><th>站点</th><th>域名</th><th>解析IP</th><th>DNS</th><th>TCP</th><th>TLS(延迟)</th><th>TTFB</th><th>总</th><th>节点→目标</th><th>接入段≈</th></tr>
{dtab}</table>
<p class=sub style=margin-top:24px>单位 ms。延迟口径见 docs/vpn-test-plan.md。生成工具:scripts/vpn_whitebox.py</p>
</div></body></html>"""
if __name__ == "__main__":
main()