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.
|
# 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
@@ -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):
|
|||||||
# 诊断:方法/路径/状态码/耗时 走 stderr(loguru default sink)
|
# 诊断:方法/路径/状态码/耗时 走 stderr(loguru 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
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
@@ -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 的 lifespan(MySQL / 路由初始化)。
|
user_id 解析 + 排除集守卫),避免触发 main.py 的 lifespan(MySQL /
|
||||||
未来 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"
|
||||||
|
|||||||
@@ -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(
|
||||||
|
|||||||
@@ -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:
|
||||||
|
|||||||
@@ -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,
|
||||||
|
|||||||
@@ -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
|
||||||
|
|||||||
@@ -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")
|
||||||
|
|||||||
Reference in New Issue
Block a user