diff --git a/backend/src/backend/audit.py b/backend/src/backend/audit.py index 831ea90..e73d5b2 100644 --- a/backend/src/backend/audit.py +++ b/backend/src/backend/audit.py @@ -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", ] diff --git a/backend/src/backend/main.py b/backend/src/backend/main.py index 451993f..4fea196 100644 --- a/backend/src/backend/main.py +++ b/backend/src/backend/main.py @@ -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 diff --git a/backend/tests/test_audit_logging.py b/backend/tests/test_audit_logging.py index 67a6166..f767971 100644 --- a/backend/tests/test_audit_logging.py +++ b/backend/tests/test_audit_logging.py @@ -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")