dealerhub/backend/scripts/audit_api_permissions.py

"""API 权限矩阵探测（开发/审计工具）。

做什么：枚举全部 URL，用「匿名 / 无租户头 / 错误租户 / 有效 JWT」四种身份发 GET，
把状态码汇总成矩阵，标出**预期外开放**的端点（潜在越权）。

判定基线（本项目的既定设计）：
- 匿名能拿 200 的只应是"公开端点"白名单：ping / auth/token / demo/enter /
  billing/plans / open/statements / storefront 客户端入口
- 其余端点匿名应为 401/403（受全局 IsAuthenticated 保护）
- 带有效 JWT 但**无租户头**时，业务端点应 400（无法识别租户）而非 200

用法：
    python scripts/audit_api_permissions.py                 # 用 dev 库 + 自动登录 alice
    python scripts/audit_api_permissions.py --base http://127.0.0.1:9000
"""

from __future__ import annotations

import argparse
import json
import os
import re
import sys
import urllib.error
import urllib.request
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.dev")

import django  # noqa: E402

django.setup()

from django.urls import get_resolver  # noqa: E402


# 公开端点前缀（匿名访问 200/4xx 都正常，不算问题）
PUBLIC_PREFIXES = (
    "api/v1/ping/",
    "api/v1/auth/token/",
    "api/v1/demo/enter/",
    "api/v1/billing/plans/",
    "api/v1/open/statements/",
)

# 需要替换路径参数的占位（探测用 1）
PARAM_PATTERNS = [
    (re.compile(r"<int:[^>]+>"), "1"),
    (re.compile(r"<uuid:[^>]+>"), "00000000-0000-0000-0000-000000000000"),
    (re.compile(r"<[^>]+:[^>]+>"), "1"),
    (re.compile(r"\(\?P<[^>]+>\[\^/\.\]\+\)"), "1"),
    (re.compile(r"\\\.\(\?P<format>\[a-z0-9\]\+\)/\\?\$"), ""),
]


def collect_urls() -> list:
    """枚举所有 API URL（把正则路径参数换成占位值）。"""
    resolver = get_resolver()
    raw = set()

    def walk(patterns, prefix=""):
        for p in patterns:
            pat = prefix + str(p.pattern)
            if hasattr(p, "url_patterns"):
                walk(p.url_patterns, pat)
            else:
                raw.add(pat)

    walk(resolver.url_patterns)

    urls = set()
    for u in raw:
        if not u.startswith("api/"):
            continue
        if "<drf_format_suffix" in u:
            continue
        path = u
        for rx, repl in PARAM_PATTERNS:
            path = rx.sub(repl, path)
        if "(" in path or "?" in path:      # 仍含正则残留，跳过
            continue
        path = path.rstrip("$").rstrip("^")
        if not path.startswith("api/"):
            path = "api/" + path
        urls.add("/" + path.lstrip("/"))
    return sorted(urls)


def request(url: str, *, token: str = "", tenant: str = "") -> int:
    """发 GET，返回状态码（0 = 连接/其他异常）。"""
    req = urllib.request.Request(url, method="GET")
    if token:
        req.add_header("Authorization", f"Bearer {token}")
    if tenant:
        req.add_header("X-Tenant-Id", tenant)
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return resp.status
    except urllib.error.HTTPError as e:
        return e.code
    except Exception:
        return 0


def login(base: str, username: str, password: str) -> str:
    body = json.dumps({"username": username, "password": password}).encode()
    req = urllib.request.Request(f"{base}/api/v1/auth/token/", data=body, method="POST")
    req.add_header("Content-Type", "application/json")
    try:
        with urllib.request.urlopen(req, timeout=10) as resp:
            return json.loads(resp.read().decode())["access"]
    except Exception as exc:
        print(f"登录失败：{exc}")
        return ""


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--base", default="http://127.0.0.1:9000")
    ap.add_argument("--user", default="alice")
    ap.add_argument("--password", default="alice12345")
    ap.add_argument("--tenant", default="default")
    ap.add_argument("--verbose", action="store_true")
    args = ap.parse_args()

    base = args.base.rstrip("/")
    token = login(base, args.user, args.password)
    if not token:
        print("无法登录，探测中止")
        return 2

    urls = collect_urls()
    print(f"探测 {len(urls)} 个端点 · base={base} · 租户={args.tenant}\n")

    findings = []
    rows = []
    for path in urls:
        url = base + path
        anon = request(url)
        no_tenant = request(url, token=token)
        ok = request(url, token=token, tenant=args.tenant)

        is_public = any(path.startswith("/" + p) for p in PUBLIC_PREFIXES)
        rows.append((path, anon, no_tenant, ok, is_public))

        # 问题 1：非公开端点匿名可访问
        if not is_public and anon == 200:
            findings.append(("匿名可访问", path, f"anon={anon}"))
        # 问题 2：带 JWT 但无租户头仍返回 200（应为 400 无法识别租户）
        if not is_public and no_tenant == 200 and ok != 200:
            findings.append(("缺租户头仍 200", path, f"no_tenant={no_tenant}"))
        # 问题 3：合法请求反而 500（服务端缺陷）
        if ok == 500:
            findings.append(("合法请求 500", path, "with_tenant=500"))
        # 问题 4：合法请求 0（路由不可达）
        if ok == 0:
            findings.append(("端点不可达", path, "network/route error"))

    print(f"{'端点':<62} {'匿名':>5} {'无租户':>7} {'正常':>5}")
    print("-" * 84)
    for path, anon, nt, ok_, is_pub in rows:
        if args.verbose or anon == 200 or ok_ in (500, 0, 401):
            mark = " [公开]" if is_pub else ""
            print(f"{path:<62} {anon:>5} {nt:>7} {ok_:>5}{mark}")

    print()
    # 明细：带 JWT 仍被拒的端点（排查鉴权问题）
    auth_rejected = [(p, a, nt, o) for p, a, nt, o, pub in rows if o in (401, 403) and not pub]
    if auth_rejected:
        print("带 JWT 仍被拒的端点：")
        for p_, a_, nt_, o_ in auth_rejected:
            print(f"  {p_}  anon={a_} no_tenant={nt_} with_tenant={o_}")

    if findings:
        print(f"发现 {len(findings)} 项需要关注：")
        for kind, path, detail in findings:
            print(f"  [{kind}] {path}  ({detail})")
    else:
        print("权限矩阵无明显异常")

    # 汇总统计
    from collections import Counter

    n_pub = sum(1 for *_, p in rows if p)
    n_401 = sum(1 for _, a, _, _, p in rows if not p and a == 401)
    anon_dist = dict(sorted(Counter(a for _, a, _, _, _ in rows).items()))
    ok_dist = dict(sorted(Counter(o for _, _, _, o, _ in rows).items()))
    print(f"\n匿名状态码分布：{anon_dist}")
    print(f"正常请求状态码分布：{ok_dist}")
    print(f"统计：公开端点 {n_pub} · 匿名被拦 {n_401} · 其他 {len(rows) - n_pub - n_401}")
    return 1 if findings else 0


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