#!/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"{html.escape(r['name'])}{html.escape(r['host'])}" f"不可达 {html.escape(r['err'][:40])}") return (f"{html.escape(r['name'])}{html.escape(r['host'])}" f"HTTP {r['code']}" f"{r['tls_ms']} ms" f"{r['ttfb_ms']} ms" f"{r['dns_ms']} ms") ftab = "\n".join(row(r) for r in f_rows) dtab = "\n".join(row(r) for r in d_rows) dlrows = "\n".join( f"{html.escape(l)}" f"{m if m is not None else '失败'} Mbps{n}B / {dt}s" for l, m, n, dt in dls) egress_cls = "pass" if tunneled else "warn" return f""" Pangolin VPN 测试报告 {started}

Pangolin VPN 测试报告

{started} · 期望节点出口 {NODE_IP}

出口 IP:{html.escape(eip or '取不到')} {html.escape(egeo)} — {'流量走隧道(出口=节点)' if tunneled else '出口≠节点(可能直连/其他 VPN/DNS 挂)'} {('
本机:tun(172.19.0.1)=' + ('有' if conn.get('tun') else '无') + ' · 系统扩展=' + (conn.get('sysext') or '未激活')) if conn else ''}
延迟口径(TTL):本表「TTL/延迟」= TLS 握手耗时(TCP 建立完成→TLS 握手完成), ≈ 真实网络链路往返 × 握手轮数,跨站可比、不含目标服务器后端处理。非 IP TTL;也不用 ping/tcp_connect(经 TUN 被隧道本地应答,几 ms 假象,不可信)。 TTFB 为参考 = 请求→首字节(含服务器处理),受目标站自身快慢影响。 判定:TLS 握手 <{LAT_GOOD}ms 优 / {LAT_GOOD}–{LAT_WARN}ms 偏高 / >{LAT_WARN}ms 高。

国外站点(应经隧道)

{ftab}
站点域名连通TTL/延迟(TLS握手)TTFB(参考)DNS

国内站点(开分流应直连)

{dtab}
站点域名连通TTL/延迟(TLS握手)TTFB(参考)DNS

吞吐(下载 ~10MB)

{dlrows or ''}
测速点速度明细
(已跳过)

方法详见 docs/vpn-test-plan.md / docs/vpn-testing-research.md。生成工具:scripts/vpn_test.py

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