From 3c94faf078cee66d79f0f9baff670c0800389dbc Mon Sep 17 00:00:00 2001 From: duxingchen Date: Tue, 22 Sep 2026 13:44:05 +0800 Subject: [PATCH] =?UTF-8?q?feat(audit):=20MOM=20=E5=9B=9E=E8=B0=83?= =?UTF-8?q?=E5=BD=92=E5=9B=A0=E5=88=B0=E5=AE=9E=E9=99=85=E6=93=8D=E4=BD=9C?= =?UTF-8?q?=E4=BA=BA=EF=BC=8C=E4=B8=8D=E5=86=8D=E6=98=BE=E7=A4=BA=E3=80=8C?= =?UTF-8?q?=E6=9C=AA=E8=AE=A4=E8=AF=81=E3=80=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 外部回调走 X-API-Key 鉴权、没有 JWT，JWT 依赖不执行，审计中间件读到的 request.state.audit_user 永远是空 —— 操作审计里就出现一堆没有归属的 「外部系统对接」记录，看不出是谁扫的码。 MOM 载荷里本来就带着实际操作人（operator，即 MOM 侧扫码的那位），写进 request.state 即可让审计归因到人；顺手解析中文姓名（查不到也不影响审计， 前端会回退显示账号）。 ⚠️ 调用位置必须在 X-API-Key 校验【之后】：密钥不对说明载荷本身就不可信， 此时把 operator 写进审计等于允许伪造人。放在部门校验之后同样有意为之 —— 被拦下的外来消息不该留下任何归属痕迹。 取不到操作人时写 "MOM系统" 而非留空：「MOM系统」至少说明这是一次机器回调， 比继续显示「未认证」（读起来像"一个匿名的人"）更准确。 实测：MOM 回调后审计记录显示实际操作人姓名；无 operator 时显示「MOM系统」。 --- backend/app/api/v1/endpoints/webhooks.py | 38 +++++++++++++++++++++++- 1 file changed, 37 insertions(+), 1 deletion(-) 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_(