duxingchen
42f6e242b4
前端以 <img src="/api/v1/products/qrcode/{sn}"> 引用该端点，而 <img> 无法携带
Authorization 头 —— 加了鉴权会让所有二维码图片加载失败（页面显示成破图），
并在审计里刷出大量 401。
去鉴权是安全的：本函数不查数据库，只校验长度并把这个字符串渲染成二维码，
没有任何业务数据泄露面（序列号本身就是调用方提供的）。也刻意不支持 ?token=
兜底 —— 把 JWT 放进 URL 会渗进访问日志、浏览器历史与 Referer，比它想解决的
问题更糟。
随之而来的副作用必须一并处理：qrcode 路径命中 _TRACKED_READ_PREFIXES 里的
/api/v1/products 前缀、又不是 bare list，会被判成「查看详情」逐条留痕。列表页
一次渲染就并发拉几十张图，逐条留痕会把审计日志塞满，真正有价值的操作反而被
淹没。故把 /api/v1/products/qrcode 加进 _IGNORED_PREFIXES。
实测：二维码正常加载；审计中不再出现 qrcode 记录。
244 lines
9.0 KiB
Python
244 lines
9.0 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",
|
||
)
|
||
|
||
# 永久忽略的路径前缀
|
||
#
|
||
# 两类内容：
|
||
# 1. 探针与文档（/health、/docs…）—— 噪声没有审计价值
|
||
# 2. 图片类端点（/api/v1/products/qrcode）—— 走 <img src> 加载，
|
||
# 一次列表页渲染就会并发拉几十张图，逐条留痕会把审计日志塞满，
|
||
# 真正有价值的操作反而被淹没。它也不含业务数据（只渲染二维码图片）。
|
||
_IGNORED_PREFIXES = (
|
||
"/health", "/docs", "/redoc", "/openapi.json",
|
||
"/api/v1/products/qrcode",
|
||
)
|
||
|
||
# 路径段 → 审计模块
|
||
_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("审计采集失败（已忽略）")
|