track-LICA/backend/app/core/audit_middleware.py

"""审计采集中间件

在响应生成后，把「谁 / 何时 / 从哪来 / 调了哪个接口 / 做了什么 / 结果如何」
落进 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> 加载，且该端点
#      刻意不加鉴权（见 endpoints/products.py 的说明）。路径命中
#      _TRACKED_READ_PREFIXES 的 /api/v1/products 前缀、又不是 bare list，
#      会被判成「查看详情」逐条留痕：一次列表页渲染就并发拉几十张图，
#      逐条留痕会把审计日志塞满，真正有价值的操作反而被淹没。
#      它也不含业务数据（只把调用方给的序列号渲染成二维码图片）。
_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("审计采集失败（已忽略）")