diff --git a/backend/app/api/v1/endpoints/webhooks.py b/backend/app/api/v1/endpoints/webhooks.py index 9bf108b..fd4dbd7 100644 --- a/backend/app/api/v1/endpoints/webhooks.py +++ b/backend/app/api/v1/endpoints/webhooks.py @@ -11,7 +11,7 @@ from __future__ import annotations from datetime import datetime -from fastapi import APIRouter, Depends, Header, HTTPException +from fastapi import APIRouter, Depends, Header, HTTPException, Request from pydantic import BaseModel from sqlalchemy import or_, select from sqlalchemy.ext.asyncio import AsyncSession @@ -67,6 +67,34 @@ def _is_foreign_company(company_name: str | None) -> bool: return (company_name or "").strip().upper() in _FOREIGN_COMPANIES +def _attribute_audit_to_mom_operator(request: Request, operator: str | None) -> None: + """把外部回调归因到 MOM 侧的实际操作人。 + + 外部回调走 X-API-Key 鉴权、没有 JWT,所以 JWT 依赖不执行, + 审计中间件读到的 request.state.audit_user 永远是空 —— + 操作审计里就出现一堆没有归属的「外部系统对接」记录。 + + 但 MOM 载荷里本来就带着实际操作人(operator,即 MOM 侧扫码的那位), + 写进 request.state 即可让审计归因到人。 + + ⚠️ 必须在 X-API-Key 校验【之后】调用:密钥不对说明载荷本身就不可信, + 此时把 operator 写进审计等于允许伪造人。 + """ + who = (operator or "").strip() + if not who: + # 取不到操作人时留一个明确的系统标记,而不是继续显示「未认证」—— + # 「MOM系统」至少说明这是一次机器回调,不是"一个匿名的人"。 + request.state.audit_user = "MOM系统" + return + request.state.audit_user = who + try: + # 尽力而为:查不到中文名也不影响审计(前端会回退显示账号) + from app.services.mom_cache import get_display_names + request.state.audit_display_name = get_display_names([who]).get(who) or "" + except Exception: # noqa: BLE001 —— 姓名解析失败绝不能影响回调处理 + pass + + def _is_outbound_revoke(payload: MomInboundPayload) -> bool: """payload 是否携带**显式**的撤回出库信号。 @@ -150,6 +178,7 @@ async def _match_inbound_product( @router.post("/mom-inbound") async def mom_inbound_webhook( payload: MomInboundPayload, + request: Request, x_api_key: str | None = Header(default=None, alias="X-API-Key"), db: AsyncSession = Depends(get_db), ) -> dict: @@ -173,6 +202,9 @@ async def mom_inbound_webhook( if _is_foreign_company(payload.company_name): return {"ok": True, "matched": False, "reason": "ignored_company"} + # 归因到 MOM 侧实际扫码的人(必须在鉴权通过之后,见函数注释) + _attribute_audit_to_mom_operator(request, payload.operator) + explicit_revoke = _is_outbound_revoke(payload) # ── 匹配产品 ── @@ -366,6 +398,7 @@ class MomOutboundPayload(BaseModel): @router.post("/mom-outbound") async def mom_outbound_webhook( payload: MomOutboundPayload, + request: Request, x_api_key: str | None = Header(default=None, alias="X-API-Key"), db: AsyncSession = Depends(get_db), ) -> dict: @@ -388,6 +421,9 @@ async def mom_outbound_webhook( if _is_foreign_company(payload.company_name): return {"ok": True, "matched": False, "reason": "ignored_company"} + # 归因到 MOM 侧实际出库的人(必须在鉴权通过之后,见函数注释) + _attribute_audit_to_mom_operator(request, payload.operator) + # ── 按 serial_number(优先)或 sku 匹配"在仓库/已入库"的产品 ── product = None where_cond = or_(