cma-management/backend/scripts/audit_api_auth_surface.py

#!/usr/bin/env python3
"""CMA API 鉴权面审计器（cma-api-auth-surface-audit-20260914）

用途
----
枚举 FastAPI app 的**全部** APIRoute，逐个判定其鉴权姿态并分类：

    A 类 — 合法 Bot 通道：以 X-BOT-KEY / bridge token 鉴权（verify_bot_key / verify_bridge_token）
    B 类 — 用户端点缺鉴权（缺口）：既无 X-BOT-KEY 也无 require_auth 依赖
    C 类 — 显式公开：在 PUBLIC_PATHS 白名单内（登录/注册/验证码/官网留资/健康探针等）

判定依据是**依赖图**（route.dependant 递归），不是源码文本 grep——router 级
dependencies=[Depends(require_auth_401)] 与端点级 Depends(require_auth) 都能识别。

用法（在 backend 目录下）
-------------------------
    ./venv/bin/python3 scripts/audit_api_auth_surface.py            # 人类可读表格
    ./venv/bin/python3 scripts/audit_api_auth_surface.py --json     # 机器可读

退出码：0 = 无 B 类缺口；1 = 存在 B 类缺口（可直接当 CI 闸门用）。
"""
import os
import sys
import json
import argparse

# 允许直接以脚本方式运行（把 backend 根加入 sys.path）
_HERE = os.path.dirname(os.path.abspath(__file__))
_BACKEND = os.path.dirname(_HERE)
if _BACKEND not in sys.path:
    sys.path.insert(0, _BACKEND)

# 测试/离线环境变量兜底（不连接真实 MySQL：仅枚举路由，不触发 get_db）
os.environ.setdefault("CMA_DB_USER", "audit")
os.environ.setdefault("CMA_DB_PASS", "audit")
os.environ.setdefault("CMA_DB_HOST", "localhost")
os.environ.setdefault("CMA_DB_PORT", "3306")
os.environ.setdefault("CMA_DB_NAME", "audit")

AUTH_STRICT = {"require_auth", "require_auth_401", "role_checker"}
BOT_AUTH = {"verify_bot_key", "verify_bridge_token"}

# C 类：显式登记的公开端点（(METHOD, path)）
PUBLIC_PATHS = {
    ("GET", "/health"),
    ("POST", "/api/cma/auth/login"),
    ("POST", "/api/cma/auth/register"),
    ("GET", "/api/cma/auth/login-entities"),
    ("GET", "/api/cma/bot/ping"),
    ("GET", "/api/cma/security/captcha/request"),
    ("GET", "/api/cma/security/captcha/request2"),
    ("POST", "/api/cma/security/captcha/verify"),
    ("POST", "/api/cma/lead"),
}


def dep_names(route) -> set:
    """递归收集 route 依赖图中所有可调用对象的名字"""
    acc = set()

    def walk(dep, depth=0):
        if depth > 12:
            return
        fn = getattr(dep.call, "__name__", None)
        acc.add(fn if fn else dep.call.__class__.__name__)
        for sub in dep.dependencies:
            walk(sub, depth + 1)

    for dep in route.dependant.dependencies:
        walk(dep)
    return acc


def collect_routes(app) -> list:
    """返回全部 APIRoute 的鉴权姿态行"""
    from fastapi.routing import APIRoute

    rows = []
    for r in app.routes:
        if not isinstance(r, APIRoute):
            continue
        methods = sorted(m for m in r.methods if m not in ("HEAD", "OPTIONS"))
        deps = dep_names(r)
        auth = sorted(deps & AUTH_STRICT)
        bot = sorted(deps & BOT_AUTH)
        public = bool(methods) and all((m, r.path) in PUBLIC_PATHS for m in methods)
        if auth:
            cls = "AUTHED"
        elif bot:
            cls = "A"          # Bot 通道
        elif public:
            cls = "C"          # 显式公开
        else:
            cls = "B"          # 缺口
        rows.append({
            "path": r.path,
            "methods": ",".join(methods),
            "handler": r.endpoint.__name__,
            "module": r.endpoint.__module__,
            "auth": auth,
            "bot_auth": bot,
            "class": cls,
        })
    rows.sort(key=lambda x: (x["class"], x["path"], x["methods"]))
    return rows


def build_app():
    """构建 app 实例（不连接真实 DB：仅导入模块）"""
    # 先让 database 模块指向内存 SQLite，避免任何意外真实连接
    from sqlalchemy import create_engine
    from sqlalchemy.orm import sessionmaker
    from sqlalchemy.pool import StaticPool
    from app import database as db_module

    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False},
                           poolclass=StaticPool)
    db_module.engine = engine
    db_module.SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

    from app.main import app
    return app


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument("--json", action="store_true", help="输出 JSON")
    args = ap.parse_args()

    rows = collect_routes(build_app())
    b = [r for r in rows if r["class"] == "B"]
    a = [r for r in rows if r["class"] == "A"]
    c = [r for r in rows if r["class"] == "C"]
    authed = [r for r in rows if r["class"] == "AUTHED"]

    if args.json:
        print(json.dumps({"total": len(rows), "authed": len(authed), "A": len(a),
                          "B": len(b), "C": len(c), "rows": rows}, ensure_ascii=False, indent=1))
    else:
        print(f"CMA API 鉴权面审计 — 端点总数 {len(rows)}")
        print(f"  AUTHED(require_auth/role) = {len(authed)}")
        print(f"  A 类(合法 Bot 通道 X-BOT-KEY) = {len(a)}")
        print(f"  C 类(显式公开)               = {len(c)}")
        print(f"  B 类(缺鉴权缺口)             = {len(b)}")
        if b:
            print("\n=== B 类缺口（需补鉴权）===")
            for r in b:
                print(f'  {r["methods"]:16s} {r["path"]:52s} {r["module"]}')
        if c:
            print("\n=== C 类（显式公开）===")
            for r in c:
                print(f'  {r["methods"]:16s} {r["path"]:52s} {r["module"]}')
    return 1 if b else 0


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