diff --git a/app/api/audit_middleware.py b/app/api/audit_middleware.py new file mode 100644 index 0000000..79336a8 --- /dev/null +++ b/app/api/audit_middleware.py @@ -0,0 +1,80 @@ +"""平台访问审计中间件（T-02 · F-02 全量留痕 / 手册 P-05）。 + +每个非排障路径请求落一行 audit_log（event_type='http_access'，agent_type= +'platform'——与 simulate 网关审计口径一致）：method/path/status/latency_ms/ +request_id 进 input_summary，actor 取 deps 鉴权成功后注入的 request.state.auth +（401 时为 anonymous）。写库失败降级 warning 不阻塞响应（业务层关键判定审计 +[authz/trade_event/suitability_check 等] 不经此路径，不受降级影响）。 + +注册顺序：trace 中间件之后注册（即执行序在 trace 之内），保证审计时 +trace_id/request_id 已绑定。 +""" + +from __future__ import annotations + +import logging +import time + +from fastapi import Request +from starlette.responses import Response + +from app.repository.risk_repository import RiskRepository +from app.utils.trace import current_request_id, current_trace + +logger = logging.getLogger(__name__) + +# 排障/文档路径不落访问审计（健康探活噪音；docs 页面无业务语义） +_SKIP_PATHS = {"/health", "/openapi.json", "/docs", "/redoc", "/favicon.ico"} + + +def _repo() -> RiskRepository: + """审计仓储入口（测试 monkeypatch 点，与 api 路由模式一致）。""" + return RiskRepository() + + +def write_http_access(request: Request, status: int, latency_ms: int) -> None: + """单请求访问审计（INSERT-only；audit_middleware 调用，测试可直调）。""" + auth = getattr(request.state, "auth", None) + _repo().insert_audit_log( + { + "trace_id": current_trace(), + "event_type": "http_access", + "agent_type": "platform", + "actor_id": auth.actor_id if auth is not None else "anonymous", + "customer_id": None, + "rule_id": None, + "input_summary": { + "method": request.method, + "path": request.url.path, + "status": status, + "latency_ms": latency_ms, + "request_id": current_request_id(), + }, + "decision": str(status), + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + + +async def audit_middleware(request: Request, call_next) -> Response: + """http_access 中间件：异常请求留痕 500 后继续抛（trace 层兜底出错误体）。""" + if request.url.path in _SKIP_PATHS or request.method == "OPTIONS": + return await call_next(request) + started = time.perf_counter() + try: + response = await call_next(request) + status = response.status_code + except Exception: + try: + write_http_access(request, 500, int((time.perf_counter() - started) * 1000)) + except Exception: + logger.warning("http_access audit failed on 500 path", exc_info=True) + raise + try: + write_http_access(request, status, int((time.perf_counter() - started) * 1000)) + except Exception: + logger.warning("http_access audit failed (degrade, not blocking)", exc_info=True) + return response diff --git a/app/api/deps.py b/app/api/deps.py index 12affcc..ce88f6e 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -86,12 +86,15 @@ def _authz_audit( ) -> None: """鉴权失败审计（event_type='authz'；手册 P-05，B6 评审 P1-1）。 - agent_type 按路由归属传入（网关路由传 'platform'，复审 P3：与放行审计 - 同口径，避免按模块检索审计时漏网关越权事件）。 + T-02 双写（挂账⑧收口）：audit_log + input_guard_log（P-05 要求双留痕； + guard_type 用 ENUM 四值内的 illegal_param 承载鉴权拒绝类）。agent_type= + 'platform'（simulate 网关路由）时跳过 input_guard_log——该表 ENUM 仅 + 四 Agent 入口，网关越权仅 audit_log 留痕（复审 P3 同口径）。 """ + trace_id = current_trace() or new_trace() risk_repo.insert_audit_log( { - "trace_id": current_trace() or new_trace(), + "trace_id": trace_id, "event_type": "authz", "agent_type": agent_type, "actor_id": auth.actor_id if auth else "anonymous", @@ -105,6 +108,15 @@ def _authz_audit( "handler_comment": None, } ) + if agent_type in AGENT_TYPES: + risk_repo.insert_input_guard_log( + trace_id=trace_id, + agent_type=agent_type, + actor_id=auth.actor_id if auth else "anonymous", + guard_type="illegal_param", + action="blocked", + raw_excerpt=code, + ) def deny( @@ -126,10 +138,14 @@ def _unauthenticated_audit( code: str, agent_type: str = "risk", ) -> None: - """401 留痕（P-05；B6 debug 通道既有口径，T-01 推广到 JWT 通道）。""" + """401 留痕（P-05；B6 debug 通道既有口径，T-01 推广到 JWT 通道）。 + + T-02 双写 input_guard_log（同 _authz_audit 口径；agent_type 限四 Agent）。 + """ + trace_id = current_trace() or new_trace() risk_repo.insert_audit_log( { - "trace_id": current_trace() or new_trace(), + "trace_id": trace_id, "event_type": "authz", "agent_type": agent_type, "actor_id": actor_id or "anonymous", @@ -143,6 +159,15 @@ def _unauthenticated_audit( "handler_comment": None, } ) + if agent_type in AGENT_TYPES: + risk_repo.insert_input_guard_log( + trace_id=trace_id, + agent_type=agent_type, + actor_id=actor_id or "anonymous", + guard_type="illegal_param", + action="blocked", + raw_excerpt=code, + ) def _claims_to_auth(claims: Claims) -> AuthContext: @@ -158,6 +183,12 @@ def _claims_to_auth(claims: Claims) -> AuthContext: ) +def _bind_state(request: Request, auth: AuthContext) -> AuthContext: + """auth 注入 request.state（T-02 访问审计取 actor；JWT/debug 通道统一出口）。""" + request.state.auth = auth + return auth + + def assert_agent_access( auth: AuthContext, agent_type: str, @@ -205,8 +236,7 @@ def get_auth_context(request: Request) -> AuthContext: raise ApiError(400, "BAD_REQUEST", f"invalid {AGENT_TYPE_HEADER}: {agent_type}") # 交叉校验失败也走 deny 全量审计（fail-closed；agent_type 归请求目标） assert_agent_access(auth, agent_type, risk_repo=RiskRepository()) - request.state.auth = auth - return auth + return _bind_state(request, auth) if settings.app_env != "development": _unauthenticated_audit(RiskRepository(), "anonymous", "AUTH_401_MISSING_BEARER") @@ -223,11 +253,14 @@ def get_auth_context(request: Request) -> AuthContext: raise ApiError( 401, "AUTH_401_MISSING_DEBUG_HEADERS", "missing X-Debug-Role/X-Debug-Actor headers" ) - return AuthContext( - actor_id=actor_id, - roles=roles, - customer_id=actor_id if "customer" in roles else None, - token_type="customer" if "customer" in roles else "staff", + return _bind_state( + request, + AuthContext( + actor_id=actor_id, + roles=roles, + customer_id=actor_id if "customer" in roles else None, + token_type="customer" if "customer" in roles else "staff", + ), ) diff --git a/app/main.py b/app/main.py index 25b3233..ccb424c 100644 --- a/app/main.py +++ b/app/main.py @@ -13,20 +13,31 @@ audit_middleware；X-Request-Id 独立生成（B7 复审 P3-4）。 from __future__ import annotations +import logging import re from contextlib import asynccontextmanager from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse from app.api import deps +from app.api.audit_middleware import audit_middleware from app.api.risk import router as risk_router from app.api.simulate import router as simulate_router from app.config.settings import settings from app.service.auth_service import jwt_ready from app.service.risk import redis_gateway from app.utils.db import dispose_engines -from app.utils.response import register_error_handlers -from app.utils.trace import new_trace, reset_trace, set_trace +from app.utils.response import error_body, register_error_handlers +from app.utils.trace import ( + bind_request_id, + new_trace, + reset_request_id, + reset_trace, + set_trace, +) + +logger = logging.getLogger(__name__) # 透传外部 X-Trace-Id 的格式白名单（防响应头注入；不合规一律新生成） _TRACE_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$") @@ -57,17 +68,37 @@ app.include_router(risk_router) app.include_router(simulate_router) +@app.middleware("http") +async def audit_middleware_entry(request: Request, call_next): + """T-02 访问审计：先注册（执行序在 trace 之内，trace_id/request_id 已绑定）。""" + return await audit_middleware(request, call_next) + + @app.middleware("http") async def trace_middleware(request: Request, call_next): - """trace_id 贯通：透传合法 X-Trace-Id，否则生成；响应头回写。""" + """trace_id/request_id 贯通 + 未捕获异常兜底（B7 复审 P2-2）。 + + 异常发生在本中间件之内时，Starlette 的 ServerErrorMiddleware（栈外层） + 生成的 500 响应不经过用户中间件——trace 头丢失的根因；此处 catch 后 + 直接产出统一错误体，保证 500 也带 X-Trace-Id/X-Request-Id。 + """ incoming = request.headers.get("X-Trace-Id", "") trace_id = incoming if _TRACE_ID_PATTERN.fullmatch(incoming) else new_trace() token = set_trace(trace_id) + request_id, rid_token = bind_request_id(request.headers.get("X-Request-Id", "")) try: - response = await call_next(request) + try: + response = await call_next(request) + except Exception: + logger.exception("unhandled error on %s %s", request.method, request.url.path) + response = JSONResponse( + status_code=500, content=error_body("INTERNAL_ERROR", "internal server error") + ) finally: reset_trace(token) + reset_request_id(rid_token) response.headers["X-Trace-Id"] = trace_id + response.headers["X-Request-Id"] = request_id return response diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py index 33f9636..7a96816 100644 --- a/app/repository/risk_repository.py +++ b/app/repository/risk_repository.py @@ -320,6 +320,40 @@ class RiskRepository: with self._engine.begin() as conn: conn.execute(_AUDIT_SQL, self._dump_audit(entry)) + # ---------- input_guard_log（输入安全防护 · 只 INSERT，手册 P-05 双写）---------- + + def insert_input_guard_log( + self, + trace_id: str, + agent_type: str, + actor_id: str, + guard_type: str, + action: str, + raw_excerpt: str | None = None, + session_id: str | None = None, + ) -> None: + """安全防护留痕（T-02 挂账⑧收口）；ENUM 口径见 01-mysql-共用底座.sql。""" + sql = text( + """ + INSERT INTO input_guard_log + (trace_id, session_id, agent_type, actor_id, guard_type, raw_excerpt, action) + VALUES (:trace_id, :session_id, :agent_type, :actor_id, :guard_type, :raw_excerpt, :action) + """ + ) + with self._engine.begin() as conn: + conn.execute( + sql, + { + "trace_id": trace_id, + "session_id": session_id, + "agent_type": agent_type, + "actor_id": actor_id, + "guard_type": guard_type, + "raw_excerpt": (raw_excerpt or "")[:1024], + "action": action, + }, + ) + @staticmethod def _dump_audit(entry: dict[str, Any]) -> dict[str, Any]: params = dict(entry) diff --git a/app/utils/response.py b/app/utils/response.py index 064b12c..c0449cd 100644 --- a/app/utils/response.py +++ b/app/utils/response.py @@ -1,24 +1,35 @@ -"""统一 API 响应外壳（B7 · 挂账④）。 +"""统一 API 响应外壳（B7 · 挂账④；T-02 扩展）。 成功响应：业务字段平铺不变（B5 评审 P2-2 口径「路由返回体不变」），trace_id -经 X-Trace-Id 响应头贯通（main 中间件）；错误响应统一 JWT 手册 §10 结构 -{error_code, message, trace_id, request_id}（request_id 沿用 trace_id， -独立请求级标识尚未引入）。main 与测试 app 共用 register_error_handlers。 +经 X-Trace-Id 响应头贯通；错误响应统一 JWT 手册 §10 结构 +{error_code, message, trace_id, request_id}——request_id 为独立请求级标识 +（T-02，B7 复审 P3-4 收口），trace_id/request_id 分别对齐响应头。 + +T-02 补齐（B7 复审 P2-2）：422 请求校验失败、404/405 路由方法错误改用统一 +错误体（原 FastAPI detail 结构）；未捕获异常的 500 由 main.trace 中间件 +兜底（异常穿透 exception handler，直接在中间件层生成错误体并回写 trace 头）。 +main 与测试 app 共用 register_error_handlers。 """ from __future__ import annotations from fastapi import FastAPI, Request +from fastapi.exceptions import RequestValidationError from fastapi.responses import JSONResponse +from starlette.exceptions import HTTPException as StarletteHTTPException from app.utils.exceptions import ApiError, PermissionDenied -from app.utils.trace import current_trace, new_trace +from app.utils.trace import current_request_id, current_trace, new_trace, new_request_id + +# 路由级 HTTP 状态 → 错误码（手册 §10 之外的平台通用码） +_HTTP_ERROR_CODES = {404: "NOT_FOUND", 405: "METHOD_NOT_ALLOWED"} def error_body(error_code: str, message: str) -> dict[str, str]: - """手册 §10 错误体（trace 缺失时兜底生成，保证响应可归因）。""" + """手册 §10 错误体（trace/request 缺失时兜底生成，保证响应可归因）。""" tid = current_trace() or new_trace() - return {"error_code": error_code, "message": message, "trace_id": tid, "request_id": tid} + rid = current_request_id() or new_request_id() + return {"error_code": error_code, "message": message, "trace_id": tid, "request_id": rid} def register_error_handlers(app: FastAPI) -> None: @@ -31,3 +42,15 @@ def register_error_handlers(app: FastAPI) -> None: @app.exception_handler(PermissionDenied) async def _permission_denied_handler(request: Request, exc: PermissionDenied) -> JSONResponse: return JSONResponse(status_code=403, content=error_body(exc.code, str(exc))) + + @app.exception_handler(StarletteHTTPException) + async def _http_exception_handler(request: Request, exc: StarletteHTTPException) -> JSONResponse: + code = _HTTP_ERROR_CODES.get(exc.status_code, "HTTP_ERROR") + return JSONResponse(status_code=exc.status_code, content=error_body(code, str(exc.detail))) + + @app.exception_handler(RequestValidationError) + async def _validation_error_handler(request: Request, exc: RequestValidationError) -> JSONResponse: + return JSONResponse( + status_code=422, + content=error_body("REQUEST_VALIDATION_FAILED", "request validation failed"), + ) diff --git a/app/utils/trace.py b/app/utils/trace.py index ace7460..d8f9f99 100644 --- a/app/utils/trace.py +++ b/app/utils/trace.py @@ -8,10 +8,17 @@ from __future__ import annotations +import re from contextvars import ContextVar, Token from uuid import uuid4 +# 透传外部 X-Trace-Id / X-Request-Id 的格式白名单（防响应头注入；不合规一律新生成） +_HEADER_ID_PATTERN = re.compile(r"^[A-Za-z0-9._-]{1,64}$") + _trace_id: ContextVar[str] = ContextVar("trace_id", default="") +# 独立请求级 ID（T-02 · B7 复审 P3-4）：trace_id 贯通链路，request_id 标识单次 +# HTTP 请求（幂等/重试对账，手册 §4.6），两者不再互用。 +_request_id: ContextVar[str] = ContextVar("request_id", default="") def new_trace(trace_id: str | None = None) -> str: @@ -40,3 +47,25 @@ def ensure_trace() -> None: """无上下文时兜底归因（脚本/引擎入口），有值时保留（中间件场景不重新 set）。""" if not current_trace(): new_trace() + + +def new_request_id() -> str: + """生成并绑定 request_id（中间件入口每请求一次）。""" + rid = f"req-{uuid4().hex[:16]}" + _request_id.set(rid) + return rid + + +def bind_request_id(request_id: str | None) -> tuple[str, Token]: + """绑定请求 request_id（合法透传否则生成）；返回 (id, token) 供中间件复位。""" + rid = request_id if _HEADER_ID_PATTERN.fullmatch(request_id or "") else f"req-{uuid4().hex[:16]}" + return rid, _request_id.set(rid) + + +def reset_request_id(token: Token) -> None: + _request_id.reset(token) + + +def current_request_id() -> str: + """读取当前请求 request_id；未初始化返回空串（错误体兜底生成）。""" + return _request_id.get() diff --git a/tests/_ddl.py b/tests/_ddl.py index b59adaa..2653b31 100644 --- a/tests/_ddl.py +++ b/tests/_ddl.py @@ -47,6 +47,13 @@ SQLITE_TABLES: dict[str, str] = { handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), created_at {_TS}) """, + "input_guard_log": f""" + CREATE TABLE input_guard_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id VARCHAR(64), session_id VARCHAR(64), + agent_type VARCHAR(16), actor_id VARCHAR(64), guard_type VARCHAR(24), + raw_excerpt VARCHAR(1024), action VARCHAR(16), + created_at {_TS}) + """, "risk_alert": f""" CREATE TABLE risk_alert ( alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64), diff --git a/tests/test_audit_middleware.py b/tests/test_audit_middleware.py new file mode 100644 index 0000000..3140ed3 --- /dev/null +++ b/tests/test_audit_middleware.py @@ -0,0 +1,182 @@ +"""T-02 审计中间件与统一错误体（http_access / request_id / 4xx-500 / guard 双写）。 + +main app 真中间件栈（audit 在 trace 内层）走 TestClient；仓储注入 sqlite。 +500 路径用 monkeypatch 令依赖抛 RuntimeError 验证 trace 层兜底错误体与 +访问审计留痕（B7 复审 P2-2 收口）。 +""" + +from __future__ import annotations + +import re + +import pytest +from fastapi.testclient import TestClient +from sqlalchemy import text + +from _ddl import create_sqlite_engine + +from app.api import deps as deps_mod +from app.api import risk as risk_api +from app.api import simulate as simulate_mod +from app.api import audit_middleware as audit_mod +from app.main import app +from app.repository.risk_repository import RiskRepository +from app.service.risk import redis_gateway + + +class FakeGateway: + def publish(self, channel, payload): + pass + + def delete(self, *keys): + pass + + def exists(self, key): + return False + + def set_ex(self, key, value, ttl): + pass + + +@pytest.fixture() +def env(monkeypatch): + engine = create_sqlite_engine() + repo = RiskRepository(engine=engine) + monkeypatch.setattr(risk_api, "_repo", lambda: repo) + monkeypatch.setattr(simulate_mod, "_repo", lambda: repo) + monkeypatch.setattr(audit_mod, "_repo", lambda: repo) + monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) + with TestClient(app) as c: + monkeypatch.setattr(redis_gateway, "_gateway", FakeGateway()) + yield {"client": c, "repo": repo, "engine": engine} + engine.dispose() + + +def _rows(engine, sql: str, **params) -> list[dict]: + with engine.connect() as conn: + return [dict(r) for r in conn.execute(text(sql), params).mappings().all()] + + +def _http_access(engine) -> list[dict]: + return _rows( + engine, + "SELECT actor_id, decision, input_summary FROM audit_log WHERE event_type = 'http_access'", + ) + + +def test_http_access_written_with_actor(env): + r = env["client"].get( + "/api/risk/alerts", headers={"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001"} + ) + assert r.status_code == 200 + rows = _http_access(env["engine"]) + assert len(rows) == 1 + assert rows[0]["actor_id"] == "STAFF-30001" + summary = rows[0]["input_summary"] + assert '"status": 200' in summary and "/api/risk/alerts" in summary + + +def test_http_access_skips_health(env): + env["client"].get("/health") + assert _http_access(env["engine"]) == [] + + +def test_http_access_401_anonymous(env): + r = env["client"].get("/api/risk/alerts") + assert r.status_code == 401 + rows = _http_access(env["engine"]) + assert rows and rows[0]["actor_id"] == "anonymous" and rows[0]["decision"] == "401" + + +def test_http_access_written_on_500(env, monkeypatch): + """未捕获异常：http_access 500 留痕 + 统一错误体 + trace 头回写（P2-2）。""" + def _boom(): + raise RuntimeError("boom") + + monkeypatch.setattr(risk_api, "_repo", _boom) + r = env["client"].get( + "/api/risk/alerts", headers={"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001"} + ) + assert r.status_code == 500 + body = r.json() + assert body["error_code"] == "INTERNAL_ERROR" + assert body["trace_id"] == r.headers["X-Trace-Id"] + assert r.headers["X-Request-Id"] + rows = _http_access(env["engine"]) + assert rows and rows[0]["decision"] == "500" + + +# ---------- 统一错误体：422/404/405（B7 复审 P2-2 补齐） ---------- + + +def test_error_body_422_validation(env): + """鉴权依赖先于 body 校验：带 debug 头 + 缺 product_id → 422 统一错误体。""" + r = env["client"].post( + "/api/risk/suitability/check", + json={"customer_id": "CUST-9527"}, + headers={"X-Debug-Role": "risk_officer", "X-Debug-Actor": "STAFF-30001"}, + ) + assert r.status_code == 422 + body = r.json() + assert body["error_code"] == "REQUEST_VALIDATION_FAILED" + assert set(body) == {"error_code", "message", "trace_id", "request_id"} + + +def test_error_body_404_route(env): + r = env["client"].get("/api/definitely-not-here") + assert r.status_code == 404 and r.json()["error_code"] == "NOT_FOUND" + + +def test_error_body_405_method(env): + r = env["client"].delete("/api/risk/alerts") + assert r.status_code == 405 and r.json()["error_code"] == "METHOD_NOT_ALLOWED" + + +# ---------- 独立 request_id（B7 复审 P3-4） ---------- + + +def test_request_id_independent_from_trace(env): + r = env["client"].get("/api/risk/alerts") + tid, rid = r.headers["X-Trace-Id"], r.headers["X-Request-Id"] + assert tid.startswith("trc-") and rid.startswith("req-") and tid != rid + assert r.json()["trace_id"] == tid # 401 错误体四键对齐各自头 + + +def test_request_id_passthrough(env): + rid = "req-abc123def45678" + r = env["client"].get("/api/risk/alerts", headers={"X-Request-Id": rid}) + assert r.headers["X-Request-Id"] == rid + + +def test_request_id_invalid_regenerated(env): + r = env["client"].get("/api/risk/alerts", headers={"X-Request-Id": "bad id!"}) + assert r.headers["X-Request-Id"] != "bad id!" and r.headers["X-Request-Id"].startswith("req-") + + +# ---------- input_guard_log 双写（挂账⑧） ---------- + + +def test_guard_log_written_on_401_and_403(env): + env["client"].get("/api/risk/alerts") # 401 + env["client"].post( + "/api/risk/alerts/A-1/handle", + json={"handler_result": "confirmed_normal"}, + headers={"X-Debug-Role": "compliance", "X-Debug-Actor": "STAFF-40001"}, # 403 + ) + rows = _rows(env["engine"], "SELECT agent_type, actor_id, guard_type, action FROM input_guard_log") + assert len(rows) == 2 + assert all(r["guard_type"] == "illegal_param" and r["action"] == "blocked" for r in rows) + + +def test_guard_log_skipped_for_platform_agent(env): + """simulate 网关（agent_type=platform）越权仅 audit_log，不写 guard（ENUM 口径）。""" + env["client"].post( + "/api/simulate/trade", + json={"customer_id": "CUST-9527", "product_id": "PROD-110022", "trade_type": "subscribe", "amount": 100}, + headers={"X-Debug-Role": "analyst", "X-Debug-Actor": "STAFF-20001"}, + ) + assert _rows(env["engine"], "SELECT 1 FROM input_guard_log") == [] + forbidden = _rows( + env["engine"], "SELECT agent_type FROM audit_log WHERE decision = 'forbidden'" + ) + assert forbidden and forbidden[0]["agent_type"] == "platform" diff --git a/tests/test_main.py b/tests/test_main.py index 21266a3..76b8698 100644 --- a/tests/test_main.py +++ b/tests/test_main.py @@ -84,14 +84,16 @@ def test_trace_header_invalid_regenerated(client): def test_unified_error_body_401_with_trace(client): - """手册 §10 错误体四键 + trace_id 与响应头一致（中间件贯通初验）。""" + """手册 §10 错误体四键 + trace_id/request_id 分别对齐响应头（T-02 独立双 ID）。""" r = client.get("/api/risk/alerts") # 无 debug 头 assert r.status_code == 401 body = r.json() assert body["error_code"] == "AUTH_401_MISSING_DEBUG_HEADERS" assert body["message"] assert set(body) == {"error_code", "message", "trace_id", "request_id"} - assert body["trace_id"] == body["request_id"] == r.headers["X-Trace-Id"] + assert body["trace_id"] == r.headers["X-Trace-Id"] + assert body["request_id"] == r.headers["X-Request-Id"] + assert body["trace_id"] != body["request_id"] def test_lifespan_rejects_debug_factory_in_non_dev(monkeypatch):