wangjia
98ba2b411e
- scripts/vpn_test.py:替换 vpn_test.sh。纯标准库(socket/ssl/urllib),覆盖国内外 常见站点矩阵(google/youtube/github/x/fb/ig/wiki/cloudflare/openai/reddit; 百度/腾讯/淘宝/京东/B站/微博/网易/知乎/阿里云),测连通+延迟+TTFB+DNS+出口IP+吞吐, 本机连通性(tun/系统扩展),生成自包含 HTML 报告(-o)。 - 延迟标准 = TLS 握手耗时(ssl 握手,纯链路往返),报告里标注 "TTL=按 TLS 计算、非 IP TTL"; TTFB 作参考。禁用 ping/tcp_connect(经 TUN 本地应答失真)。 - 删除 scripts/vpn_test.sh;文档引用同步更新为 .py。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01JEHzjEcFzvGwgbxT6Wbt6c
290 lines
13 KiB
Python
Executable File
290 lines
13 KiB
Python
Executable File
#!/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"<tr><td>{html.escape(r['name'])}</td><td class=mono>{html.escape(r['host'])}</td>"
|
||
f"<td class=fail colspan=4>不可达 {html.escape(r['err'][:40])}</td></tr>")
|
||
return (f"<tr><td>{html.escape(r['name'])}</td><td class=mono>{html.escape(r['host'])}</td>"
|
||
f"<td>HTTP {r['code']}</td>"
|
||
f"<td class='mono {cls(r['tls_ms'])}'>{r['tls_ms']} ms</td>"
|
||
f"<td class=mono>{r['ttfb_ms']} ms</td>"
|
||
f"<td class=mono>{r['dns_ms']} ms</td></tr>")
|
||
ftab = "\n".join(row(r) for r in f_rows)
|
||
dtab = "\n".join(row(r) for r in d_rows)
|
||
dlrows = "\n".join(
|
||
f"<tr><td>{html.escape(l)}</td><td class='mono {'pass' if m else 'fail'}'>"
|
||
f"{m if m is not None else '失败'} Mbps</td><td class=mono>{n}B / {dt}s</td></tr>"
|
||
for l, m, n, dt in dls)
|
||
egress_cls = "pass" if tunneled else "warn"
|
||
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 VPN 测试报告 {started}</title>
|
||
<style>
|
||
body{{font-family:-apple-system,"PingFang SC",Arial,sans-serif;margin:0;background:#f6f7f9;color:#1c2330;line-height:1.6}}
|
||
.wrap{{max-width:960px;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:32px 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:14px}}
|
||
th,td{{text-align:left;padding:9px 12px;border-bottom:1px solid #eef0f3}}
|
||
th{{background:#fafbfc;color:#6b7280;font-weight:600;font-size:13px}}
|
||
.mono{{font-family:"SF Mono",Menlo,monospace;font-size:13px}}
|
||
.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:14px 18px;margin:16px 0;font-size:13px;color:#7a5b25}}
|
||
code{{background:#eef0f3;padding:1px 5px;border-radius:4px;font-family:"SF Mono",monospace;font-size:.9em}}
|
||
.big{{font-size:15px}}
|
||
</style></head><body><div class=wrap>
|
||
<h1>Pangolin VPN 测试报告</h1>
|
||
<p class=sub>{started} · 期望节点出口 {NODE_IP}</p>
|
||
|
||
<div class=card big>
|
||
出口 IP:<b class="mono {egress_cls}">{html.escape(eip or '取不到')}</b> {html.escape(egeo)}
|
||
— {'<b class=pass>流量走隧道(出口=节点)</b>' if tunneled else '<b class=warn>出口≠节点(可能直连/其他 VPN/DNS 挂)</b>'}
|
||
{('<br>本机:tun(172.19.0.1)=' + ('有' if conn.get('tun') else '无') + ' · 系统扩展=' + (conn.get('sysext') or '未激活')) if conn else ''}
|
||
</div>
|
||
|
||
<div class=note>
|
||
<b>延迟口径(TTL):</b>本表「TTL/延迟」= <b>TLS 握手耗时</b>(TCP 建立完成→TLS 握手完成),
|
||
≈ 真实网络链路往返 × 握手轮数,跨站可比、不含目标服务器后端处理。<b>非 IP TTL</b>;也不用
|
||
<code>ping</code>/<code>tcp_connect</code>(经 TUN 被隧道本地应答,几 ms 假象,不可信)。
|
||
<b>TTFB</b> 为参考 = 请求→首字节(含服务器处理),受目标站自身快慢影响。
|
||
判定:TLS 握手 <{LAT_GOOD}ms 优 / {LAT_GOOD}–{LAT_WARN}ms 偏高 / >{LAT_WARN}ms 高。
|
||
</div>
|
||
|
||
<h2>国外站点(应经隧道)</h2>
|
||
<table><tr><th>站点</th><th>域名</th><th>连通</th><th>TTL/延迟(TLS握手)</th><th>TTFB(参考)</th><th>DNS</th></tr>
|
||
{ftab}</table>
|
||
|
||
<h2>国内站点(开分流应直连)</h2>
|
||
<table><tr><th>站点</th><th>域名</th><th>连通</th><th>TTL/延迟(TLS握手)</th><th>TTFB(参考)</th><th>DNS</th></tr>
|
||
{dtab}</table>
|
||
|
||
<h2>吞吐(下载 ~10MB)</h2>
|
||
<table><tr><th>测速点</th><th>速度</th><th>明细</th></tr>
|
||
{dlrows or '<tr><td colspan=3>(已跳过)</td></tr>'}</table>
|
||
|
||
<p class=sub style=margin-top:32px>方法详见 docs/vpn-test-plan.md / docs/vpn-testing-research.md。生成工具:scripts/vpn_test.py</p>
|
||
</div></body></html>"""
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|