feat(audit): skip audit log for excluded health/root paths

健康检查与根路径(/health/live、/health/ready、/api/v1/health、
/、/health/storage)没有用户、没业务动作,每秒被 K8s/LB
探针刷一次只会灌进无意义噪音。命中排除集即跳过审计行;
诊断日志(method/path/status/ms 走 stderr)照常打,对容器
运维排错仍有用。

* settings.audit_excluded_paths: list[str] 默认覆盖 5 条
  基础设施路径,env AUDIT_EXCLUDED_PATHS 用逗号分隔
  (pydantic NoDecode + field_validator 兼容 str/list)
* main.py 模块级 _AUDIT_EXCLUDED = frozenset(...),
  access_log 的 success/exception 两条审计行各加守卫
  诊断无条件打
* 测试用 _AccessLogReplica 复刻 access_log 契约(不 import
  真实 main.py),新增 4 个 case:排除根路径、排除 /health/live、
  不排除路径照写审计、自定义排除集

顺带 schedule 模块:ExecutionResult 与 context 已迁到
schedule.domain.*(execution.py / orchestrator.py /
scheduler.py / worker.py),调用点跟进;schedule 自身
18 个测试在改前改后均通过。
This commit is contained in:
tao.chen
2026-09-02 10:10:41 +08:00
committed by tao.chen
parent 8dc2311536
commit 9f7b6ba18c
8 changed files with 162 additions and 45 deletions
+103 -14
View File
@@ -53,29 +53,38 @@ class _AccessLogReplica(BaseHTTPMiddleware):
"""复制 main.py access_log 写审计行的契约(不 import 真实 main.py)。
测试只覆盖 access_log 的审计行为(success / exception 两条路径 +
user_id 解析),避免触发 main.py 的 lifespanMySQL / 路由初始化)。
未来 main.py 改 access_log 字段时,这里同步改即可,测试不会假阳/假阴。
user_id 解析 + 排除集守卫),避免触发 main.py 的 lifespanMySQL /
路由初始化)。``excluded_paths`` 对应 main.py 的 ``_AUDIT_EXCLUDED``
精确路径命中即跳过审计行(诊断 stderr 照常)。未来 main.py 改
access_log 字段时,这里同步改即可,测试不会假阳/假阴。
"""
def __init__(self, app, excluded_paths: frozenset[str] = frozenset()):
super().__init__(app)
self.excluded_paths = excluded_paths
async def dispatch(
self, request: Request, call_next: Callable[..., Awaitable[Response]]
) -> Response:
skip_audit = request.url.path in self.excluded_paths
try:
response = await call_next(request)
except Exception:
if not skip_audit:
logger.bind(
user_id=_audit_user_id(request),
method=request.method,
path=request.url.path,
status=500,
).info("audit")
raise
if not skip_audit:
logger.bind(
user_id=_audit_user_id(request),
method=request.method,
path=request.url.path,
status=500,
status=response.status_code,
).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
@@ -95,12 +104,17 @@ def _reset_audit_logging():
def _build_client(
log_dir: str, *, raise_server_exceptions: bool = True
log_dir: str,
*,
raise_server_exceptions: bool = True,
excluded_paths: frozenset[str] = frozenset(),
) -> TestClient:
"""配置审计日志并返回挂上 access_log 契约复刻中间件的测试 app 客户端。
只注册路由,不碰数据库与业务逻辑;中间件在路由之后注册,与 main.py
的 access_log 行为一致。``/boom`` 用于验证 exception 路径500)。
只注册少量测试路由,不碰数据库与业务逻辑;中间件在路由之后注册,与
main.py 的 access_log 行为一致。``/boom`` 用于验证 exception 路径
500);``/health/live`` ``/api/v1/scripts`` ``/custom`` ``/other``
``/`` 供排除集用例使用。
"""
audit.configure_audit_logging(log_dir, retention_days=30)
@@ -114,7 +128,27 @@ def _build_client(
def boom() -> dict:
raise RuntimeError("boom")
app.add_middleware(_AccessLogReplica)
@app.get("/health/live")
def health_live() -> dict:
return {"ok": True}
@app.get("/api/v1/scripts")
def scripts() -> dict:
return {"ok": True}
@app.get("/custom")
def custom() -> dict:
return {"ok": True}
@app.get("/other")
def other() -> dict:
return {"ok": True}
@app.get("/")
def root() -> dict:
return {"ok": True}
app.add_middleware(_AccessLogReplica, excluded_paths=excluded_paths)
return TestClient(app, raise_server_exceptions=raise_server_exceptions)
@@ -206,6 +240,61 @@ def test_audit_log_bearer_header_resolves_user(tmp_path: Path) -> None:
assert f"| {user_id} | GET /x -> 200" in lines[0]
def test_access_log_skips_audit_for_excluded_path(tmp_path: Path) -> None:
"""命中排除集(/health/live):审计行不写,诊断 stderr 照常。
stderr 走 loguru default sink,难以用 caplog 抓取,这里直接断言
审计文件无新行即可。
"""
client = _build_client(
str(tmp_path), excluded_paths=frozenset({"/health/live"})
)
response = client.get("/health/live")
assert response.status_code == 200
assert _audit_lines(tmp_path) == []
def test_access_log_skips_audit_for_root_path(tmp_path: Path) -> None:
"""根路径 `/` 命中排除集:审计行不写。"""
client = _build_client(str(tmp_path), excluded_paths=frozenset({"/"}))
response = client.get("/")
assert response.status_code == 200
assert _audit_lines(tmp_path) == []
def test_access_log_still_writes_audit_for_non_excluded_path(
tmp_path: Path,
) -> None:
"""未排除路径(/api/v1/scripts)照常写审计行。"""
client = _build_client(
str(tmp_path), excluded_paths=frozenset({"/health/live", "/"})
)
response = client.get("/api/v1/scripts")
assert response.status_code == 200
lines = _audit_lines(tmp_path)
assert lines
assert "GET /api/v1/scripts -> 200" in lines[0]
def test_access_log_custom_excluded_paths_from_settings(tmp_path: Path) -> None:
"""自定义排除集:/custom 命中跳过、/other 未命中照常写。"""
client = _build_client(
str(tmp_path), excluded_paths=frozenset({"/custom"})
)
response = client.get("/custom")
assert response.status_code == 200
assert _audit_lines(tmp_path) == []
response = client.get("/other")
assert response.status_code == 200
lines = _audit_lines(tmp_path)
assert lines
assert "GET /other -> 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"