#!/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, re, glob, statistics, urllib.request
from datetime import datetime
NODE_IP = os.environ.get("NODE_IP", "103.119.13.48")
TIMEOUT = float(os.environ.get("TIMEOUT", "15"))
# 测试产物根目录:每次跑落 runs/<时间>-<主机>.html,并重建 index.html 索引(见 tests/vpn/README.md)。
HERE = os.path.dirname(os.path.abspath(__file__))
TEST_HOME = os.path.normpath(os.path.join(HERE, "..", "tests", "vpn"))
# 站点矩阵(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=None,
help="报告输出路径(默认 tests/vpn/runs/<时间>-<主机>.html)")
ap.add_argument("--runs-dir", default=os.path.join(TEST_HOME, "runs"),
help="run 报告目录(默认 tests/vpn/runs)")
ap.add_argument("--index", default=os.path.join(TEST_HOME, "index.html"),
help="索引页路径(默认 tests/vpn/index.html)")
ap.add_argument("--no-index", action="store_true", help="不重建索引")
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)")
# 输出路径:默认落 runs/<时间>-<主机>.html
host_id = socket.gethostname().split(".")[0]
run_id = datetime.now().strftime("%Y%m%d-%H%M%S") + "-" + host_id
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_meta(run_id, started, host_id, eip, egeo, tunneled, f_rows, d_rows, dls)
html_out = render_html(started, eip, egeo, tunneled, f_rows, d_rows, dls, conn, meta)
with open(out, "w", encoding="utf-8") as fp:
fp.write(html_out)
print(f"\n报告已生成: {os.path.abspath(out)}")
# 重建索引(扫描 runs/ 内嵌 meta,inline 渲染,file:// 直开可用)
if not args.no_index:
try:
n = regenerate_index(args.runs_dir, args.index)
print(f"索引已更新: {os.path.abspath(args.index)} ({n} 条记录)")
except Exception as e:
print(f"索引重建失败(不影响本次报告): {e}")
# 汇总
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 _median(vals):
vals = [v for v in vals if v is not None]
return round(statistics.median(vals), 1) if vals else None
def run_meta(run_id, started, host_id, eip, egeo, tunneled, f_rows, d_rows, dls):
"""本次 run 的摘要,内嵌进报告供索引页读取。"""
dl = {l: m for (l, m, _n, _dt) in dls}
return {
"run_id": run_id, "started": started, "host": host_id,
"egress": eip, "geo": egeo, "tunneled": tunneled,
"foreign_ok": sum(1 for r in f_rows if r["ok"]), "foreign_n": len(f_rows),
"domestic_ok": sum(1 for r in d_rows if r["ok"]), "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": dl,
}
def regenerate_index(runs_dir, index_path):
"""扫描 runs/*.html 内嵌的 meta,重建索引页(inline 数据,file:// 可直开)。返回记录数。"""
metas = []
for f in glob.glob(os.path.join(runs_dir, "*.html")):
try:
with open(f, encoding="utf-8") as fp:
txt = fp.read()
m = re.search(r'',
txt, re.DOTALL)
if not m:
continue
d = json.loads(m.group(1))
d["_file"] = os.path.relpath(f, os.path.dirname(os.path.abspath(index_path)))
metas.append(d)
except Exception:
continue
metas.sort(key=lambda d: d.get("started", ""), reverse=True)
def cell(v, unit="", good=None, warn=None, hi_bad=True):
if v is None:
return '
—
'
c = ""
if good is not None:
if hi_bad:
c = "pass" if v < good else ("warn" if v < warn else "fail")
else:
c = "pass" if v > good else ("warn" if v > warn else "fail")
return f'
{v}{unit}
'
rows = []
for d in metas:
dl = d.get("dl", {})
dlf = next((v for k, v in dl.items() if "国外" in k or "隧道" in k), None)
dld = next((v for k, v in dl.items() if "国内" in k or "直连" in k), None)
eg = html.escape(str(d.get("egress") or "—"))
eg_c = "pass" if d.get("tunneled") else "warn"
kind = "白盒" if d.get("kind") == "whitebox" else "黑盒"
kind_c = "warn" if kind == "白盒" else ""
rows.append(
f"