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。
This commit is contained in:
@@ -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")
|
||||
|
||||
|
||||
Reference in New Issue
Block a user