Files
pangolin/scripts/vpn_whitebox.py
T
wangjia 7a4d0cf7cd fix(diag): host 参数走 base64 避免 GFW 拦明文敏感词诊断请求
控制面跑明文 HTTP 在国外 IP 上,白盒诊断请求 URL 里出现 google/youtube 等 GFW
敏感词时被墙重置(返回空)→ 出海段拿不到、报告显示「—」。改:host 用 base64url
传(host_b64),响应也回 b64、不回明文,明文里无敏感词(不影响真实加密隧道流量)。

- diag.go: EgressTiming 支持 ?host_b64=,解码后查白名单,响应 echo b64
- vpn_whitebox.py: node_egress 用 base64url 编码 host

验证:cara(墙内)google/youtube 出海段从 None → 8-9ms,与 github/cloudflare 一致。

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-06-22 14:34:05 +08:00

332 lines
18 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
#!/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"<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()