ykxmz/_tests/test_rescue_section.py

# -*- coding: utf-8 -*-
"""后台「抢救报告」分区 —— 渲染 + 下载 + 不缓存。

用户要求：
    「如果紧急预案未能解决，则将问题打包成 .txt，然后在后台加一个
      分区专门放还有下载这个 txt」

这个测试把两侧都验一遍：
    一、后台页能渲染出「抢救报告」分区（没有报告时给正面提示）
    二、有报告时能列出，并带正确的下载链接
    三、/rescue/<名字> 能下载到，而且**不发长缓存**（和 /download/ 不同）
    四、文件名校验能挡住目录穿越
    五、报告目录空 / 不存在时不能把后台页搞崩
"""
from __future__ import annotations

import importlib.util
import os
import re
import sys
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parent.parent
RELAY_DIR = ROOT / "relay"

passed = 0
failed = 0


def check(name, ok, detail=""):
    global passed, failed
    if ok:
        passed += 1
        print("  [PASS] " + name)
    else:
        failed += 1
        print("  [FAIL] " + name + (("  —— " + str(detail)) if detail else ""))


def load(name: str, path: Path):
    spec = importlib.util.spec_from_file_location(name, path)
    mod = importlib.util.module_from_spec(spec)
    sys.modules[name] = mod
    spec.loader.exec_module(mod)
    return mod


def main() -> int:
    print("=" * 62)
    print("  后台「抢救报告」分区")
    print("=" * 62)

    sys.path.insert(0, str(RELAY_DIR))
    os.chdir(RELAY_DIR)
    try:
        relay = load("relay_under_test", RELAY_DIR / "relay.py")
        admin = load("admin_under_test", RELAY_DIR / "admin.py")
    finally:
        os.chdir(ROOT)

    snap = {"version": "2.0.0", "uptimeSeconds": 123,
            "devices": [], "allowNew": True}

    # ---------------------------------------------------------------- 一
    print("  ---- 一、没有报告时 ----")
    try:
        html = admin.render_admin(snap, "https://x", "/zhongzhuanzhan", 0,
                                 lambda: [], rescue_fn=lambda: [])
        ok = True
        err = ""
    except Exception as exc:                              # noqa: BLE001
        ok = False
        err = "%s: %s" % (type(exc).__name__, exc)
        html = ""
    check("后台页能渲染（不抛异常）", ok, err)
    check("有「抢救报告」分区", "抢救报告" in html)
    check("没有报告时给正面提示（不是干巴巴的暂无数据）",
          "还没有报告" in html, "没找到提示语")
    check("渲染出的页面有实际内容", len(html) > 2000, "%d 字符" % len(html))

    # ---------------------------------------------------------------- 二
    print("  ---- 二、有报告时能列出来 ----")
    items = [
        {"name": "20261008-173000.txt", "size": 4200,
         "mtime": time.time() - 60,
         "summary": "本机是好的，问题在隧道或 Cloudflare 那一侧"},
        {"name": "20261007-220000.txt", "size": 900,
         "mtime": time.time() - 90000, "summary": ""},
    ]
    html2 = admin.render_admin(snap, "https://x", "/zhongzhuanzhan", 0,
                               lambda: [], rescue_fn=lambda: items)
    check("列出了第一份报告的时间", "10-08" in html2 or "2026100" in html2
          or "17:30" in html2)
    check("带下载链接 /rescue/", "/rescue/20261008-173000.txt" in html2,
          "没找到链接")
    check("显示了结论摘要", "问题在隧道" in html2)
    check("没有摘要的那份显示占位文字", "没有结论行" in html2)
    check("两份都列出来了",
          html2.count("/rescue/") >= 2, "只出现 %d 次" % html2.count("/rescue/"))

    # ---------------------------------------------------------------- 三
    print("  ---- 三、文件名校验（防目录穿越）----")
    good = relay._safe_rescue_name("20261008-173000.txt")
    check("正常时间戳名通过", good == "20261008-173000.txt", repr(good))
    for bad in ("../etc/passwd", "a/b.txt", "..\\x.txt", "x.sh",
                "x.txt/../y", "", "a" * 200 + ".txt"):
        check("挡住 %r" % bad[:24], relay._safe_rescue_name(bad) == "",
              "居然通过了")

    # ---------------------------------------------------------------- 四
    print("  ---- 四、下载路由的行为 ----")
    src = (RELAY_DIR / "relay.py").read_text(encoding="utf-8")
    get_part = src.split("def do_GET")[1].split("def do_POST")[0]
    head_part = src.split("def do_HEAD")[1].split("def do_GET")[0]

    def rescue_block(part: str) -> str:
        """取出整个 /rescue/ 分支。

        ⚠ 不能用 part.split("/rescue/")[1] —— "/rescue/" 在这个分支里
          出现两次（startswith 和切片），split 出来的是两个之间的
          短片段，什么都查不到。我第一次就是这么把测试写红的。
          改成从 startswith 那句开始、到下一个 if 分支为止。
        """
        key = 'if path.startswith("/rescue/")'
        i = part.find(key)
        if i < 0:
            return ""
        j = part.find('if path.startswith("', i + len(key))
        return part[i:j] if j > i else part[i:i + 2000]

    gb = rescue_block(get_part)
    hb = rescue_block(head_part)

    check("do_GET 里有 /rescue/ 路由", bool(gb))
    check("do_HEAD 里有 /rescue/ 路由", bool(hb))
    check("下载发 no-store（不能给边缘缓存）", "no-store" in gb,
          "没看到 no-store")
    check("下载设了 Content-Disposition（点了是下载不是预览）",
          "Content-Disposition" in gb)
    check("do_HEAD 那段不写 body", "wfile.write" not in hb,
          "HEAD 里出现了 wfile.write")

    # 鉴权 —— 这条是实测踩出来的安全问题：
    # 报告里有内网 IP、进程表、日志尾部、MTU/路由/iptables，
    # 第一版把路由插在 _admin_gate() 之前，等于匿名可下。
    check("do_GET 里 /rescue/ 要求登录", "_admin_gate" in gb,
          "没有鉴权 -> 任何人都能下诊断报告")
    check("do_HEAD 里 /rescue/ 要求登录", "_admin_gate" in hb,
          "没有鉴权")
    # 顺序也要对：鉴权必须在读文件之前
    if gb:
        check("先鉴权再读文件",
              gb.find("_admin_gate") < gb.find("read_bytes"),
              "顺序反了：先读了文件才鉴权")

    # ⚠ 鉴权失败必须**自己发响应**。_admin_gate() 只返回布尔值，
    #   裸 return 在 BaseHTTPRequestHandler 里就是什么都不做 ——
    #   连接保持、没有响应头，客户端一直等到超时（curl HTTP 000）。
    #   实测踩过：/rescue/ 会挂住 8 秒，比报 401 严重得多。
    # 查整个分支，别用 split("_admin_gate()") 切片 ——
    # 注释里也会提到 _admin_gate()，切出来的可能是注释后面那一小段，
    # 查不到真正的 _json(401。（这个测试第一次就是这么假红的。）
    if gb:
        check("do_GET 鉴权失败会发 401", "_json(401" in gb,
              "裸 return -> 请求会挂到超时")
    if hb:
        check("do_HEAD 鉴权失败会发 401", "no(401)" in hb,
              "裸 return 或状态码不对")

    # ---------------------------------------------------------------- 五
    print("  ---- 五、报告目录不存在也不能崩 ----")
    try:
        r = relay._list_rescue_reports()
        check("_list_rescue_reports() 不抛异常", isinstance(r, list))
    except Exception as exc:                              # noqa: BLE001
        check("_list_rescue_reports() 不抛异常", False,
              "%s: %s" % (type(exc).__name__, exc))

    print()
    print("=" * 62)
    print("  通过 %d 项，失败 %d 项" % (passed, failed))
    return 0 if failed == 0 else 1


if __name__ == "__main__":
    sys.exit(main())