#!/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(节点本机量出海段)。 # host 用 base64url 传(host_b64):控制面是国外 IP 上的明文 HTTP,URL 里出现 # google/youtube 等敏感词会被 GFW 重置;base64 后明文不含敏感词,避免诊断被拦。 import base64 hb = base64.urlsafe_b64encode(host.encode()).rstrip(b"=").decode() try: u = CONTROL_PLANE.rstrip("/") + "/v1/diag/egress?host_b64=" + hb 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"{html.escape(r['name'])}{html.escape(r['host'])}" f"FAIL {html.escape(r['err'][:40])}") 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"{html.escape(r['name'])}{html.escape(r['host'])}" f"{ip}" f"{r['dns_ms']}{r['tcp_ms']}" f"{r['tls_ms']}{r['ttfb_ms']}" f"{r['total_ms']}" f"{node}{access}") ftab = "\n".join(seg_row(r) for r in f_rows) dtab = "\n".join(seg_row(r) for r in d_rows) cn_verdict = ('国内也被隧道(分流失效/未开)' if cnt else ('国内直连(出口≠节点),分流生效' if cip and cip != fip else '数据不足,需重试')) # 阶段拆解:把整轮(客户端)拆成 出海段(节点侧实测)+ 接入段(差值)。用国外站(经隧道)。 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"""

② 阶段拆解:接入段 vs 出海段(国外站 · 经隧道)

把端到端整轮(客户端测)拆成 出海段(节点→目标,控制面 /v1/diag/egress 实测) 与 接入段(客户端→节点的隧道这一跳,= 整轮 − 出海段)—— 对应时序图的橙(接入)/绿(出海)两段。
阶段(时序步骤)整轮 · 客户端出海段 · 节点→目标(绿)接入段 · 隧道(橙,差值)
DNS 解析 ①②③{cell(fc_dns)}{cell(fn_dns)}{cell(acc_dns)}
TLS 握手 ④⑤⑥⑦{cell(fc_tls)}{cell(fn_tls)}{cell(acc_tls)}
📌 {pct} —— 延迟几乎全在「中国 → {NODE_IP}(LA)」这一跳,节点出海很快。治本 = 换近节点。
""" else: breakdown_html = ("""

② 阶段拆解:接入段 vs 出海段

未取到节点侧出海段数据(控制面 /v1/diag/egress 不可达或未部署)。""" """部署诊断端点 / 设 PANGOLIN_API 或 NODE_SSH 后重跑即可拆出接入/出海。
""") metajson = json.dumps(meta, ensure_ascii=False) return f""" Pangolin 白盒下钻 {started}

Pangolin VPN 白盒下钻

{started} · 期望节点 {NODE_IP} · ← 返回索引

① 路由判定:国内是不是也走了代理?

国外 echo 出口:{html.escape(str(fip or '取不到'))} {'(=节点,走隧道 ✓)' if tf else '(≠节点)'} via {html.escape(str(fsrc or ''))}
国内 echo 出口:{html.escape(str(cip or '取不到'))} via {html.escape(str(csrc or ''))}
结论:{cn_verdict}
原理:国外 echo 反映隧道出口(应=节点);国内 echo 域名是国内站,分流正常时这次请求**直连**, 反映本机真实 ISP 出口。若国内 echo 也 = 节点 IP,说明国内流量被隧道了(split_cn 没开或 geosite-cn 规则未命中)。
{breakdown_html}

③ 逐站五段拆解:慢在链路哪一段?

各段=该阶段独立耗时(非累计)。TLS 段高 = 链路 RTT 高(走隧道/出海远); DNS 高 = 域名解析走了远端(经隧道到 8.8.8.8);TTFB 高 = 目标后端/出海段慢。 末两列(设 NODE_SSH 才有):节点→目标 TLS = 出海段;接入段 ≈ 客户端TLS − 出海段。

国外(经隧道)

{ftab}
站点域名解析IPDNSTCPTLS(延迟)TTFB总节点→目标接入段≈

国内(开分流应直连)

{dtab}
站点域名解析IPDNSTCPTLS(延迟)TTFB总节点→目标接入段≈

单位 ms。延迟口径见 docs/vpn-test-plan.md。生成工具:scripts/vpn_whitebox.py

""" if __name__ == "__main__": main()