workbuddy-portal/tools/smoke.py

#!/usr/bin/env python
# -*- coding: utf-8 -*-
# SPDX-License-Identifier: MIT
# Copyright (c) 2026 Wang Chuanli

"""离线回归：用 Flask test_client 对**真实库**做全页面只读渲染 + 缺陷防回归断言。

与 tools/check_live.py 的分工：
  * check_live.py  对运行中的服务发真实 HTTP，验「起没起来、登录/CSRF/API 通不通」
  * smoke.py（本脚本）不发网络请求，直接把请求灌进 WSGI 应用，
    因此能覆盖到「页面模板渲染是否正确」，且不需要先起服务、不需要密码。

覆盖内容：
  1. 全页面渲染（含 /users 与 /logs，均需管理员身份）——模板报错会直接暴露成 500
  2. 模板未渲染残留（HTML 里出现 {{ / {% 说明有变量名写错）
  3. 历史缺陷防回归（见 5. 的 ①~⑭）
  4. 多用户：数据隔离 / 凭证保密 / 注册与验证码 / **普通账号的越权面**（4. 与 4b.）
  5. CSV 导出可被标准 csv 解析、列数一致
  6. 页面 HTML 里的 class 与 app.css 的选择器做差集（抓类名拼写错误）
  9. 对外暴露面：下载响应头收敛 / 内部异常不外泄 / 读接口限速 / 安全响应头

权限模型（改断言前先读这一行）：
  普通账号**只能写** `config.USER_EDITABLE_KEYS`（本人的 cookie / user_agent）；
  调度、采集参数、接口地址、注册策略全部只有管理员能写，且一律存在实例级
  `user_id=0`。所以「越权写」的期望结果是 **400**，而不是「写进去但看不到」。

写库说明：会写少量 audit_log 行；另外会**临时**建两个普通账号
（一个用来验权限边界，一个用来走完整注册链路），无论成功失败都在 finally 里删掉。
不会改动任何用量数据。

用法：
    cd workbuddy-portal
    python tools/smoke.py
退出码：0 全通过；1 有失败项。
"""
from __future__ import annotations

import csv
import io
import json
import os
import random
import re
import string
import sys

BASE = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, BASE)

OK = 0
FAIL = 0
FAILS: list[str] = []
NOTES: list[str] = []


def chk(name: str, cond: bool, extra: str = "") -> bool:
    global OK, FAIL
    if cond:
        OK += 1
        print("  [OK]   %s %s" % (name, extra))
    else:
        FAIL += 1
        FAILS.append(name)
        print("  [FAIL] %s %s" % (name, extra))
    return bool(cond)


def note(msg: str) -> None:
    NOTES.append(msg)
    print("  [note] %s" % msg)


def login(cli, uid: int, csrf: str = "smoke-csrf-token"):
    """注入会话绕过登录：GET 不触发 CSRF，因此可直接测页面渲染。

    只有 `uid` 是真正生效的键 —— `security.current_user()` 每请求回查
    users 表（这样做是为了「停用账号立即失效」），所以 `uname/dname/adm`
    只是写给自己看的标记，改不了权限。
    """
    with cli.session_transaction() as s:
        s["uid"] = uid
        s["uname"] = "smoke-%s" % uid
        s["dname"] = "smoke"
        s["adm"] = 0
        s["_csrf"] = csrf


def page(cli, path, method="GET", **kw):
    r = getattr(cli, method.lower())(path, **kw)
    return r.status_code, r.get_data(as_text=True)


def _rand(n=8):
    return "".join(random.choice(string.ascii_lowercase) for _ in range(n))


def run() -> None:
    from workbuddy_portal import captcha, config, create_app, crypto, db, query, security

    print("== 0. 构建应用 ==")
    # 先把真实库迁到与代码同版本。init_db 是幂等的，顺带把迁移路径也验一遍。
    # 少了这一步，「库还停在旧 schema」会被 current_user() 判成「未登录」，
    # 现场表现是下面所有页面断言集体变 302 —— 看起来像产品坏了，
    # 其实只是前置条件没满足。
    db.init_db()
    app = create_app(start_scheduler=False, do_init_db=False)
    app.config["WTF_CSRF_ENABLED"] = False
    n_routes = len([r for r in app.url_map.iter_rules()])
    chk("create_app 成功", app is not None)
    chk("路由数量 >= 40", n_routes >= 40, "routes=%d" % n_routes)

    # 真实库里的账号：管理员必须有，普通账号按需临时造
    conn = db.connect()
    uv = conn.execute("PRAGMA user_version").fetchone()[0]
    chk("库 schema 已与代码同版本", uv == db.DB_SCHEMA_VERSION,
        "uv=%d 期望=%d" % (uv, db.DB_SCHEMA_VERSION))
    admin_row = conn.execute("SELECT id FROM users WHERE is_admin=1 AND status='active'"
                             " ORDER BY id LIMIT 1").fetchone()
    if admin_row is None:
        print("\n[FATAL] 库里没有启用的管理员账号，先跑 python manage.py init")
        return
    ADMIN = admin_row["id"]

    # 实例级配置快照：整轮跑完必须一条不少。
    # 历史缺陷：清理语句写成 `user_id NOT IN (SELECT id FROM users)`，会把
    # user_id=0（实例级）当成孤儿一起删掉 —— 表现为「跑一次 smoke，全实例的
    # 调度/采集参数被重置」，而且不报任何错。所以这里前后各取一次快照。
    inst_before = {r["key"] for r in conn.execute("SELECT key FROM settings WHERE user_id=0")}
    chk("实例级配置非空（否则下面那条断言会空转）", len(inst_before) > 0,
        "keys=%d" % len(inst_before))
    chk("实例级不含凭证类键（cookie/user_agent 恒为个人级）",
        not (inst_before & config.USER_EDITABLE_KEYS),
        "混入=%s" % sorted(inst_before & config.USER_EDITABLE_KEYS))

    # ---------------- 1. 未登录 ----------------
    print("== 1. 未登录：受保护页应跳登录、API 应 401 ==")
    with app.test_client() as cli:
        for p in ("/", "/records", "/tasks", "/config", "/logs", "/users", "/profile"):
            st, _ = page(cli, p)
            chk("GET %-10s 未登录=302" % p, st == 302, "status=%s" % st)
        for p in ("/api/summary", "/api/users", "/api/settings", "/api/audit"):
            st, _ = page(cli, p)
            chk("GET %-14s 未登录=401" % p, st == 401, "status=%s" % st)
        # 只认 POST 的接口：GET 应当 405（而不是落到 401 或被 GET 直接执行）
        # 未登录的 POST 会被 before_request 里的 CSRF 先拦下（400）——
        # 这比先鉴权更好：没有会话就不该被允许碰任何写接口。
        for p in ("/api/profile", "/api/captcha"):
            st, _ = page(cli, p)
            chk("GET %-14s 未登录=405" % p, st == 405, "status=%s" % st)
            st, _ = page(cli, p, method="POST")
            chk("POST %-13s 无 CSRF=400" % p, st == 400, "status=%s" % st)
        st, html = page(cli, "/login")
        chk("登录页含 CSRF 隐藏域", 'name="_csrf"' in html)
        chk("登录页含验证码图", "capimg" in html and 'name="captcha"' in html)
        chk("登录页含自助注册入口", "/register" in html)

    # ---------------- 2. 管理员：全页面渲染 ----------------
    print("== 2. 管理员：全页面渲染 ==")
    with app.test_client() as cli:
        login(cli, ADMIN)
        pages = [
            ("/", "概览"), ("/records", "数据明细"), ("/tasks", "任务管理"),
            ("/config", "配置管理"), ("/logs", "日志管理"), ("/users", "用户管理"),
            ("/profile", "个人中心"), ("/dashboard", "<html"),
        ]
        for p, kw in pages:
            st, html = page(cli, p)
            ok = chk("GET %-10s 200" % p, st == 200, "status=%s len=%d" % (st, len(html)))
            if ok:
                chk("  └ 含关键字 %s" % kw, kw in html)
                chk("  └ 无模板残留 {{ / {%%", "{{" not in html and "{%" not in html)
                chk("  └ 含导航栏", "topbar" in html or p == "/dashboard")

        st, html = page(cli, "/users")
        chk("用户管理页列出账号", 'data-uid=' in html)
        chk("用户管理页含新建表单", 'id="formNewUser"' in html)
        chk("用户管理页含账号操作审计", "账号操作审计" in html)
        chk("用户管理页含状态列", "status" in html and "停用" in html)
        chk("用户管理页含邮箱列", "邮箱" in html)

        st, cfg = page(cli, "/config")
        chk("配置页含维护按钮组", cfg.count("data-maint=") >= 3, "n=%d" % cfg.count("data-maint="))
        chk("配置页含 TLS 校验下拉", 'name="ssl_verify"' in cfg)
        chk("配置页含实例级设置区", "仅管理员可改" in cfg)
        chk("配置页显示凭证状态而非明文", "cookie_hint" in cfg or "字符" in cfg)
        st, rec = page(cli, "/records")
        chk("明细页含快捷区间", 'data-range="today"' in rec and 'data-range="30d"' in rec)
        chk("明细页表格包在 .tablewrap", "tablewrap" in rec)
        st, pf = page(cli, "/profile")
        chk("个人中心含改密表单", "/api/password" in pf or "password" in pf)
        chk("个人中心说明凭证归属本人", "本人凭证" in pf or "我的 Cookie" in pf)
        # 防回归：base.html 里曾用 {% set me = current_user() %}，把子模板的 me
        # 覆盖掉了 —— current_user() 只有 id/username/display_name/is_admin，
        # 于是「注册于」渲染成空。变量已改名 cur，这里把两处都盯住。
        lead = re.search(r'<p class="lead">(.*?)</p>', pf, re.S)
        lead_txt = " ".join(lead.group(1).split()) if lead else ""
        chk("个人中心「注册于」有真实时间",
            bool(re.search(r"注册于\s+\d{4}-\d{2}-\d{2}", lead_txt)), lead_txt[:80])
        chk("个人中心「最近登录」不是空占位",
            bool(re.search(r"最近登录\s+\S", lead_txt)), lead_txt[:80])
        chk("base.html 未再用 me 作局部变量（防覆盖子模板）",
            "set me = " not in open(os.path.join(
                BASE, "workbuddy_portal", "web", "templates", "base.html"),
                encoding="utf-8").read())

    # ---------------- 2b. 静态资源引用可解析 ----------------
    print("== 2b. 页面引用的静态资源全部可达 ==")
    asset_re = re.compile(r"\.(?:js|css|svg|png|jpe?g|gif|webp|ico|woff2?)(?:\?|$)", re.I)
    with app.test_client() as cli:
        login(cli, ADMIN)
        for p in ("/", "/records", "/tasks", "/config", "/logs", "/users",
                  "/profile", "/dashboard"):
            _, html = page(cli, p)
            # 先剥掉 HTML 注释：注释里常写示例路径（src="vendor/x.js"），
            # 不剥会把示例当真实引用误报。
            html = re.sub(r"<!--.*?-->", "", html, flags=re.S)
            refs = set(re.findall(r'src="([^"]+)"', html))
            refs |= {h for h in re.findall(r'href="([^"]+)"', html) if asset_re.search(h)}
            bad, n = [], 0
            for r in sorted(refs):
                if r.startswith(("data:", "http:", "https:", "//", "#")):
                    continue
                target = r
                if not r.startswith("/"):        # 相对路径按该页 URL 解析（曾因此 404）
                    target = (p if p.endswith("/") else p.rsplit("/", 1)[0] + "/") + r
                n += 1
                ast, _ = page(cli, target)
                if ast != 200:
                    bad.append("%s -> %s(%s)" % (r, target, ast))
            chk("%-11s 资源引用全部 200" % p, not bad,
                ("坏引用=%s" % bad) if bad else "%d 个引用" % n)

    # ---------------- 3. 多用户：隔离 / 保密 / 注册与验证码 ----------------
    print("== 3. 多用户：数据隔离 / 凭证保密 / 注册与验证码 ==")
    viewer = "smoke_v_%s" % _rand()
    regged = "smoke_r_%s" % _rand()
    created: list[str] = [viewer]
    saved_ua_inst = None

    def _mk_user(name):
        conn.execute("INSERT INTO users(username,password_hash,display_name,is_admin,"
                     "status,created_at) VALUES(?,?,?,0,'active',?)",
                     (name, security.hash_password("Smoke-Pass1"), "冒烟账号", db.now_str()))
        return conn.execute("SELECT id FROM users WHERE username=?", (name,)).fetchone()["id"]

    try:
        VIEWER = _mk_user(viewer)

        # ① 凭证密文入库
        row = conn.execute("SELECT value FROM settings WHERE key='cookie' AND user_id=?",
                           (ADMIN,)).fetchone()
        if row and row["value"]:
            chk("① Cookie 以密文入库（v1. 前缀）", crypto.is_encrypted(row["value"]),
                "head=%s" % row["value"][:12])
            chk("① 密文不含明文片段",
                crypto.is_encrypted(row["value"]) and ";" not in row["value"][:4])
            plain = db.get_secret(conn, "cookie", ADMIN)
            chk("① 解回来长度合理（>100 字符）", len(plain) > 100, "chars=%d" % len(plain))
        else:
            note("库里没有 Cookie，跳过密文断言")

        # ② 凭证绝不跨账号回落
        chk("② NO_FALLBACK_KEYS 含 cookie/user_agent",
            {"cookie", "user_agent"} <= db.NO_FALLBACK_KEYS)
        chk("② 新账号读不到别人的 Cookie", db.get_secret(conn, "cookie", VIEWER) == "")
        # User-Agent 本身不是秘密，新账号拿到 DEFAULTS 里的**通用** UA 是对的；
        # 要守住的是「不能继承别人存下来的那一份」。用一个哨兵值把这点钉死：
        sentinel = "SMOKE-SENTINEL-UA/%s" % _rand()
        # 注意：get_setting 在没有行时会回落到 DEFAULTS，所以「还原是否成功」
        # 要拿**行为**比（读出来一样），而不是拿「行在不在」比 —— 这两件事
        # 在实例级是分开的，混起来会让断言永远失败。
        before_ua = db.get_setting(conn, "user_agent", "", 0)
        had_row = conn.execute("SELECT 1 FROM settings WHERE user_id=0 AND key='user_agent'"
                               ).fetchone() is not None
        saved_ua_inst = before_ua if had_row else None
        db.set_setting(conn, "user_agent", sentinel, 0)      # 写实例级
        chk("② 实例级放哨兵后，新账号仍看不到它",
            db.get_setting(conn, "user_agent", "", VIEWER) != sentinel,
            "new=%s…" % (db.get_setting(conn, "user_agent", "", VIEWER) or "")[:22])
        chk("② 哨兵在实例级确实生效（证明上面的断言不是在空跑）",
            db.get_setting(conn, "user_agent", "", 0) == sentinel)
        # 还原：**原本没有这一行就删掉**。实例级本不该存在凭证类键
        # （v1.3.0 起 cookie / user_agent 恒为个人级），写回空串只会留下
        # 一个多余行，下次跑就会让「实例级不含凭证键」的断言失败。
        if had_row:
            db.set_setting(conn, "user_agent", before_ua, 0)
        else:
            conn.execute("DELETE FROM settings WHERE user_id=0 AND key='user_agent'")
        chk("② 已还原实例级 UA（读出来与放哨兵前一致）",
            db.get_setting(conn, "user_agent", "", 0) == before_ua)
        chk("② 还原后实例级不留 user_agent 行（原本有则保留）",
            (conn.execute("SELECT 1 FROM settings WHERE user_id=0 AND key='user_agent'"
                          ).fetchone() is not None) == had_row)
        # 普通配置应当能回落到实例级（否则每个新账号都拿到空配置）
        chk("② 普通配置仍回落实例级",
            db.get_setting(conn, "page_size", None, VIEWER) ==
            db.get_setting(conn, "page_size", None, 0))

        # ③ /api/settings 只给掩码，绝不给明文
        with app.test_client() as cli:
            login(cli, ADMIN)
            st, body = page(cli, "/api/settings")
            j = json.loads(body)
            chk("③ /api/settings 200", st == 200, "status=%s" % st)
            plain = db.get_secret(conn, "cookie", ADMIN)
            chk("③ 响应体不含 Cookie 明文", not plain or plain not in body)
            chk("③ cookie 字段被置空", not (j.get("cookie") or "").strip())
            chk("③ 只给 cookie_hint 掩码", "cookie_hint" in j and "cookie_broken" in j)
            chk("③ 标注实例级键清单", isinstance(j.get("_globalKeys"), list) and j["_globalKeys"])
            chk("③ 管理员 _canEditGlobal=True", j.get("_canEditGlobal") is True)
            chk("③ 无 slot:* 内部键", "slot:" not in body)

        # ④ 验证码：不落 session、一次性、出图禁缓存
        with app.test_client() as cli:
            r = cli.get("/captcha.png?purpose=login")
            chk("④ /captcha.png=200", r.status_code == 200, "status=%s" % r.status_code)
            chk("④ 是 PNG 字节流",
                r.headers.get("Content-Type", "").startswith("image/png")
                and r.get_data()[:8] == b"\x89PNG\r\n\x1a\n")
            chk("④ 出图禁缓存", "no-store" in r.headers.get("Cache-Control", ""))
            with cli.session_transaction() as s:
                cid = s.get("cap_login")
            chk("④ 会话里只存验证码 id", bool(cid), "id=%s" % (cid or "无"))
            ans = conn.execute("SELECT answer,purpose FROM captchas WHERE id=?",
                               (cid,)).fetchone() if cid else None
            chk("④ 答案只存在服务端 captchas 表",
                ans is not None and len(ans["answer"]) >= 4 and ans["purpose"] == "login")
            if ans:
                st, html = page(cli, "/login")
                chk("④ 页面 HTML 里搜不到答案", ans["answer"] not in html)
                chk("④ 页面 JS 里也搜不到会话密钥", cid not in html)
            # 一次性：同一个 id 用两次，第二次必须失败
            if ans:
                chk("④ 首次校验通过",
                    captcha.verify(conn, cid, ans["answer"], "login"))
                chk("④ 同 id 二次校验失败（已消费）",
                    not captcha.verify(conn, cid, ans["answer"], "login"))
                chk("④ 消费后记录已删除",
                    conn.execute("SELECT COUNT(*) FROM captchas WHERE id=?",
                                 (cid,)).fetchone()[0] == 0)

        # ⑤ 自助注册全链路（取答案 -> POST /register -> 账号可用）
        with app.test_client() as cli:
            cli.get("/register")
            # 验证码图是浏览器去取的，test_client 不会自动加载 <img>，
            # 所以这里显式打一次 —— 这一步正是「注册页有没有发挑战」的验证
            r = cli.get("/captcha.png?purpose=register")
            chk("⑤ 注册页的验证码接口可用", r.status_code == 200
                and r.get_data()[:4] == b"\x89PNG", "status=%s" % r.status_code)
            with cli.session_transaction() as s:
                cid = s.get("cap_register")
                # 这个客户端没有走 login() 注入固定 token，所以要取真实值；
                # 顺手也证明了 /register 的 CSRF 校验确实在生效
                csrf = s.get("_csrf")
            a2 = conn.execute("SELECT answer,purpose FROM captchas WHERE id=?",
                              (cid,)).fetchone() if cid else None
            chk("⑤ 注册用的挑战落在 register 用途下",
                a2 is not None and a2["purpose"] == "register")
            if a2:
                st, _ = page(cli, "/register", method="POST", data={
                    "username": regged, "display_name": "冒烟注册", "email": "",
                    "password": "Smoke-Pass1", "password2": "Smoke-Pass1",
                    "captcha": a2["answer"]},
                    headers={"X-CSRF-Token": csrf or ""})
                chk("⑤ 注册成功=302", st == 302, "status=%s" % st)
                created.append(regged)
                u = conn.execute("SELECT id,is_admin,status,display_name,last_login_ip"
                                 " FROM users WHERE username=?", (regged,)).fetchone()
                chk("⑤ 建出的是普通账号",
                    u is not None and u["is_admin"] == 0 and u["status"] == "active")
                chk("⑤ 注册即登录（会话已建立）",
                    u is not None and u["id"] == db.user_by_name(conn, regged)["id"])
            # 缺 CSRF 必须 400
            st, _ = page(cli, "/register", method="POST", data={"username": "x" * 3})
            chk("⑤ 注册缺 CSRF=400", st == 400, "status=%s" % st)

        # ⑥ 权限边界：普通账号改不了实例级配置
        with app.test_client() as cli:
            login(cli, VIEWER)
            st, j = page(cli, "/api/settings")
            chk("⑥ 非管理员 _canEditGlobal=False", json.loads(j).get("_canEditGlobal") is False)
            evil = "http://evil.invalid"
            st, _ = page(cli, "/api/settings", method="POST", json={"api_base": evil},
                         headers={"X-CSRF-Token": "smoke-csrf-token"})
            chk("⑥ 非管理员改实例级配置=400", st == 400, "status=%s" % st)
            chk("⑥ 且确实没写进去",
                db.get_setting(conn, "api_base", "", VIEWER) != evil
                and db.get_setting(conn, "api_base", "", 0) != evil)

        # ⑦ 数据隔离：所有查询函数都必须显式带 uid
        try:
            query.daily(conn)
            chk("⑦ query.daily 漏传 uid 会报错", False, "居然没报错")
        except TypeError:
            chk("⑦ query.daily 漏传 uid 会报错", True)
        d_admin = query.daily(conn, ADMIN)
        d_viewer = query.daily(conn, VIEWER)
        chk("⑦ 不同账号的 daily 互不相同",
            not d_admin or d_viewer != d_admin or len(d_viewer) == 0)
        chk("⑦ 新账号 totals 为空", query.totals(conn, VIEWER)["records"] == 0)
        t_admin = query.totals(conn, ADMIN)
        chk("⑦ 管理员 totals 有数据", t_admin["records"] > 0, "records=%d" % t_admin["records"])
        chk("⑦ totals(uid=0) 不含任何人的数据",
            query.totals(conn, 0)["records"] == 0)
    finally:
        # 哨兵 UA 一定要还原（否则下次真采集会带着测试字符串发出去）。
        # 原本实例级没有这一行时，**删掉**而不是写回空串 —— 见第 ② 条断言。
        if saved_ua_inst is not None:
            db.set_setting(conn, "user_agent", saved_ua_inst, 0)
        else:
            conn.execute("DELETE FROM settings WHERE user_id=0 AND key='user_agent'")
        for name in created:
            conn.execute("DELETE FROM users WHERE username=?", (name,))
            # 注意 `user_id<>0` 不能省：user_id=0 是**实例级配置**（调度、采集参数、
            # 接口地址、注册策略都在那里），它不属于任何账号，所以
            # `NOT IN (SELECT id FROM users)` 会把它当孤儿一起删掉 ——
            # 表现成「跑一次 smoke，全实例的配置被重置」，且不报任何错。
            conn.execute("DELETE FROM settings WHERE user_id<>0"
                         " AND user_id NOT IN (SELECT id FROM users)")
            conn.execute("DELETE FROM usage_records WHERE user_id<>0"
                         " AND user_id NOT IN (SELECT id FROM users)")

    # 确认清理干净
    left = conn.execute("SELECT COUNT(*) FROM users WHERE username LIKE 'smoke\\_%' ESCAPE '\\'"
                        ).fetchone()[0]
    chk("3. 临时账号已清理", left == 0, "残留=%d" % left)

    # ---------------- 4. 普通账号的权限边界 ----------------
    # 规则只有一条（config.writable_by）：普通账号只能写本人的 cookie / user_agent，
    # 其余（调度、采集参数、接口地址、注册策略）一律 400。页面隐藏 / disabled
    # 只是「不给出误导性按钮」，真正的闸门在服务端，所以这里全部走真实请求。
    print("== 4. 非管理员：越权面必须全部关死 ==")
    viewer2 = "smoke_w_%s" % _rand()
    try:
        V2 = _mk_user(viewer2)
        with app.test_client() as cli:
            login(cli, V2)
            st, html = page(cli, "/users")
            chk("GET /users 非管理员=403", st == 403, "status=%s" % st)
            st, _ = page(cli, "/api/users")
            chk("GET /api/users 非管理员=403", st == 403, "status=%s" % st)
            # ④ 日志是**实例级**运行信息（含数据库路径 / 账号名 / 来源 IP），
            #    普通账号整页 403 —— 不是「只看到自己那份」。
            st, _ = page(cli, "/logs")
            chk("GET /logs 非管理员=403", st == 403, "status=%s" % st)
            st, _ = page(cli, "/logs/tail?lines=10")
            chk("GET /logs/tail 非管理员=403", st == 403, "status=%s" % st)
            # ⑤ 其余页面（都只渲染本人数据）必须照常能开
            for p in ("/", "/dashboard", "/records", "/tasks", "/config", "/profile"):
                st, _ = page(cli, p)
                chk("GET %-11s 非管理员=200" % p, st == 200, "status=%s" % st)
            st, html = page(cli, "/")
            chk("概览导航不含「用户管理」", "用户管理" not in html)
            chk("概览导航不含「日志管理」", "日志管理" not in html)
            chk("普通账号导航含「个人中心」入口", 'class="who"' in html)
            # ⑥ 越权写：调度 / 采集参数 / 实例级键，逐个试，全部必须 400
            keep_times = db.get_setting(conn, "schedule_times", "", 0)
            for key, val in (("schedule_times", "23:59"),
                             ("schedule_enabled", "0"),
                             ("catch_up", "0"),
                             ("page_size", "1000"),
                             ("timeout", "300"),
                             ("ssl_verify", "0"),
                             ("max_prompt", "0"),
                             ("api_base", "http://evil.invalid"),
                             ("allow_register", "0")):
                st, body = page(cli, "/api/settings", method="POST", json={key: val},
                                headers={"X-CSRF-Token": "smoke-csrf-token"})
                chk("⑥ 越权写 %-16s =400" % key, st == 400, "status=%s" % st)
                chk("   └ 报错里点名 %s" % key, key in body)
            chk("⑥ 越权尝试确实没落库（schedule_times 未变）",
                db.get_setting(conn, "schedule_times", "", 0) == keep_times)
            chk("⑥ 实例级 api_base 未被改写",
                "evil" not in db.get_setting(conn, "api_base", "", 0))
            # ⑦ 但本人凭证必须写得进去（否则普通账号根本没法采集）
            st, _ = page(cli, "/api/settings", method="POST",
                         json={"user_agent": "SMOKE-VIEWER-UA/1.0"},
                         headers={"X-CSRF-Token": "smoke-csrf-token"})
            chk("⑦ 普通账号写本人 user_agent=200", st == 200, "status=%s" % st)
            chk("⑦ 且只写进了自己名下",
                db.get_setting(conn, "user_agent", "", V2) == "SMOKE-VIEWER-UA/1.0")
            # ⑧ 任务页给普通账号渲染的是只读表，且没有「保存调度配置」按钮
            st, tk = page(cli, "/tasks")
            chk("⑧ 任务页标注调度只读", "仅管理员可改" in tk)
            chk("⑧ 任务页无调度保存按钮", "保存调度配置" not in tk)
            # ⑨ 配置页对普通账号只给凭证表单，采集参数渲染成只读表
            st, cf = page(cli, "/config")
            chk("⑨ 配置页有凭证表单", 'id="formCred"' in cf)
            chk("⑨ 配置页无采集参数表单", 'id="formCollect"' not in cf)
            chk("⑨ 配置页无实例级设置表单", 'id="formGlobal"' not in cf)
            chk("⑨ 配置页说明范围", "只有这一块可改" in cf)
            # ⑩ /api/status 的角色字段（大屏与前端靠它显隐管理员入口）
            st, sj = page(cli, "/api/status")
            chk("⑩ GET /api/status 普通账号=200", st == 200, "status=%s" % st)
            if st == 200:
                j = json.loads(sj)
                chk("⑩ is_admin=False", j.get("is_admin") is False, "%s" % j.get("is_admin"))
                chk("⑩ can_edit_schedule=False", j.get("can_edit_schedule") is False)
                chk("⑩ can_view_logs=False", j.get("can_view_logs") is False)
    finally:
        conn.execute("DELETE FROM users WHERE username=?", (viewer2,))
        # `user_id<>0` 是必须的：0 是实例级配置，不能当孤儿清理（见第 3 节的说明）
        conn.execute("DELETE FROM settings WHERE user_id<>0"
                     " AND user_id NOT IN (SELECT id FROM users)")

    # ---------------- 4b. 全局键的落库位置 ----------------
    # 这一节盯的是「管理员改了但只有自己生效」这类**静默** bug：
    # 全局键若被写进管理员的 user_id，其它账号读取时会回落到 DEFAULTS，
    # 表现成「设置莫名其妙不生效」，而且不报任何错。
    print("== 4b. 全局键必须落在实例级 user_id=0 ==")
    chk("schedule_times 是全局键", config.is_global_key("schedule_times"))
    chk("page_size 是全局键", config.is_global_key("page_size"))
    chk("cookie 不是全局键（本人凭证）", not config.is_global_key("cookie"))
    chk("slot:* 仍是个人级（每人各自记今天跑过没）",
        not config.is_global_key("slot:09:00"))
    chk("普通账号只被允许写 cookie/user_agent",
        config.writable_by("cookie", False) and config.writable_by("user_agent", False)
        and not config.writable_by("schedule_times", False)
        and not config.writable_by("page_size", False))
    with app.test_client() as cli:
        login(cli, ADMIN)
        same = db.get_setting(conn, "schedule_times", "", 0)
        st, _ = page(cli, "/api/settings", method="POST",
                     json={"schedule_times": same or "09:00"},
                     headers={"X-CSRF-Token": "smoke-csrf-token"})
        chk("管理员写 schedule_times=200", st == 200, "status=%s" % st)
        chk("只存在实例级那一份",
            conn.execute("SELECT COUNT(*) FROM settings WHERE user_id=0"
                         " AND key='schedule_times'").fetchone()[0] == 1)
        chk("管理员名下不留个人级副本（否则别人读不到）",
            conn.execute("SELECT COUNT(*) FROM settings WHERE user_id<>0"
                         " AND key='schedule_times'").fetchone()[0] == 0)
        # /api/status 是大屏与前端判断角色用的接口，必须真的能开且角色正确
        # （它曾经因为改字段时引用了未定义的变量而 500，两层测试都没覆盖到）
        st, sj = page(cli, "/api/status")
        chk("GET /api/status 管理员=200", st == 200, "status=%s" % st)
        if st == 200:
            j = json.loads(sj)
            chk("└ is_admin=True", j.get("is_admin") is True, "%s" % j.get("is_admin"))
            chk("└ can_edit_schedule=True", j.get("can_edit_schedule") is True)
            chk("└ 凭证只回「有没有 / 多少字符」",
                all(k in j for k in ("cookie_set", "cookie_chars", "cookie_broken"))
                and "cookie" not in j)
    # 收尾自检：实例级配置必须还在（对照开头那份快照）
    inst_after = {r["key"] for r in conn.execute("SELECT key FROM settings WHERE user_id=0")}
    chk("跑完整轮 smoke 后，实例级配置一条不少", inst_before <= inst_after,
        "丢失=%s" % sorted(inst_before - inst_after))
    conn.close()

    # ---------------- 5. 历史缺陷防回归 ----------------
    print("== 5. 历史缺陷防回归 ==")
    with app.test_client() as cli:
        login(cli, ADMIN)
        # ① 非法日期曾 500
        st, body = page(cli, "/api/summary?from=abc&to=def")
        chk("① /api/summary 非法日期=400", st == 400, "status=%s" % st)
        chk("   └ 返回 JSON 错误体", '"ok": false' in body.replace('":', '": '))
        # ② /tasks 非法页码曾 500
        st, _ = page(cli, "/tasks?page=abc")
        chk("② /tasks?page=abc=200", st == 200, "status=%s" % st)
        # ③ 日志尾部非法行数曾 500
        st, _ = page(cli, "/logs/tail?lines=abc")
        chk("③ /logs/tail?lines=abc=200", st == 200, "status=%s" % st)
        # ④ 内部簿记键 slot:* 曾泄漏到 /api/settings
        st, body = page(cli, "/api/settings")
        chk("④ /api/settings 无 slot:* 键", "slot:" not in body, "status=%s" % st)
        # ⑤ 概览「云端」列曾因 SQL 少选列而恒为空
        st, ov = page(cli, "/")
        chk("⑤ 概览含「云端」列", "云端" in ov)
        # ⑥ 明细页日期回填：模板曾读 f.from，导致输入框永远为空
        st, rec = page(cli, "/records?from=2026-09-08&to=2026-09-10")
        chk("⑥ 明细页回填 from", 'value="2026-09-08"' in rec)
        chk("⑥ 明细页回填 to", 'value="2026-09-10"' in rec)
        # ⑦ 导出链接必须带规范参数名 from（模板内部用 frm，拼 URL 时要换回来）
        m = re.search(r'href="(/records/export[^"]*)"', rec)
        chk("⑦ 导出链接存在", bool(m))
        if m:
            chk("⑦ 导出链接带 from=", "from=2026-09-08" in m.group(1), m.group(1))
            chk("⑦ 导出链接带 to=", "to=2026-09-10" in m.group(1))
        # ⑧ 导出接口本身也要认 from/to
        st, csv_body = page(cli, "/records/export?from=2026-09-08&to=2026-09-10")
        chk("⑧ GET /records/export=200", st == 200, "status=%s" % st)
        rows = list(csv.reader(io.StringIO(csv_body.lstrip("\ufeff"))))
        chk("⑧ CSV 至少含表头", len(rows) >= 1, "rows=%d" % len(rows))
        chk("⑧ CSV 列数一致",
            len({len(r) for r in rows if r}) == 1,
            "列数集合=%s" % sorted({len(r) for r in rows if r}))
        chk("⑧ CSV 表头为官方同构列",
            rows and rows[0] == ["RequestID", "积分消耗", "User Prompt", "模型", "客户端", "时间"],
            "header=%s" % (rows[0] if rows else None))
        # ⑨ 非法设置必须在写入时被拒（曾让采集崩掉）
        st, body = page(cli, "/api/settings", method="POST", json={"page_size": "abc"},
                        headers={"X-CSRF-Token": "smoke-csrf-token"})
        chk("⑨ 非法设置写入=400", st == 400, "status=%s" % st)
        st, body = page(cli, "/api/settings", method="POST", json={"slot:09:00": "x"},
                        headers={"X-CSRF-Token": "smoke-csrf-token"})
        chk("⑨ 内部键被忽略而非写入",
            st == 200 and "slot:09:00" in (json.loads(body).get("ignored") or []),
            "status=%s body=%s" % (st, body[:140]))
        # ⑩ 维护动作
        st, _ = page(cli, "/api/maintenance/nope", method="POST", json={},
                     headers={"X-CSRF-Token": "smoke-csrf-token"})
        chk("⑩ 未知维护动作=404", st == 404, "status=%s" % st)
        st, body = page(cli, "/api/maintenance/recount", method="POST", json={},
                        headers={"X-CSRF-Token": "smoke-csrf-token"})
        chk("⑩ recount=200", st == 200, "status=%s body=%s" % (st, body[:90]))
        # ⑪ 管理员不能删自己
        st, _ = page(cli, "/api/users/%d/delete" % ADMIN, method="POST", json={},
                     headers={"X-CSRF-Token": "smoke-csrf-token"})
        chk("⑪ 删除自己=400（不允许）", st == 400, "status=%s" % st)
        # ⑫ CSRF 缺失必须 400
        st, _ = page(cli, "/api/settings", method="POST", json={"max_prompt": "100"})
        chk("⑫ 缺 CSRF=400", st == 400, "status=%s" % st)
        # ⑬ 分页/筛选参数非法不能 500
        for p in ("/logs?page=abc", "/logs?apage=abc", "/records?page=abc&size=abc",
                  "/records?from=2026-13-99", "/logs?status=%27%20OR%201=1--"):
            st, _ = page(cli, p)
            chk("⑬ GET %-30s =200" % p, st == 200, "status=%s" % st)
        # ⑭ 审计筛选：只返回指定动作，且计数与筛选一致
        st, lg = page(cli, "/logs")
        chk("⑭ 日志页含操作审计筛选", "操作审计" in lg and "seg" in lg)
        m = re.search(r'href="/logs\?act=([^"&]+)', lg)
        if chk("⑭ 审计筛选有可选动作", bool(m), "act=%s" % (m.group(1) if m else "无")):
            act = m.group(1)
            st, lg2 = page(cli, "/logs?act=" + act)
            chk("⑭ 带 act=%s 仍 200" % act, st == 200, "status=%s" % st)
            # 从审计表的 id 处切片：前面采集表里的 trigger 也用了 tag mute，不能混入
            i = lg2.find('id="auditTable"')
            tail = lg2[i:] if i >= 0 else ""
            chk("⑭ 审计表存在", i >= 0)
            tags = re.findall(r'<td><span class="tag mute">([^<]+)</span></td>', tail)
            others = sorted({t for t in tags if t != act})
            chk("⑭ 审计筛选结果不含其他动作", not others and bool(tags),
                "命中=%d 混入=%s" % (len(tags), others))

    # ---------------- 6. 数据自洽 ----------------
    print("== 6. 数据自洽（只读） ==")
    with app.test_client() as cli:
        login(cli, ADMIN)
        mf = json.loads(page(cli, "/api/manifest")[1])
        src = (mf.get("sources") or [{}])[0]
        chk("manifest 存档条数 == 数据源条数",
            mf["totals"]["records"] == src.get("count"),
            "%s vs %s" % (mf["totals"]["records"], src.get("count")))
        sm = json.loads(page(cli, "/api/summary")[1])
        chk("summary 全量 calls == 存档条数",
            sm.get("calls") == mf["totals"]["records"],
            "calls=%s records=%s" % (sm.get("calls"), mf["totals"]["records"]))
        chk("summary 全量 credits 自洽",
            abs(float(sm.get("credits", 0)) - float(mf["totals"]["credits"])) < 0.005,
            "%s vs %s" % (sm.get("credits"), mf["totals"]["credits"]))
        d = query.daily(db.get_db(), ADMIN)
        chk("daily 逐日积分求和 == 存档总额",
            abs(round(sum(float(x["c"]) for x in d), 2)
                - round(float(mf["totals"]["credits"]), 2)) < 0.005)
        chk("daily 逐日 h[24] 求和 == 当日积分",
            all(abs(round(sum(x["h"]), 2) - round(x["c"], 2)) < 0.005 for x in d))
        note("管理员存档 %s 条 / %s 积分 / %d 天" % (mf["totals"]["records"],
                                                 mf["totals"]["credits"], len(d)))

    # ---------------- 7. class 名与 CSS 选择器对账 ----------------
    print("== 7. 页面 class 与 app.css 选择器对账 ==")
    css = open(os.path.join(BASE, "workbuddy_portal", "web", "static", "css", "app.css"),
               encoding="utf-8").read()
    css_classes = set(re.findall(r"\.([A-Za-z][\w-]*)", css))
    anon = ("/login", "/register")
    auth = ("/", "/records", "/tasks", "/config", "/logs", "/users", "/profile")
    used: set[str] = set()
    for p in anon:
        with app.test_client() as c2:
            html = page(c2, p)[1]
        for m in re.findall(r'class="([^"]*)"', html):
            used.update(t for t in m.split() if t)
    with app.test_client() as cli:
        login(cli, ADMIN)
        for p in auth:
            html = page(cli, p)[1]
            for m in re.findall(r'class="([^"]*)"', html):
                used.update(t for t in m.split() if t)
    # 允许的无样式类：JS 钩子、第三方/语义标记
    allow = {"no-js", "on", "cur", "gap", "meta", "unit", "field-err"}
    missing = sorted(c for c in used - css_classes - allow)
    chk("无「用了但 CSS 里不存在」的类名", not missing, "缺失=%s" % missing if missing else "")

    # ---------------- 8. 备份 ----------------
    # 这一节盯三件事：
    #   ① `safe_name` 是下载/恢复接口**唯一**吃文件名的收口点。漏检就是任意文件
    #      读取 —— `../../data/instance.json` 能直接把主密钥拿走；
    #   ② 归档里必须同时有 manifest / 主库 / instance.json。少了 instance.json，
    #      settings 里的凭证密文就永远解不开了（cookie_key 在里面）；
    #   ③ 备份是管理员专属能力，普通账号连列表都不能看。
    # 「破坏 → 恢复 → 比对」的往返**不在这里做**：它要整库替换，不适合对真实库
    # 执行，由隔离环境里的专项验证覆盖（见 docs/DEPLOYMENT.md 8.3）。
    print("== 8. 备份：路径收口 / 归档完整 / 越权面 ==")
    import shutil as _sh
    import tempfile as _tf
    import zipfile as _zf
    from workbuddy_portal import backup

    for bad in ("../../etc/passwd", "x.txt", "", "..", "a b.zip", "a.zip/../../x.zip"):
        try:
            got = backup.safe_name(bad)
            # `a.zip/../../x.zip` 这类会被 basename 收敛成合法的 `x.zip`。
            # 收敛不算漏检，但结果里**必须**不含任何路径成分。
            bad_ok = bool(got) and got == os.path.basename(got) \
                and "/" not in got and "\\" not in got and ".." not in got
        except backup.BackupError:
            bad_ok = True
        chk("⑧ safe_name 收口 %r" % bad, bad_ok)

    real_dir = config.BACKUP_DIR
    tmp_bk = _tf.mkdtemp(prefix="wb-smoke-bk-")
    # 第 5 节末尾把 conn 关掉了（那是它自己的收尾动作），这里重新拿一个独立的。
    bconn = db.connect()
    seen_before = {r["filename"] for r in bconn.execute("SELECT filename FROM backups")}
    try:
        config.BACKUP_DIR = tmp_bk               # 别把真实 backups/ 搅乱
        res = backup.create(bconn, "manual", "smoke", "smoke 断言")
        ap = backup.path_of(res["filename"])
        chk("⑧ 备份生成成功", bool(res["ok"]) and os.path.exists(ap), res["message"])
        with _zf.ZipFile(ap) as z:
            names = set(z.namelist())
        chk("⑧ 归档含 manifest.json", backup.MANIFEST in names)
        chk("⑧ 归档含主库快照", os.path.basename(config.SQLITE_PATH) in names)
        chk("⑧ 归档含 instance.json（否则 cookie_key 丢失）",
            backup.INSTANCE_NAME in names)

        v = backup.verify(ap)
        chk("⑧ verify 通过", v["integrity"] == "ok",
            "uv=%s 条数=%s 账号=%s" % (v["schema_ver"], v["records"], v["users"]))
        chk("⑧ verify 报的条数与库一致",
            v["records"] == bconn.execute("SELECT COUNT(*) FROM usage_records").fetchone()[0],
            "records=%s" % v["records"])

        backup.create(bconn, "manual", "smoke", "第二份")
        backup.sync_index(bconn)
        backup.prune(bconn, keep=1, actor="smoke")
        left = [f for f in os.listdir(tmp_bk) if f.endswith(backup.SUFFIX)]
        chk("⑧ prune keep=1 后只剩 1 份", len(left) == 1, "剩=%d" % len(left))

        tmp_user = "smoke_b_%s" % _rand()
        bconn.execute("INSERT INTO users(username,password_hash,display_name,is_admin,"
                      "status,created_at) VALUES(?,?,'备份越权探针',0,'active',?)",
                      (tmp_user, security.hash_password("Smoke-Pass1"), db.now_str()))
        bid = bconn.execute("SELECT id FROM users WHERE username=?", (tmp_user,)).fetchone()["id"]
        try:
            with app.test_client() as cb:
                login(cb, bid)
                for p in ("/backups", "/api/backups"):
                    st, _ = page(cb, p)
                    chk("⑧ 普通账号 GET %s 被拒" % p, st in (403, 401), "status=%s" % st)
                # 导出是 zip，不能走 page()（它按文本解码，会炸在二进制上）
                er = cb.get("/profile/export")
                body = er.get_data()
                try:
                    with _zf.ZipFile(io.BytesIO(body)) as z:
                        zn = z.namelist()
                except Exception:                   # noqa: BLE001
                    zn = []
                chk("⑧ 普通账号可导出本人数据（zip）",
                    er.status_code == 200 and len(zn) >= 4,
                    "状态=%s 条目=%s" % (er.status_code, zn))
                chk("⑧ 导出包里没有 cookie 明文",
                    not any("cookie" in n.lower() for n in zn))
        finally:
            bconn.execute("DELETE FROM users WHERE id=?", (bid,))
    finally:
        config.BACKUP_DIR = real_dir
        _sh.rmtree(tmp_bk, ignore_errors=True)
        # 归档文件已经随临时目录没了，登记行留着就是脏数据（列表里会显示「已丢失」）
        fresh = [r["filename"] for r in bconn.execute("SELECT filename FROM backups")
                 if r["filename"] not in seen_before]
        for n in fresh:
            bconn.execute("DELETE FROM backups WHERE filename=?", (n,))
        bconn.commit()
        bconn.close()

    # ---------------- 9. 对外暴露面（安全与隐私基线） ----------------
    # 这一节盯的是「直接挂到公网」时最容易漏掉的三类东西：
    #   ① 下载文件名 / 响应头注入 —— 用户名并不总是注册正则的产物
    #      （manage.py passwd 建号、老库升级上来的名字都可能带引号或 CR/LF）；
    #   ② 内部异常直出 —— 原始异常文本会带绝对路径与 SQL 片段，是踩点的现成材料；
    #   ③ 读接口没有刹车 —— 一个注册账号循环调 /api/bundle 就能持续吃满 CPU。
    print("== 9. 对外暴露面：响应头收敛 / 异常不外泄 / 读接口限速 ==")

    for bad in ('a"b.csv', "a\r\nX-Evil: 1", "../../etc/passwd", "a\\b.zip", "..",
                ".", "   ", "", "中文\n名.csv"):
        got = security.safe_filename(bad, fallback="download")
        chk("⑨ safe_filename 收敛 %r" % bad,
            bool(got) and not any(c in got for c in '"\\\r\n/') and ".." not in got,
            "-> %r" % got)

    cd = security.content_disposition('a"b\r\nX-Evil: 1.zip')
    quoted = cd.split('filename="', 1)[1].split('"', 1)[0] if 'filename="' in cd else "?"
    chk("⑨ content_disposition 的 ASCII 名段无引号/换行/路径",
        not any(c in quoted for c in '"\\\r\n/'), "-> %r" % cd[:80])
    chk("⑨ content_disposition 带 RFC 5987 原名段", "filename*=UTF-8''" in cd)

    # 端到端：直接造一个「非法用户名」的账号，看真实导出响应头是否仍然干净。
    # 走 SQL 插入而不是接口，正是为了模拟「老库里已经有这种名字」的现状。
    api_src = open(os.path.join(BASE, "workbuddy_portal", "web", "api.py"),
                   encoding="utf-8").read()
    evil = 'smoke_ev"%s\\x' % _rand(4)
    nconn = db.connect()
    try:
        nconn.execute("INSERT INTO users(username,password_hash,display_name,is_admin,"
                      "status,created_at) VALUES(?,?,'越权探针',0,'active',?)",
                      (evil, security.hash_password("Smoke-Pass1"), db.now_str()))
        eid = nconn.execute("SELECT id FROM users WHERE username=?", (evil,)).fetchone()["id"]
        with app.test_client() as ce:
            login(ce, eid)
            for path, what in (("/records/export", "按明细导出"), ("/profile/export", "个人数据包")):
                h = ce.get(path).headers.get("Content-Disposition", "")
                chk("⑨ %s 响应头已收敛" % what,
                    bool(h) and "\r" not in h and "\n" not in h and 'filename="' in h,
                    "-> %r" % h[:80])
    finally:
        nconn.execute("DELETE FROM users WHERE username=?", (evil,))
        nconn.commit()
        nconn.close()

    # ② 接口层的兜底 catch 不能再把原始异常文本回给客户端。
    #    注意断言的是「兜底分支」而不是「所有 str(e)」：BadParam / Busy /
    #    NotReady / ApiError / BackupError 这些都是**面向用户写的**业务异常，
    #    文案本身就是要给用户看的，把它们也禁掉是过度收紧。
    #    真正会带出绝对路径与 SQL 的只有 `except Exception` 那一条路。
    generic = re.findall(r"except Exception as e:[^\n]*\n([^\n]*)", api_src)
    chk("⑨ 每个 except Exception 都走 _internal（不回原始异常文本）",
        bool(generic) and all("_internal(" in g for g in generic),
        "n=%d" % len(generic))

    # ③ 读接口限速：把阈值临时调小，验机制本身（真阈值在 security._API_RATE_MAX）
    chk("⑨ 读接口阈值给得足够宽松（不会误伤正常翻页）",
        security._API_RATE_MAX >= 60, "max=%d/60s" % security._API_RATE_MAX)
    old_max = security._API_RATE_MAX
    try:
        security._API_RATE_MAX = 5
        security.api_rate_reset()
        with app.test_client() as cr:
            login(cr, ADMIN)
            codes = [page(cr, "/api/status")[0] for _ in range(6)]
        chk("⑨ 读接口限速：前 5 次放行", codes[:5] == [200] * 5, "codes=%s" % codes)
        chk("⑨ 读接口限速：第 6 次 429", codes[5] == 429, "codes=%s" % codes)
    finally:
        security._API_RATE_MAX = old_max
        security.api_rate_reset()

    # ④ 安全响应头：对外部署时这些是扫描器与浏览器共同依赖的基线
    with app.test_client() as ch_:
        h = ch_.get("/login").headers
    for name in ("Content-Security-Policy", "Permissions-Policy",
                 "X-Content-Type-Options", "X-Frame-Options", "Referrer-Policy"):
        chk("⑨ 响应头 %s 已下发" % name, name in h)
    chk("⑨ 会话 Cookie 标了 HttpOnly", "HttpOnly" in h.get("Set-Cookie", ""))

    # ⑥ 内部实现细节不下发给普通账号：表名、库文件路径这些是踩点材料，
    #    普通账号看自己的数据，没有任何理由知道正本落在哪、叫什么表。
    probe = "smoke_p_%s" % _rand()
    nconn2 = db.connect()
    try:
        nconn2.execute("INSERT INTO users(username,password_hash,display_name,is_admin,"
                       "status,created_at) VALUES(?,?,'隐私探针',0,'active',?)",
                       (probe, security.hash_password("Smoke-Pass1"), db.now_str()))
        pid = nconn2.execute("SELECT id FROM users WHERE username=?",
                             (probe,)).fetchone()["id"]
        with app.test_client() as cu:
            login(cu, pid)
            ju = json.loads(page(cu, "/api/manifest")[1])
        chk("⑨ 普通账号拿不到「数据源」清单", ju.get("sources") == [],
            "sources=%s" % ju.get("sources"))
        chk("⑨ 普通账号拿不到库文件路径", not ju.get("archive"),
            "archive=%r" % ju.get("archive"))
        chk("⑨ 但仍能拿到自己的存档总量（不是把整块砍掉）",
            ju.get("totals", {}).get("records") == 0, "records=%s"
            % ju.get("totals", {}).get("records"))
        with app.test_client() as ca:
            login(ca, ADMIN)
            ja = json.loads(page(ca, "/api/manifest")[1])
        chk("⑨ 管理员仍能看到「数据源」清单", bool(ja.get("sources")),
            "n=%d" % len(ja.get("sources") or []))
    finally:
        nconn2.execute("DELETE FROM users WHERE username=?", (probe,))
        nconn2.commit()
        nconn2.close()

    # ⑤ 大屏页的数据插值必须全部转义（这里只钉住这次修掉的那几处，
    #    它们是「模型名 / 客户端名来自上游接口」这条链路上最短的几根引线）
    dash = open(os.path.join(BASE, "workbuddy_portal", "web", "static", "dashboard",
                             "index.html"), encoding="utf-8").read()
    for probe in ("<b>${p.name}", "<b>${d.name}", "${x.model}</td>", "${x.client}</td>"):
        chk("⑨ 大屏页已转义 %s" % probe, probe not in dash)


def main() -> int:
    print("工程目录：%s\n" % BASE)
    try:
        run()
    except Exception as e:                       # noqa: BLE001
        import traceback
        traceback.print_exc()
        print("\n[FATAL] 脚本本身异常：%s" % e)
        return 1
    print("\nRESULT: ok=%d fail=%d" % (OK, FAIL))
    if NOTES:
        print("备注：")
        for n in NOTES:
            print("  - " + n)
    if FAILS:
        print("失败项：\n  - " + "\n  - ".join(FAILS))
    return 1 if FAIL else 0


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