duxingchen
192c8ee9cc
【扩大采集：核心业务数据的「查看详情」】
新增 _TRACKED_READ_PREFIXES（notifications / tasks / orders /
products / records）—— products 是补的：GET /products/scan/{sn}（扫码查询）
是整个车间最高频的读操作，不采它等于没采"活跃度"。
_should_audit 的 GET 分支改为三级判断：
敏感读(export/download/print) → 受跟踪前缀且非裸列表 → 否则不采
用 _is_bare_list 跳过「拉整个列表」：
· 列表接口被前端高频轮询（消息、任务列表尤其明显），逐条留痕会让
audit_logs 迅速膨胀，真正有价值的操作反而被淹没；
· 只有「查看详情」(/tasks/{id}) 才代表用户真的点开了某条业务数据。
判定用「去掉末尾斜杠后是否恰好等于某前缀」，天然排除查询串。
【修正文案】
- "read": "查询" → "查看详情"（前者易被误解成"随便搜了一下"）
- 新增 "mark_read": "标为已读"，并在 _SEGMENT_ACTION 补 "read" 映射 ——
否则 PUT /notifications/{id}/read 会回退到 _METHOD_ACTION(PUT→update)，
把"点开一条通知"记成"修改了某样东西"
- "refresh": "刷新令牌" → "上线"（token 2 小时一换，业务上视作一次上线）
⚠️ 两点须知：
1. notifications / orders 目录下【只有列表路由】，而列表按规则不采集，
故这两个模块不会产生查看记录 —— 后端没有"查看单条消息"的接口。
且消息列表被前端轮询，采集它反而会造成日志爆炸，跳过是正确的。
2. products 被纳入后，工人每扫一次码就多一条记录（50 人 × 每天数百次
≈ 上万条/天）。若嫌吵，去掉该前缀一行即可。
注：读接口鉴权已在上一提交补齐，故这些"查看详情"记录能正确挂上操作人。
235 lines
8.6 KiB
Python
235 lines
8.6 KiB
Python
"""审计采集中间件
|
||
|
||
在响应生成后，把「谁 / 何时 / 从哪来 / 调了哪个接口 / 做了什么 / 结果如何」
|
||
落进 audit_logs。
|
||
|
||
为什么用中间件自动采集，而不是在每个业务函数里手写 record_audit
|
||
------------------------------------------------------------------
|
||
1. 手写必然漏。新加的端点很容易忘记补审计，而审计的价值恰恰建立在「完整」上。
|
||
现状可佐证：task_logs 全项目只有 4 处写入点，凡是不挂在任务上的动作
|
||
（登录、导出、改产品）全都没有留痕。
|
||
2. 中间件能拿到业务函数拿不到的事实：真实来源 IP、UA、最终状态码、
|
||
以及与结构化日志对齐的 request_id。
|
||
3. 业务语义（module / target）由路径推导，不如手写精确，但对「谁动了什么」
|
||
的追责场景已经够用；关键动作后续可再调 record_audit 补 details 做增强。
|
||
|
||
采集范围
|
||
--------
|
||
- 所有写操作（POST/PUT/PATCH/DELETE）
|
||
- 少数**读但敏感**的操作：导出、下载、打印（本项目 GET /people-history/export
|
||
就是导出，只按方法过滤会漏掉）
|
||
|
||
明确不采集：GET /health*、/docs、/openapi.json —— 探针与文档的噪声没有审计价值。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
|
||
from fastapi import Request
|
||
from starlette.middleware.base import BaseHTTPMiddleware, RequestResponseEndpoint
|
||
from starlette.responses import Response
|
||
|
||
from app.services.audit_service import record_audit
|
||
|
||
logger = logging.getLogger("track.audit")
|
||
|
||
# 写操作一律采集
|
||
_MUTATING_METHODS = frozenset({"POST", "PUT", "PATCH", "DELETE"})
|
||
|
||
# 读操作里需要留痕的（导出/下载/打印属于「读」，但把数据带出了系统）
|
||
_SENSITIVE_READ_KEYWORDS = frozenset({"export", "download", "print"})
|
||
|
||
# 核心业务模块 —— 这些前缀下的「查看详情」GET 也采集，
|
||
# 用于回答「谁在什么时候看过哪条业务数据」，而不只是「谁改过」。
|
||
#
|
||
# ⚠️ 只覆盖【核心业务实体】：
|
||
# products —— 移动端扫码查询 GET /products/scan/{sn} 是车间最高频的读操作
|
||
# tasks —— 查看任务详情 /tasks/{id}
|
||
# records —— 任务记录
|
||
# notifications / orders —— 见下方 _is_bare_list 的说明
|
||
_TRACKED_READ_PREFIXES = (
|
||
"/api/v1/notifications",
|
||
"/api/v1/tasks",
|
||
"/api/v1/orders",
|
||
"/api/v1/products",
|
||
"/api/v1/records",
|
||
)
|
||
|
||
# 永久忽略的路径前缀
|
||
_IGNORED_PREFIXES = ("/health", "/docs", "/redoc", "/openapi.json")
|
||
|
||
# 路径段 → 审计模块
|
||
_PATH_MODULE: dict[str, str] = {
|
||
"products": "product",
|
||
"tasks": "task",
|
||
"orders": "order",
|
||
"records": "record",
|
||
"print": "print",
|
||
"materials": "material",
|
||
"users": "user",
|
||
"upload": "upload",
|
||
"notifications": "notification",
|
||
"app-version": "app",
|
||
"analytics": "analytics",
|
||
"dashboard": "dashboard",
|
||
"holidays": "holiday",
|
||
"screen": "screen",
|
||
"webhooks": "external",
|
||
"external": "external",
|
||
"audit": "audit",
|
||
"auth": "auth",
|
||
}
|
||
|
||
# 路径段 → 动作（优先于按 HTTP 方法推断）
|
||
_SEGMENT_ACTION: dict[str, str] = {
|
||
"login": "login",
|
||
"logout": "logout",
|
||
"refresh": "refresh",
|
||
"export": "export",
|
||
"download": "export",
|
||
"print": "print",
|
||
"upload": "upload",
|
||
"finalize": "finalize",
|
||
"receive": "receive",
|
||
"transfer": "transfer",
|
||
"reject": "reject",
|
||
"recall": "recall",
|
||
"spawn": "spawn",
|
||
"complete": "complete",
|
||
"end": "end",
|
||
# 消息已读：PUT /notifications/{id}/read。
|
||
# 没有这一条时会回退到 _METHOD_ACTION（PUT → update → "修改"），
|
||
# 把"点开一条通知"记成"修改了某样东西"，语义完全走样。
|
||
"read": "mark_read",
|
||
}
|
||
|
||
_METHOD_ACTION: dict[str, str] = {
|
||
"POST": "create",
|
||
"PUT": "update",
|
||
"PATCH": "update",
|
||
"DELETE": "delete",
|
||
"GET": "read",
|
||
}
|
||
|
||
# 不可能是业务 ID 的路径段，避免把动作词误当成 target_id
|
||
_NON_ID_SEGMENTS = frozenset(
|
||
set(_SEGMENT_ACTION) | {"api", "v1", "me", "options", "export", "lookup", "batch"}
|
||
)
|
||
|
||
|
||
def _is_bare_list(path: str) -> bool:
|
||
"""判断是否只是「拉整个列表」（如 GET /api/v1/tasks/）。
|
||
|
||
这类请求【不采集】，理由：
|
||
· 列表接口被前端高频轮询（消息、任务列表尤其明显），逐条留痕会让
|
||
audit_logs 迅速膨胀，真正有价值的操作反而被淹没；
|
||
· 「查看详情」(/tasks/{id}) 才代表用户真的点开了某条业务数据。
|
||
|
||
判定用「去掉末尾斜杠后是否恰好等于某个受跟踪前缀」，
|
||
比正则更直观，也天然把查询串排除在外（request.url.path 不含 ?query）。
|
||
"""
|
||
return path.rstrip("/") in _TRACKED_READ_PREFIXES
|
||
|
||
|
||
def _derive_module_and_action(path: str, method: str) -> tuple[str, str, str | None]:
|
||
"""由请求路径与 HTTP 方法推导 (module, action, target_id)"""
|
||
parts = [p for p in path.split("/") if p]
|
||
|
||
module = "other"
|
||
module_idx = -1
|
||
for i, seg in enumerate(parts):
|
||
if seg in _PATH_MODULE:
|
||
module = _PATH_MODULE[seg]
|
||
module_idx = i
|
||
break
|
||
|
||
action = None
|
||
for seg in reversed(parts):
|
||
if seg in _SEGMENT_ACTION:
|
||
action = _SEGMENT_ACTION[seg]
|
||
break
|
||
if action is None:
|
||
action = _METHOD_ACTION.get(method, method.lower())
|
||
|
||
target_id = None
|
||
if module_idx >= 0 and module_idx + 1 < len(parts):
|
||
candidate = parts[module_idx + 1]
|
||
if candidate not in _NON_ID_SEGMENTS:
|
||
target_id = candidate
|
||
|
||
return module, action, target_id
|
||
|
||
|
||
class AuditMiddleware(BaseHTTPMiddleware):
|
||
"""写操作审计采集。
|
||
|
||
必须注册在 RequestContextMiddleware **内层**，因为它依赖后者写入
|
||
request.state 的 request_id 才能与结构化日志对账。
|
||
"""
|
||
|
||
def _should_audit(self, request: Request) -> bool:
|
||
path = request.url.path
|
||
if path.startswith(_IGNORED_PREFIXES):
|
||
return False
|
||
if request.method in _MUTATING_METHODS:
|
||
return True
|
||
if request.method == "GET":
|
||
lowered = path.lower()
|
||
if any(kw in lowered for kw in _SENSITIVE_READ_KEYWORDS):
|
||
return True
|
||
# 核心业务数据的「查看详情」也留痕（证明用户在真的使用系统）
|
||
if path.startswith(_TRACKED_READ_PREFIXES):
|
||
return not _is_bare_list(path)
|
||
return False
|
||
return False
|
||
|
||
async def dispatch(
|
||
self, request: Request, call_next: RequestResponseEndpoint
|
||
) -> Response:
|
||
if not self._should_audit(request):
|
||
return await call_next(request)
|
||
|
||
status_code = 500
|
||
error_message: str | None = None
|
||
try:
|
||
response = await call_next(request)
|
||
status_code = response.status_code
|
||
return response
|
||
except Exception as exc:
|
||
# 异常最终由 ServerErrorMiddleware 转成 500；这里先标记，
|
||
# 保证「失败的操作也有审计」——这正是选用独立 session 的目的
|
||
error_message = f"{type(exc).__name__}: {exc}"[:1000]
|
||
raise
|
||
finally:
|
||
await self._write(request, status_code, error_message)
|
||
|
||
async def _write(
|
||
self, request: Request, status_code: int, error_message: str | None
|
||
) -> None:
|
||
try:
|
||
module, action, target_id = _derive_module_and_action(
|
||
request.url.path, request.method
|
||
)
|
||
client = request.client
|
||
await record_audit(
|
||
action=action,
|
||
module=module,
|
||
user_id=getattr(request.state, "audit_user", None),
|
||
display_name=getattr(request.state, "audit_display_name", None),
|
||
role=getattr(request.state, "audit_role", None),
|
||
target_type=module,
|
||
target_id=target_id,
|
||
# 对产品而言路径里的 ID 就是身份证号，本身即人可读的标识
|
||
target_name=target_id if module == "product" else None,
|
||
ip_address=client.host if client else None,
|
||
user_agent=request.headers.get("user-agent"),
|
||
method=request.method,
|
||
url=request.url.path,
|
||
status_code=status_code,
|
||
error_message=error_message,
|
||
request_id=getattr(request.state, "request_id", None),
|
||
)
|
||
except Exception:
|
||
# record_audit 内部已兜底；这里再兜一层，确保审计绝不冒泡成 500
|
||
logger.exception("审计采集失败（已忽略）")
|