pangolin/scripts/vpn_test.py

#!/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'<script type="application/json" id="pangolin-run-meta">(.*?)</script>',
                          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 '<td class=mono>—</td>'
        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'<td class="mono {c}">{v}{unit}</td>'

    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"<tr onclick=\"location.href='{html.escape(d['_file'])}'\" style=cursor:pointer>"
            f"<td class=mono>{html.escape(d.get('started',''))}</td>"
            f"<td class='{kind_c}'>{kind}</td>"
            f"<td>{html.escape(d.get('host',''))}</td>"
            f"<td class='mono {eg_c}'>{eg}</td>"
            f"<td class=mono>{d.get('foreign_ok','?')}/{d.get('foreign_n','?')}</td>"
            f"<td class=mono>{d.get('domestic_ok','?')}/{d.get('domestic_n','?')}</td>"
            + cell(d.get("foreign_tls_median"), " ms", LAT_GOOD, LAT_WARN)
            + cell(d.get("domestic_tls_median"), " ms", LAT_GOOD, LAT_WARN)
            + cell(dlf, " M", 20, 5, hi_bad=False)
            + cell(dld, " M", 20, 5, hi_bad=False)
            + f"<td><a href='{html.escape(d['_file'])}'>查看 →</a></td></tr>")
    body = "\n".join(rows) or "<tr><td colspan=11>暂无测试记录,跑一次 vpn_test.py 即可。</td></tr>"

    # 架构图:三节点(cara / LA / Google)+ 编号箭头标流向 + 下方步骤表。内联 SVG,离线可看。
    flow_svg = """
<h2 style="font-size:18px;margin:34px 0 10px;border-bottom:2px solid #e5e7eb;padding-bottom:6px">时序图:cara 访问 Google 的流程(时间从上往下)</h2>
<div style="background:#fff;border-radius:10px;padding:18px 16px;box-shadow:0 1px 3px rgba(0,0,0,.06);overflow-x:auto">
<svg viewBox="0 0 1040 400" width="100%" style="min-width:760px;font-family:-apple-system,'PingFang SC',sans-serif">
  <defs>
    <marker id="mo" markerWidth="9" markerHeight="9" refX="7" refY="3" orient="auto"><path d="M0,0 L8,3 L0,6 Z" fill="#c2700c"/></marker>
    <marker id="mg" markerWidth="9" markerHeight="9" refX="7" refY="3" orient="auto"><path d="M0,0 L8,3 L0,6 Z" fill="#16a34a"/></marker>
    <marker id="mr" markerWidth="9" markerHeight="9" refX="7" refY="3" orient="auto"><path d="M0,0 L8,3 L0,6 Z" fill="#dc2626"/></marker>
  </defs>
  <!-- 生命线(虚线) -->
  <line x1="160" y1="54" x2="160" y2="384" stroke="#cbd5e1" stroke-width="1.5" stroke-dasharray="4,4"/>
  <line x1="540" y1="54" x2="540" y2="384" stroke="#cbd5e1" stroke-width="1.5" stroke-dasharray="4,4"/>
  <line x1="900" y1="54" x2="900" y2="384" stroke="#cbd5e1" stroke-width="1.5" stroke-dasharray="4,4"/>
  <!-- 三个角色表头 -->
  <rect x="85" y="8" width="150" height="46" rx="10" fill="#eef4ff" stroke="#c7d8f5"/>
  <text x="160" y="32" text-anchor="middle" font-size="15" font-weight="700" fill="#1c2330">cara</text>
  <text x="160" y="47" text-anchor="middle" font-size="11" fill="#6b7280">中国 · 浏览器</text>
  <rect x="448" y="8" width="184" height="46" rx="10" fill="#fff1e6" stroke="#eab676"/>
  <text x="540" y="30" text-anchor="middle" font-size="15" font-weight="700" fill="#1c2330">LA 节点</text>
  <text x="540" y="46" text-anchor="middle" font-size="10.5" fill="#6b7280">103.119.13.48 · sing-box</text>
  <rect x="826" y="8" width="148" height="46" rx="10" fill="#fff" stroke="#d1d5db"/>
  <text x="900" y="32" text-anchor="middle" font-size="15" font-weight="700" fill="#1c2330">Google</text>
  <text x="900" y="47" text-anchor="middle" font-size="11" fill="#6b7280">8.8.8.8 / 142.251.x</text>
  <!-- 1 cara→LA(隧道) -->
  <text x="350" y="68" text-anchor="middle" font-size="12" fill="#b45309">DNS 查询(经隧道)</text>
  <line x1="160" y1="84" x2="534" y2="84" stroke="#c2700c" stroke-width="2" marker-end="url(#mo)"/>
  <circle cx="350" cy="84" r="12" fill="#c2700c"/><text x="350" y="88.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">1</text>
  <!-- 2 LA→google(出海) -->
  <text x="720" y="112" text-anchor="middle" font-size="12" fill="#15803d">向 8.8.8.8 解析域名</text>
  <line x1="540" y1="128" x2="894" y2="128" stroke="#16a34a" stroke-width="2" marker-end="url(#mg)"/>
  <circle cx="720" cy="128" r="12" fill="#16a34a"/><text x="720" y="132.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">2</text>
  <!-- 3 LA→cara(隧道) -->
  <text x="350" y="156" text-anchor="middle" font-size="12" fill="#b45309">返回 google IP(经隧道)</text>
  <line x1="540" y1="172" x2="166" y2="172" stroke="#c2700c" stroke-width="2" marker-end="url(#mo)"/>
  <circle cx="350" cy="172" r="12" fill="#c2700c"/><text x="350" y="176.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">3</text>
  <!-- 4 cara→LA(隧道·瓶颈) -->
  <text x="350" y="200" text-anchor="middle" font-size="12" font-weight="700" fill="#dc2626">HTTPS 数据 · 接入段 · 瓶颈 ⚠️</text>
  <line x1="160" y1="216" x2="534" y2="216" stroke="#dc2626" stroke-width="2.5" marker-end="url(#mr)"/>
  <circle cx="350" cy="216" r="12" fill="#dc2626"/><text x="350" y="220.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">4</text>
  <!-- 5 LA→google(出海) -->
  <text x="720" y="244" text-anchor="middle" font-size="12" fill="#15803d">节点出海连 google(~43ms)</text>
  <line x1="540" y1="260" x2="894" y2="260" stroke="#16a34a" stroke-width="2" marker-end="url(#mg)"/>
  <circle cx="720" cy="260" r="12" fill="#16a34a"/><text x="720" y="264.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">5</text>
  <!-- 6 google→LA(出海) -->
  <text x="720" y="288" text-anchor="middle" font-size="12" fill="#15803d">google 响应</text>
  <line x1="900" y1="304" x2="546" y2="304" stroke="#16a34a" stroke-width="2" marker-end="url(#mg)"/>
  <circle cx="720" cy="304" r="12" fill="#16a34a"/><text x="720" y="308.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">6</text>
  <!-- 7 LA→cara(隧道) -->
  <text x="350" y="332" text-anchor="middle" font-size="12" fill="#b45309">响应回 cara(经隧道),网页打开</text>
  <line x1="540" y1="348" x2="166" y2="348" stroke="#c2700c" stroke-width="2" marker-end="url(#mo)"/>
  <circle cx="350" cy="348" r="12" fill="#c2700c"/><text x="350" y="352.5" text-anchor="middle" font-size="12.5" font-weight="700" fill="#fff">7</text>
</svg>
<div style="margin-top:6px;font-size:12px;color:#6b7280">图例:时间从上往下;<b style="color:#c2700c">橙=经 REALITY 隧道(接入段,慢)</b> · <b style="color:#16a34a">绿=节点出海(快)</b> · <b style="color:#dc2626">④红=延迟瓶颈</b></div>
<table style="margin-top:16px">
 <tr><th style="width:54px">步骤</th><th style="width:150px">路段</th><th>做什么</th></tr>
 <tr><td class=mono>① 1</td><td class=mono>cara → LA</td><td><b>DNS 查询走隧道</b>:google 是国外域名(不命中 geosite-cn),DNS 请求经 REALITY 隧道发往 LA 节点。</td></tr>
 <tr><td class=mono>② 2</td><td class=mono>LA → Google</td><td>节点把 DNS 查询转给 <code>8.8.8.8</code> 解析,得到 google 的 IP。</td></tr>
 <tr><td class=mono>③ 3</td><td class=mono>LA → cara</td><td>解析出的 google IP 经隧道回传给 cara。</td></tr>
 <tr><td class=mono>④ 4</td><td class=mono>cara → LA</td><td>cara 用该 IP 发起 HTTPS,数据经 REALITY 隧道到节点 —— <b class=warn>接入段,~1000–2700ms,延迟瓶颈</b>(IP 也不命中 geoip-cn,故走隧道)。</td></tr>
 <tr><td class=mono>⑤ 5</td><td class=mono>LA → Google</td><td>节点代 cara 连接 google —— <b class=pass>出海段,~43ms</b>(LA 离 Google 很近)。</td></tr>
 <tr><td class=mono>⑥ 6</td><td class=mono>Google → LA</td><td>google 把响应数据返回给节点。</td></tr>
 <tr><td class=mono>⑦ 7</td><td class=mono>LA → cara</td><td>响应经隧道回到 cara,网页打开。</td></tr>
</table>
<div style="margin-top:12px;background:#e9f9ef;border:1px solid #bce8cd;border-radius:10px;padding:12px 16px;font-size:13px;color:#15603a;line-height:1.8">
 📌 <b>对比 · 国内站(如 baidu)不走这条路</b>:域名命中 geosite-cn → 用 <b>local(223.5.5.5)国内 DNS 直连</b>解析 → 拿到真 CN IP → 命中 geoip-cn → 走 <b>direct 直连</b>,<b>全程不经 LA 节点 / 不进隧道</b> → TLS ~50ms。本次修复正是补上了 DNS 面的分流。
</div>
</div>
"""

    page = 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 测试索引</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:1100px;margin:0 auto;padding:32px 20px 80px}}
 h1{{font-size:24px;margin:0 0 4px}} .sub{{color:#6b7280;margin:0 0 20px}}
 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}}
 tr:hover td{{background:#fafcff}}
 .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}}
 a{{color:#2563eb;text-decoration:none}} a:hover{{text-decoration:underline}}
 .note{{background:#fff8ee;border:1px solid #f5e3c4;border-radius:10px;padding:12px 16px;margin:16px 0;font-size:13px;color:#7a5b25}}
</style></head><body><div class=wrap>
<h1>Pangolin VPN 测试索引</h1>
<p class=sub>共 {len(metas)} 次记录 · 最近更新 {datetime.now().strftime('%Y-%m-%d %H:%M:%S')} · 点行进报告</p>
<div class=note>延迟列 = <b>TLS 握手中位数</b>(&lt;{LAT_GOOD}ms 绿 / &lt;{LAT_WARN}ms 橙 / 更高红)。
 下载列 = Mbps(&gt;20 绿 / &gt;5 橙)。出口绿=走隧道(=节点),橙=非节点。口径见 docs/vpn-test-plan.md。</div>
{flow_svg}
<h2 style="font-size:18px;margin:34px 0 10px;border-bottom:2px solid #e5e7eb;padding-bottom:6px">历次测试记录</h2>
<table>
<tr><th>时间</th><th>类型</th><th>主机</th><th>出口IP</th><th>国外可达</th><th>国内可达</th>
<th>国外延迟</th><th>国内延迟</th><th>国外下载</th><th>国内下载</th><th></th></tr>
{body}
</table>
</div></body></html>"""
    os.makedirs(os.path.dirname(os.path.abspath(index_path)), exist_ok=True)
    with open(index_path, "w", encoding="utf-8") as fp:
        fp.write(page)
    return len(metas)


def render_html(started, eip, egeo, tunneled, f_rows, d_rows, dls, conn=None, meta=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>
<script type="application/json" id="pangolin-run-meta">{json.dumps(meta or {}, ensure_ascii=False)}</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: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} · <a href="../index.html" style="color:#2563eb">← 返回索引</a></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 握手 &lt;{LAT_GOOD}ms 优 / {LAT_GOOD}–{LAT_WARN}ms 偏高 / &gt;{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()