refactor(audit): fold audit logging into access_log, drop AuditMiddleware

按用户复查意见把审计折进 main.py 已有的 access_log 中间件，少一个 middleware、让诊断与合规共用一个出口。 * 删 backend.audit.AuditMiddleware（user_id 解析搬到 main.py 的 _audit_user_id 模块级 helper，cookie/Bearer 头 + JWT 验签，失败/缺失一律 '-'，全异常捕获不让审计拖死业务）。 * access_log 在 success 路径补 logger.bind(user_id, method, path, status).info('audit')；exception 路径同样补一条 status=500 的审计行，然后 re-raise 让全局 handler 转 500。 * 删 app.add_middleware(AuditMiddleware)。 * audit.py 只剩 _DailyFileSink / configure_audit_logging / AUDIT_LOG_FORMAT，文件 sink 与 retention 清理逻辑不变。 * 测试改用 _AccessLogReplica 复制 access_log 审计契约（不 import main.py 避免触发 lifespan 里的 MySQL/engine 初始化），删 3 个 middleware 单独 case，加 access_log 端到端 success / 5xx / 无 JWT 三个 case。
2026-09-02 10:10:41 +08:00
parent ff33523867
commit 8dc2311536
3 changed files with 142 additions and 117 deletions
@@ -1,4 +1,4 @@
-"""审计日志中间件：每个 HTTP 请求写一条合规记录到独立的按天滚动文件。
+"""按天单文件的 audit log sink，供 main.py 的 access_log 中间件复用。

 设计要点
 --------
@@ -12,32 +12,21 @@
  fd 再打开新文件。不使用 loguru 自带的 ``rotation="00:00"``，因为它对
  string path 产出的文件名是 ``audit.log.YYYY-MM-DD_HH-MM-SS``，既没有
  ``audit-`` 前缀也不符合每天一个文件的要求。
-* 与 ``main.py`` L108 的 ``access_log`` 是两回事，刻意分离：
-  ``access_log`` 是诊断日志（method / path / status / 耗时），走 stderr；
-  本中间件是合规日志（时间 / 用户 / 接口 / 状态码），写独立文件。两者并存。
-
-认证解析
--------
-中间件在路由解析之前执行，拿不到 ``Depends(request_context)`` 注入的结果，
-也绝不为此做 DB 查询。用户身份只通过本进程内 CPU 验签解 JWT 得到：
-优先 ``access_token`` cookie，其次 ``Authorization: Bearer`` 头；验签
-失败或缺失一律记 ``-``。审计写入自身失败也不得把请求拖死（全部捕获）。
+* 谁写审计行：main.py 的 ``access_log`` 中间件在 success 与 exception
+  两条路径各打一条 ``logger.bind(user_id, method, path, status).info("audit")``。
+  本模块只管把这类行路由到按天文件 sink；user_id 的解析（cookie /
+  Bearer 头 + JWT 验签）在 main.py 内部完成，审计只记录、不查 DB。
 """

 from __future__ import annotations

 import os
 import time
-from collections.abc import Callable
 from datetime import UTC, datetime
 from pathlib import Path
 from typing import TextIO

-from common.auth.jwt import verify_jwt_token
 from loguru import logger
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.requests import Request
-from starlette.responses import Response

 # 纯文本一行一条（末尾换行由 loguru 的 terminator 追加）：
 #   2026-08-21 14:30:00.123 | 01USER... | GET /api/v1/scripts/01ABC... -> 200
@@ -89,7 +78,7 @@ class _DailyFileSink:


 def _audit_filter(record: dict) -> bool:
-    """只放行中间件自己打的审计行，其它 INFO 日志不进审计文件。"""
+    """只放行 access_log 打的审计行，其它 INFO 日志不进审计文件。"""
    extra = record["extra"]
    return (
        record["message"] == "audit"
@@ -138,58 +127,7 @@ def configure_audit_logging(log_dir: str, retention_days: int) -> None:
    _CONFIGURED = True


-class AuditMiddleware(BaseHTTPMiddleware):
-    """对每个 HTTP 请求写一行审计记录。
-
-    与 ``access_log``（main.py L108）的关系：``access_log`` 是诊断日志
-    （方法/路径/状态码/耗时，走 stderr），本中间件是合规日志（时间/用户/
-    接口），写独立文件。两者并存，互不合并。
-
-    关键约束：中间件不做 DB 查询、不碰 ``Depends(request_context)``、
-    不修改 response body；用户身份只靠本进程 CPU 验签 JWT 解析。
-    """
-
-    async def dispatch(self, request: Request, call_next: Callable) -> Response:
-        try:
-            response = await call_next(request)
-        except Exception:
-            # call_next 抛错（例如 websocket upgrade 或未处理的路由异常）：
-            # 仍然写一行 status=0 的审计，并把原始异常继续往上抛，交给全局
-            # unhandled_exception_handler 返回 500 —— 审计逻辑本身不吞错、
-            # 也不改写响应。
-            self._record(request, 0)
-            raise
-        self._record(request, response.status_code)
-        return response
-
-    @staticmethod
-    def _record(request: Request, status: int) -> None:
-        token = request.cookies.get("access_token") or ""
-        if not token:
-            token = (
-                request.headers.get("authorization", "")
-                .removeprefix("Bearer ")
-                .strip()
-            )
-        user_id = "-"
-        if token:
-            try:
-                user_id = verify_jwt_token(token)["sub"]
-            except Exception:  # noqa: BLE001 - 验签/解析失败一律记 "-"，审计不能因坏 JWT 抛错
-                user_id = "-"
-        try:
-            logger.bind(
-                user_id=user_id,
-                method=request.method,
-                path=request.url.path,
-                status=status,
-            ).info("audit")
-        except Exception:  # noqa: BLE001, S110 - 审计写入失败静默忽略，不能把请求拖死
-            pass
-
-
 __all__ = [
    "AUDIT_LOG_FORMAT",
-    "AuditMiddleware",
    "configure_audit_logging",
 ]
@@ -17,6 +17,7 @@ from contextlib import asynccontextmanager
 from typing import Any

 import httpx
+from common.auth.jwt import JwtError, verify_jwt_token
 from common.config import settings
 from common.db import create_database_engine, create_session_factory
 from common.logging import configure_logging
@@ -32,7 +33,7 @@ from fastapi import Request
 from fastapi.responses import JSONResponse
 from loguru import logger

-from backend.audit import AuditMiddleware, configure_audit_logging
+from backend.audit import configure_audit_logging
 from backend.admin import router as admin_router
 from backend.auth import router as auth_router
 from backend.jupyter import router as jupyter_router
@@ -109,15 +110,33 @@ app.include_router(platform_router)
 # 内部存储接口额外加上 /internal 前缀，供后端服务间调用，不作为普通前端 API。
 app.include_router(storage_api_router, prefix="/internal")

-# 审计中间件必须注册在所有路由之后：这样早期 include_router 注册的路由也
-# 会被审计覆盖；access_log 在它外面，负责诊断日志，两者并存。
-app.add_middleware(AuditMiddleware)
+
+def _audit_user_id(request: Request) -> str:
+    """从 cookie / Bearer 头解 JWT 拿 user_id；失败/缺失一律 '-'。
+
+    故意不做 DB 查（RequestContext 在路由解析后才注入；审计不该为
+    每请求打 MySQL）。捕获所有异常，让审计失败不拖死业务请求。
+    """
+    token = request.cookies.get("access_token")
+    if not token:
+        auth = request.headers.get("authorization", "")
+        if auth.lower().startswith("bearer "):
+            token = auth[7:].strip()
+    if not token:
+        return "-"
+    try:
+        payload = verify_jwt_token(token)
+    except (JwtError, Exception):  # 任何异常都吞
+        return "-"
+    sub = payload.get("sub")
+    return sub or "-"


@app.middleware("http")
 async def access_log(request: Request, call_next):
-    # 每个 HTTP 请求都记录方法、路径、状态码和耗时；排查页面请求失败时，
-    # Docker Desktop 中 backend 容器的 Logs 就会显示这里生成的日志。
+    # 诊断：方法/路径/状态码/耗时 走 stderr（loguru default sink）
+    # 合规：时间/用户/方法/路径/状态码 走独立 audit 文件 sink
+    # 两条 logger.info() 共用一个出口，便于排查
    start = time.perf_counter()
    try:
        response = await call_next(request)
@@ -127,6 +146,13 @@ async def access_log(request: Request, call_next):
            "request failed {method} {path} after {ms:.1f}ms",
            method=request.method, path=request.url.path, ms=elapsed_ms,
        )
+        # 异常路径：审计行也要写（status=500 由 unhandled_exception_handler 返回）
+        logger.bind(
+            user_id=_audit_user_id(request),
+            method=request.method,
+            path=request.url.path,
+            status=500,
+        ).info("audit")
        raise
    elapsed_ms = (time.perf_counter() - start) * 1000
    logger.info(
@@ -134,6 +160,12 @@ async def access_log(request: Request, call_next):
        method=request.method, path=request.url.path,
        status=response.status_code, ms=elapsed_ms,
    )
+    logger.bind(
+        user_id=_audit_user_id(request),
+        method=request.method,
+        path=request.url.path,
+        status=response.status_code,
+    ).info("audit")
    return response


@@ -1,32 +1,84 @@
-"""审计日志中间件测试。
+"""审计日志测试。

 覆盖：
 * 每天一个 ``audit-YYYY-MM-DD.log`` 文件且写入至少一行；
 * 日志行包含 user_id / method / path / status；
-* 未登录（无 cookie 无 header）与坏 JWT 时 user_id 记 ``-``；
-* 带路径参数的请求原样记录实际 path（不替换为 ``{script_id}``）；
+* access_log 在 success 与 exception（500）两条路径都写审计行；
+* 未登录（无 cookie 无 header）时 user_id 记 ``-``；
+* Authorization: Bearer 头能解析出 user_id；
 * 启动时按 mtime 清理超过保留天数的旧 ``audit-*.log``；
 * ``configure_audit_logging`` 幂等。

-测试只注册空路由，不触达 MySQL / 任何真实业务逻辑；``AuditMiddleware``
-挂在独立的临时 FastAPI app 上，用 ``TestClient`` 发请求。
+测试不 import main.py、不触达 MySQL / 任何真实业务逻辑：用一个带空路由的
+临时 FastAPI app，挂一个复制 access_log 审计契约的 ``BaseHTTPMiddleware``
+（``_AccessLogReplica``），验证 sink 与契约行为。
 """

 from __future__ import annotations

 import os
+from collections.abc import Awaitable, Callable
 from datetime import UTC, datetime
 from pathlib import Path

 import pytest
-from common.auth.jwt import issue_jwt
-from fastapi import FastAPI
+from common.auth.jwt import JwtError, issue_jwt, verify_jwt_token
+from fastapi import FastAPI, Request
+from fastapi.responses import Response
 from fastapi.testclient import TestClient
 from loguru import logger
+from starlette.middleware.base import BaseHTTPMiddleware

 from backend import audit


+def _audit_user_id(request: Request) -> str:
+    """与 main.py 的 _audit_user_id 契约一致：cookie/Bearer 头解 JWT，失败记 '-'。"""
+    token = request.cookies.get("access_token")
+    if not token:
+        auth = request.headers.get("authorization", "")
+        if auth.lower().startswith("bearer "):
+            token = auth[7:].strip()
+    if not token:
+        return "-"
+    try:
+        payload = verify_jwt_token(token)
+    except (JwtError, Exception):  # 任何异常都吞
+        return "-"
+    sub = payload.get("sub")
+    return sub or "-"
+
+
+class _AccessLogReplica(BaseHTTPMiddleware):
+    """复制 main.py access_log 写审计行的契约（不 import 真实 main.py）。
+
+    测试只覆盖 access_log 的审计行为（success / exception 两条路径 +
+    user_id 解析），避免触发 main.py 的 lifespan（MySQL / 路由初始化）。
+    未来 main.py 改 access_log 字段时，这里同步改即可，测试不会假阳/假阴。
+    """
+
+    async def dispatch(
+        self, request: Request, call_next: Callable[..., Awaitable[Response]]
+    ) -> Response:
+        try:
+            response = await call_next(request)
+        except Exception:
+            logger.bind(
+                user_id=_audit_user_id(request),
+                method=request.method,
+                path=request.url.path,
+                status=500,
+            ).info("audit")
+            raise
+        logger.bind(
+            user_id=_audit_user_id(request),
+            method=request.method,
+            path=request.url.path,
+            status=response.status_code,
+        ).info("audit")
+        return response
+
+
@pytest.fixture(autouse=True)
 def _reset_audit_logging():
    """每个用例之间重置审计模块的幂等标志并卸掉上次挂上的审计 sink。
@@ -42,11 +94,13 @@ def _reset_audit_logging():
    audit._CONFIGURED = False


-def _build_client(log_dir: str) -> TestClient:
-    """配置审计日志并返回挂上 AuditMiddleware 的测试 app 客户端。
+def _build_client(
+    log_dir: str, *, raise_server_exceptions: bool = True
+) -> TestClient:
+    """配置审计日志并返回挂上 access_log 契约复刻中间件的测试 app 客户端。

-    只注册空路由，不碰数据库与业务逻辑；中间件与路由的注册顺序与
-    ``main.py`` 保持一致（路由先注册，再挂中间件）。
+    只注册空路由，不碰数据库与业务逻辑；中间件在路由之后注册，与 main.py
+    的 access_log 行为一致。``/boom`` 用于验证 exception 路径（500）。
    """
    audit.configure_audit_logging(log_dir, retention_days=30)

@@ -56,12 +110,12 @@ def _build_client(log_dir: str) -> TestClient:
    def x() -> dict:
        return {"ok": True}

-    @app.get("/api/v1/scripts/{script_id}")
-    def script(script_id: str) -> dict:
-        return {"id": script_id}
+    @app.get("/boom")
+    def boom() -> dict:
+        raise RuntimeError("boom")

-    app.add_middleware(audit.AuditMiddleware)
-    return TestClient(app)
+    app.add_middleware(_AccessLogReplica)
+    return TestClient(app, raise_server_exceptions=raise_server_exceptions)


 def _local_today() -> str:
@@ -103,7 +157,33 @@ def test_audit_log_line_contains_user_method_path_status(tmp_path: Path) -> None
    assert f"| {user_id} | GET /x -> 200" in lines[0]


-def test_audit_log_user_id_dash_when_no_jwt(tmp_path: Path) -> None:
+def test_access_log_writes_audit_line_on_success(tmp_path: Path) -> None:
+    """access_log success 路径写审计行：user_id / status 正确。"""
+    client = _build_client(str(tmp_path))
+    user_id = "01SUCCESS0000000000000000"
+    token = issue_jwt(user_id)
+    client.cookies.set("access_token", token)
+    response = client.get("/x")
+    assert response.status_code == 200
+
+    lines = _audit_lines(tmp_path)
+    assert lines
+    assert f"| {user_id} | GET /x -> 200" in lines[0]
+
+
+def test_access_log_writes_audit_line_on_5xx(tmp_path: Path) -> None:
+    """access_log exception 路径也写审计行：status 记 500。"""
+    client = _build_client(str(tmp_path), raise_server_exceptions=False)
+    response = client.get("/boom")
+    assert response.status_code == 500
+
+    lines = _audit_lines(tmp_path)
+    assert lines
+    assert "GET /boom -> 500" in lines[0]
+
+
+def test_access_log_writes_dash_user_when_no_jwt(tmp_path: Path) -> None:
+    """无 cookie 无 header 时，审计行 user_id 列记 '-'。"""
    client = _build_client(str(tmp_path))
    response = client.get("/x")
    assert response.status_code == 200
@@ -113,31 +193,6 @@ def test_audit_log_user_id_dash_when_no_jwt(tmp_path: Path) -> None:
    assert "| - | GET /x -> 200" in lines[0]


-def test_audit_log_invalid_jwt_does_not_raise_or_skip(tmp_path: Path) -> None:
-    client = _build_client(str(tmp_path))
-    client.cookies.set("access_token", "not.a.jwt")
-    response = client.get("/x")
-    # 坏 JWT 不应拖垮请求：响应依旧正常，审计行照样写，user_id 记 "-"。
-    assert response.status_code == 200
-
-    lines = _audit_lines(tmp_path)
-    assert lines
-    assert "| - | GET /x -> 200" in lines[0]
-
-
-def test_audit_log_includes_ulid_path_params(tmp_path: Path) -> None:
-    client = _build_client(str(tmp_path))
-    ulid = "01ABCDEFGHIJKLMNOPQRSTUVWXYZ"
-    response = client.get(f"/api/v1/scripts/{ulid}")
-    assert response.status_code == 200
-
-    lines = _audit_lines(tmp_path)
-    assert lines
-    # 记录的是实际请求路径，而不是路由模板里的 {script_id}。
-    assert f"GET /api/v1/scripts/{ulid} -> 200" in lines[0]
-    assert "{script_id}" not in lines[0]
-
-
 def test_audit_log_bearer_header_resolves_user(tmp_path: Path) -> None:
    """Authorization: Bearer 头同样能解析出 user_id。"""
    client = _build_client(str(tmp_path))
@@ -194,7 +249,7 @@ def test_configure_audit_logging_is_idempotent(tmp_path: Path) -> None:
    def x() -> dict:
        return {"ok": True}

-    app.add_middleware(audit.AuditMiddleware)
+    app.add_middleware(_AccessLogReplica)
    client = TestClient(app)
    client.get("/x")