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-08-21 13:47:44 +08:00
parent d2c87de32c
commit cdcfcb2e43
8 changed files with 162 additions and 45 deletions
+5
View File
@@ -36,6 +36,11 @@ LOG_LEVEL=INFO
# container). AUDIT_LOG_RETENTION_DAYS=0 disables cleanup of old files. # container). AUDIT_LOG_RETENTION_DAYS=0 disables cleanup of old files.
AUDIT_LOG_DIR= AUDIT_LOG_DIR=
AUDIT_LOG_RETENTION_DAYS=30 AUDIT_LOG_RETENTION_DAYS=30
# Exact paths excluded from the audit line (health/root probes carry no
# business value but fire every second from K8s/LB). The default already
# covers /health/live /health/ready /api/v1/health / /health/storage;
# leave empty to keep the default. Comma-separated, e.g. /health/live,/api/v1/health.
AUDIT_EXCLUDED_PATHS=
# Object storage. Two modes are supported: # Object storage. Two modes are supported:
# STORAGE_BACKEND=s3 — connects to an S3-compatible service (MinIO, # STORAGE_BACKEND=s3 — connects to an S3-compatible service (MinIO,
+22 -12
View File
@@ -53,6 +53,12 @@ configure_audit_logging(
) )
# 审计排除的精确路径集(不含 query):命中即跳过审计行,诊断日志照常打。
# 健康检查 / 根路径探针每秒刷审计文件但无业务价值;可用
# settings.audit_excluded_paths / AUDIT_EXCLUDED_PATHS 覆盖默认值。
_AUDIT_EXCLUDED: frozenset[str] = frozenset(settings.audit_excluded_paths)
@asynccontextmanager @asynccontextmanager
async def lifespan(app: Any) -> AsyncIterator[None]: async def lifespan(app: Any) -> AsyncIterator[None]:
# 生命周期内创建的对象挂在 app.state 上,路由通过 Depends 或 Request # 生命周期内创建的对象挂在 app.state 上,路由通过 Depends 或 Request
@@ -137,7 +143,9 @@ async def access_log(request: Request, call_next):
# 诊断:方法/路径/状态码/耗时 走 stderrloguru default sink # 诊断:方法/路径/状态码/耗时 走 stderrloguru default sink
# 合规:时间/用户/方法/路径/状态码 走独立 audit 文件 sink # 合规:时间/用户/方法/路径/状态码 走独立 audit 文件 sink
# 两条 logger.info() 共用一个出口,便于排查 # 两条 logger.info() 共用一个出口,便于排查
# 排除集是精确路径匹配(不含 query),命中即跳过审计行;诊断日志照常。
start = time.perf_counter() start = time.perf_counter()
skip_audit = request.url.path in _AUDIT_EXCLUDED
try: try:
response = await call_next(request) response = await call_next(request)
except Exception: except Exception:
@@ -147,12 +155,13 @@ async def access_log(request: Request, call_next):
method=request.method, path=request.url.path, ms=elapsed_ms, method=request.method, path=request.url.path, ms=elapsed_ms,
) )
# 异常路径:审计行也要写(status=500 由 unhandled_exception_handler 返回) # 异常路径:审计行也要写(status=500 由 unhandled_exception_handler 返回)
logger.bind( if not skip_audit:
user_id=_audit_user_id(request), logger.bind(
method=request.method, user_id=_audit_user_id(request),
path=request.url.path, method=request.method,
status=500, path=request.url.path,
).info("audit") status=500,
).info("audit")
raise raise
elapsed_ms = (time.perf_counter() - start) * 1000 elapsed_ms = (time.perf_counter() - start) * 1000
logger.info( logger.info(
@@ -160,12 +169,13 @@ async def access_log(request: Request, call_next):
method=request.method, path=request.url.path, method=request.method, path=request.url.path,
status=response.status_code, ms=elapsed_ms, status=response.status_code, ms=elapsed_ms,
) )
logger.bind( if not skip_audit:
user_id=_audit_user_id(request), logger.bind(
method=request.method, user_id=_audit_user_id(request),
path=request.url.path, method=request.method,
status=response.status_code, path=request.url.path,
).info("audit") status=response.status_code,
).info("audit")
return response return response
+103 -14
View File
@@ -53,29 +53,38 @@ class _AccessLogReplica(BaseHTTPMiddleware):
"""复制 main.py access_log 写审计行的契约(不 import 真实 main.py)。 """复制 main.py access_log 写审计行的契约(不 import 真实 main.py)。
测试只覆盖 access_log 的审计行为(success / exception 两条路径 + 测试只覆盖 access_log 的审计行为(success / exception 两条路径 +
user_id 解析),避免触发 main.py 的 lifespanMySQL / 路由初始化)。 user_id 解析 + 排除集守卫),避免触发 main.py 的 lifespanMySQL /
未来 main.py 改 access_log 字段时,这里同步改即可,测试不会假阳/假阴。 路由初始化)。``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( async def dispatch(
self, request: Request, call_next: Callable[..., Awaitable[Response]] self, request: Request, call_next: Callable[..., Awaitable[Response]]
) -> Response: ) -> Response:
skip_audit = request.url.path in self.excluded_paths
try: try:
response = await call_next(request) response = await call_next(request)
except Exception: 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( logger.bind(
user_id=_audit_user_id(request), user_id=_audit_user_id(request),
method=request.method, method=request.method,
path=request.url.path, path=request.url.path,
status=500, status=response.status_code,
).info("audit") ).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 return response
@@ -95,12 +104,17 @@ def _reset_audit_logging():
def _build_client( 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: ) -> TestClient:
"""配置审计日志并返回挂上 access_log 契约复刻中间件的测试 app 客户端。 """配置审计日志并返回挂上 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) audit.configure_audit_logging(log_dir, retention_days=30)
@@ -114,7 +128,27 @@ def _build_client(
def boom() -> dict: def boom() -> dict:
raise RuntimeError("boom") 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) 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] 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: def test_retention_cleanup_removes_old_files(tmp_path: Path) -> None:
old = tmp_path / "audit-2024-01-01.log" old = tmp_path / "audit-2024-01-01.log"
recent = tmp_path / "audit-2024-06-01.log" recent = tmp_path / "audit-2024-06-01.log"
+26 -2
View File
@@ -18,9 +18,10 @@ Rules for adding a new variable:
from __future__ import annotations from __future__ import annotations
from functools import lru_cache from functools import lru_cache
from typing import Annotated
from pydantic import Field from pydantic import Field, field_validator
from pydantic_settings import BaseSettings, SettingsConfigDict from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
class Settings(BaseSettings): class Settings(BaseSettings):
@@ -79,6 +80,29 @@ class Settings(BaseSettings):
default=30, default=30,
description="审计日志保留天数;过期文件启动时清理。设 0 关闭清理。", description="审计日志保留天数;过期文件启动时清理。设 0 关闭清理。",
) )
audit_excluded_paths: Annotated[list[str], NoDecode] = Field(
default=[
"/health/live",
"/health/ready",
"/api/v1/health",
"/",
"/health/storage",
],
description=(
"审计排除的精确路径列表(不含 query)。命中即不写审计行。"
"诊断日志(method/path/status/ms)仍写 stderr。"
"环境变量 AUDIT_EXCLUDED_PATHS 用逗号分隔,例如"
" '/health/live,/api/v1/health'"
),
)
@field_validator("audit_excluded_paths", mode="before")
@classmethod
def _split_audit_paths(cls, v):
# env 进来是 "a,b,c";代码里直接传 list 也行
if isinstance(v, str):
return [s.strip() for s in v.split(",") if s.strip()]
return v
# ── runtime container endpoint ─────────────────────────────── # ── runtime container endpoint ───────────────────────────────
runtime_api_url: str = Field( runtime_api_url: str = Field(
+1 -13
View File
@@ -4,10 +4,10 @@ import asyncio
import json import json
import sys import sys
import tempfile import tempfile
from dataclasses import dataclass
from pathlib import Path, PurePosixPath from pathlib import Path, PurePosixPath
from loguru import logger from loguru import logger
from schedule.domain.execution import ExecutionResult
MAX_LOG_BYTES = 4 * 1024 * 1024 MAX_LOG_BYTES = 4 * 1024 * 1024
@@ -26,18 +26,6 @@ def _resolve_python(version: str) -> str:
raise ValueError(f"unsupported python_version: {version}") raise ValueError(f"unsupported python_version: {version}")
@dataclass(frozen=True)
class ExecutionResult:
status: str
exit_code: int | None
logs: bytes
result: bytes
result_file_name: str
result_content_type: str
error_code: str | None = None
error_message: str | None = None
def _limited_log(value: str) -> bytes: def _limited_log(value: str) -> bytes:
encoded = value.encode("utf-8", errors="replace") encoded = value.encode("utf-8", errors="replace")
if len(encoded) <= MAX_LOG_BYTES: if len(encoded) <= MAX_LOG_BYTES:
+1 -1
View File
@@ -30,7 +30,7 @@ from loguru import logger
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from schedule.context import ( from schedule.domain.context import (
FAILED_NODE_STATES, FAILED_NODE_STATES,
TERMINAL_NODE_STATES, TERMINAL_NODE_STATES,
TERMINAL_RUN_STATES, TERMINAL_RUN_STATES,
+1 -1
View File
@@ -28,7 +28,7 @@ from loguru import logger
from sqlalchemy import select from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from schedule.context import naive_utc from schedule.domain.context import naive_utc
# APScheduler's persistent SQLAlchemy job store pickles each job. A bound # APScheduler's persistent SQLAlchemy job store pickles each job. A bound
# ``CronScheduler`` method captures this instance (including SQLAlchemy engine # ``CronScheduler`` method captures this instance (including SQLAlchemy engine
+3 -2
View File
@@ -41,8 +41,9 @@ from common.scheduler.trigger import SYSTEM_CRON_USER_ID
from loguru import logger from loguru import logger
from sqlalchemy import select from sqlalchemy import select
from schedule.context import TERMINAL_NODE_STATES from schedule.domain.context import TERMINAL_NODE_STATES
from schedule.execution import ExecutionResult, execute_artifact from schedule.domain.execution import ExecutionResult
from schedule.execution import execute_artifact
NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute") NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute")
NODE_FINISHED_EVENT = schedule_event_type("job.node.finished") NODE_FINISHED_EVENT = schedule_event_type("job.node.finished")