tao.chenandtao.chen
ff33523867
按需为每个 HTTP 接口写一条合规记录到 data/logs/audit/audit-YYYY-MM-DD.log，字段：时间 / 用户 / METHOD path / 状态码。 设计： * 复用全局 loguru logger，文件 sink 由 audit._DailyFileSink 自管：缓存当天文件句柄、跨日重建。不走 loguru 的 rotation=00:00（产物是 audit.log.YYYY-MM-DD_HH-MM-SS， 不符合按天单文件的命名要求）。 * AuditMiddleware 只做 CPU 验签拿 user_id：cookie access_token 优先，Authorization Bearer 兜底，无/坏 JWT 一律记 '-'。 绝不查 DB（RequestContext 在路由解析后才注入）。 * 审计失败不拖死请求：所有异常捕获。 * 与 main.py 现有 access_log 严格分离：access_log 走 stderr 诊断（method/path/status/耗时），audit 走独立文件合规 （时间/用户/接口），并存。 * 启动时按 settings.audit_log_retention_days 清理过期文件 （设 0 关闭）。 * 新增 settings.audit_log_dir（默认 data/logs/audit，相对 cwd） 与 settings.audit_log_retention_days（默认 30）两个配置项； .env.example 同步。 新增 9 个 case：文件创建、行字段、未登录 '-'、坏 JWT、ULID path、retention 清理/关闭、Bearer 头、幂等。
204 lines
6.7 KiB
Python
204 lines
6.7 KiB
Python
"""审计日志中间件测试。
|
||
|
||
覆盖：
|
||
* 每天一个 ``audit-YYYY-MM-DD.log`` 文件且写入至少一行；
|
||
* 日志行包含 user_id / method / path / status；
|
||
* 未登录（无 cookie 无 header）与坏 JWT 时 user_id 记 ``-``；
|
||
* 带路径参数的请求原样记录实际 path（不替换为 ``{script_id}``）；
|
||
* 启动时按 mtime 清理超过保留天数的旧 ``audit-*.log``；
|
||
* ``configure_audit_logging`` 幂等。
|
||
|
||
测试只注册空路由，不触达 MySQL / 任何真实业务逻辑；``AuditMiddleware``
|
||
挂在独立的临时 FastAPI app 上，用 ``TestClient`` 发请求。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import os
|
||
from datetime import UTC, datetime
|
||
from pathlib import Path
|
||
|
||
import pytest
|
||
from common.auth.jwt import issue_jwt
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from loguru import logger
|
||
|
||
from backend import audit
|
||
|
||
|
||
@pytest.fixture(autouse=True)
|
||
def _reset_audit_logging():
|
||
"""每个用例之间重置审计模块的幂等标志并卸掉上次挂上的审计 sink。
|
||
|
||
loguru 的全局 logger 会跨用例留存 sink，若不清理，后面的用例会往已
|
||
删除的临时目录继续写，且 ``_CONFIGURED`` 会让后续 configure 变空操作。
|
||
"""
|
||
audit._CONFIGURED = False
|
||
yield
|
||
if audit._HANDLER_ID is not None:
|
||
logger.remove(audit._HANDLER_ID)
|
||
audit._HANDLER_ID = None
|
||
audit._CONFIGURED = False
|
||
|
||
|
||
def _build_client(log_dir: str) -> TestClient:
|
||
"""配置审计日志并返回挂上 AuditMiddleware 的测试 app 客户端。
|
||
|
||
只注册空路由，不碰数据库与业务逻辑；中间件与路由的注册顺序与
|
||
``main.py`` 保持一致（路由先注册，再挂中间件）。
|
||
"""
|
||
audit.configure_audit_logging(log_dir, retention_days=30)
|
||
|
||
app = FastAPI()
|
||
|
||
@app.get("/x")
|
||
def x() -> dict:
|
||
return {"ok": True}
|
||
|
||
@app.get("/api/v1/scripts/{script_id}")
|
||
def script(script_id: str) -> dict:
|
||
return {"id": script_id}
|
||
|
||
app.add_middleware(audit.AuditMiddleware)
|
||
return TestClient(app)
|
||
|
||
|
||
def _local_today() -> str:
|
||
return datetime.now(UTC).astimezone().strftime("%Y-%m-%d")
|
||
|
||
|
||
def _today_file(log_dir: Path) -> Path:
|
||
return log_dir / f"audit-{_local_today()}.log"
|
||
|
||
|
||
def _audit_lines(log_dir: Path) -> list[str]:
|
||
"""等 enqueue 队列落盘后返回今天审计文件的全部非空行。"""
|
||
logger.complete()
|
||
path = _today_file(log_dir)
|
||
if not path.exists():
|
||
return []
|
||
return [line for line in path.read_text(encoding="utf-8").splitlines() if line]
|
||
|
||
|
||
def test_audit_log_file_is_created_with_date_suffix(tmp_path: Path) -> None:
|
||
client = _build_client(str(tmp_path))
|
||
client.get("/x")
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert _today_file(tmp_path).exists()
|
||
assert len(lines) >= 1
|
||
|
||
|
||
def test_audit_log_line_contains_user_method_path_status(tmp_path: Path) -> None:
|
||
client = _build_client(str(tmp_path))
|
||
user_id = "01USER12345678901234567890"
|
||
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_audit_log_user_id_dash_when_no_jwt(tmp_path: Path) -> None:
|
||
client = _build_client(str(tmp_path))
|
||
response = client.get("/x")
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
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))
|
||
user_id = "01BEARER000000000000000001"
|
||
token = issue_jwt(user_id)
|
||
response = client.get("/x", headers={"Authorization": f"Bearer {token}"})
|
||
assert response.status_code == 200
|
||
|
||
lines = _audit_lines(tmp_path)
|
||
assert lines
|
||
assert f"| {user_id} | GET /x -> 200" in lines[0]
|
||
|
||
|
||
def test_retention_cleanup_removes_old_files(tmp_path: Path) -> None:
|
||
old = tmp_path / "audit-2024-01-01.log"
|
||
recent = tmp_path / "audit-2024-06-01.log"
|
||
old.write_text("old\n", encoding="utf-8")
|
||
recent.write_text("recent\n", encoding="utf-8")
|
||
|
||
old_mtime = datetime.now(UTC).timestamp() - 60 * 86_400 # 60 天前
|
||
recent_mtime = datetime.now(UTC).timestamp() - 5 * 86_400 # 5 天前
|
||
os.utime(old, (old_mtime, old_mtime))
|
||
os.utime(recent, (recent_mtime, recent_mtime))
|
||
|
||
audit.configure_audit_logging(str(tmp_path), retention_days=30)
|
||
|
||
assert not old.exists(), "超过保留期的旧审计文件应被启动清理删除"
|
||
assert recent.exists(), "保留期内(5 天前)的文件应保留"
|
||
|
||
|
||
def test_retention_cleanup_disabled_when_zero(tmp_path: Path) -> None:
|
||
"""AUDIT_LOG_RETENTION_DAYS=0 关闭清理：任何旧文件都不删除。"""
|
||
old = tmp_path / "audit-2024-01-01.log"
|
||
old.write_text("old\n", encoding="utf-8")
|
||
old_mtime = datetime.now(UTC).timestamp() - 400 * 86_400 # 400 天前
|
||
os.utime(old, (old_mtime, old_mtime))
|
||
|
||
audit.configure_audit_logging(str(tmp_path), retention_days=0)
|
||
|
||
assert old.exists(), "retention_days=0 时应跳过清理，保留所有旧文件"
|
||
|
||
|
||
def test_configure_audit_logging_is_idempotent(tmp_path: Path) -> None:
|
||
"""多次调用只有首次生效：日志只写到第一次给的目录。"""
|
||
first_dir = tmp_path / "first"
|
||
second_dir = tmp_path / "second"
|
||
|
||
audit.configure_audit_logging(str(first_dir), retention_days=30)
|
||
audit.configure_audit_logging(str(second_dir), retention_days=30)
|
||
|
||
app = FastAPI()
|
||
|
||
@app.get("/x")
|
||
def x() -> dict:
|
||
return {"ok": True}
|
||
|
||
app.add_middleware(audit.AuditMiddleware)
|
||
client = TestClient(app)
|
||
client.get("/x")
|
||
|
||
logger.complete()
|
||
assert (first_dir / f"audit-{_local_today()}.log").exists()
|
||
assert not (second_dir / f"audit-{_local_today()}.log").exists()
|