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:
@@ -36,6 +36,11 @@ LOG_LEVEL=INFO
|
||||
# container). AUDIT_LOG_RETENTION_DAYS=0 disables cleanup of old files.
|
||||
AUDIT_LOG_DIR=
|
||||
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:
|
||||
# STORAGE_BACKEND=s3 — connects to an S3-compatible service (MinIO,
|
||||
|
||||
+22
-12
@@ -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
|
||||
async def lifespan(app: Any) -> AsyncIterator[None]:
|
||||
# 生命周期内创建的对象挂在 app.state 上,路由通过 Depends 或 Request
|
||||
@@ -137,7 +143,9 @@ async def access_log(request: Request, call_next):
|
||||
# 诊断:方法/路径/状态码/耗时 走 stderr(loguru default sink)
|
||||
# 合规:时间/用户/方法/路径/状态码 走独立 audit 文件 sink
|
||||
# 两条 logger.info() 共用一个出口,便于排查
|
||||
# 排除集是精确路径匹配(不含 query),命中即跳过审计行;诊断日志照常。
|
||||
start = time.perf_counter()
|
||||
skip_audit = request.url.path in _AUDIT_EXCLUDED
|
||||
try:
|
||||
response = await call_next(request)
|
||||
except Exception:
|
||||
@@ -147,12 +155,13 @@ async def access_log(request: Request, call_next):
|
||||
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")
|
||||
if not skip_audit:
|
||||
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(
|
||||
@@ -160,12 +169,13 @@ 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")
|
||||
if not skip_audit:
|
||||
logger.bind(
|
||||
user_id=_audit_user_id(request),
|
||||
method=request.method,
|
||||
path=request.url.path,
|
||||
status=response.status_code,
|
||||
).info("audit")
|
||||
return response
|
||||
|
||||
|
||||
|
||||
@@ -53,29 +53,38 @@ class _AccessLogReplica(BaseHTTPMiddleware):
|
||||
"""复制 main.py access_log 写审计行的契约(不 import 真实 main.py)。
|
||||
|
||||
测试只覆盖 access_log 的审计行为(success / exception 两条路径 +
|
||||
user_id 解析),避免触发 main.py 的 lifespan(MySQL / 路由初始化)。
|
||||
未来 main.py 改 access_log 字段时,这里同步改即可,测试不会假阳/假阴。
|
||||
user_id 解析 + 排除集守卫),避免触发 main.py 的 lifespan(MySQL /
|
||||
路由初始化)。``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"
|
||||
|
||||
@@ -18,9 +18,10 @@ Rules for adding a new variable:
|
||||
from __future__ import annotations
|
||||
|
||||
from functools import lru_cache
|
||||
from typing import Annotated
|
||||
|
||||
from pydantic import Field
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
from pydantic import Field, field_validator
|
||||
from pydantic_settings import BaseSettings, NoDecode, SettingsConfigDict
|
||||
|
||||
|
||||
class Settings(BaseSettings):
|
||||
@@ -79,6 +80,29 @@ class Settings(BaseSettings):
|
||||
default=30,
|
||||
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_api_url: str = Field(
|
||||
|
||||
@@ -4,10 +4,10 @@ import asyncio
|
||||
import json
|
||||
import sys
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
from pathlib import Path, PurePosixPath
|
||||
|
||||
from loguru import logger
|
||||
from schedule.domain.execution import ExecutionResult
|
||||
|
||||
MAX_LOG_BYTES = 4 * 1024 * 1024
|
||||
|
||||
@@ -26,18 +26,6 @@ def _resolve_python(version: str) -> str:
|
||||
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:
|
||||
encoded = value.encode("utf-8", errors="replace")
|
||||
if len(encoded) <= MAX_LOG_BYTES:
|
||||
|
||||
@@ -30,7 +30,7 @@ from loguru import logger
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
|
||||
|
||||
from schedule.context import (
|
||||
from schedule.domain.context import (
|
||||
FAILED_NODE_STATES,
|
||||
TERMINAL_NODE_STATES,
|
||||
TERMINAL_RUN_STATES,
|
||||
|
||||
@@ -28,7 +28,7 @@ from loguru import logger
|
||||
from sqlalchemy import select
|
||||
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
|
||||
# ``CronScheduler`` method captures this instance (including SQLAlchemy engine
|
||||
|
||||
@@ -41,8 +41,9 @@ from common.scheduler.trigger import SYSTEM_CRON_USER_ID
|
||||
from loguru import logger
|
||||
from sqlalchemy import select
|
||||
|
||||
from schedule.context import TERMINAL_NODE_STATES
|
||||
from schedule.execution import ExecutionResult, execute_artifact
|
||||
from schedule.domain.context import TERMINAL_NODE_STATES
|
||||
from schedule.domain.execution import ExecutionResult
|
||||
from schedule.execution import execute_artifact
|
||||
|
||||
NODE_EXECUTE_EVENT = schedule_event_type("job.node.execute")
|
||||
NODE_FINISHED_EVENT = schedule_event_type("job.node.finished")
|
||||
|
||||
Reference in New Issue
Block a user