feat(audit): MOM 回调归因到实际操作人,不再显示「未认证」

外部回调走 X-API-Key 鉴权、没有 JWT,JWT 依赖不执行,审计中间件读到的
request.state.audit_user 永远是空 —— 操作审计里就出现一堆没有归属的
「外部系统对接」记录,看不出是谁扫的码。

MOM 载荷里本来就带着实际操作人(operator,即 MOM 侧扫码的那位),写进
request.state 即可让审计归因到人;顺手解析中文姓名(查不到也不影响审计,
前端会回退显示账号)。

⚠️ 调用位置必须在 X-API-Key 校验【之后】:密钥不对说明载荷本身就不可信,
   此时把 operator 写进审计等于允许伪造人。放在部门校验之后同样有意为之 ——
   被拦下的外来消息不该留下任何归属痕迹。

取不到操作人时写 "MOM系统" 而非留空:「MOM系统」至少说明这是一次机器回调,
比继续显示「未认证」(读起来像"一个匿名的人")更准确。

实测:MOM 回调后审计记录显示实际操作人姓名;无 operator 时显示「MOM系统」。
This commit is contained in:
2026-09-22 13:44:05 +08:00
parent 817183062d
commit 3c94faf078

View File

@ -11,7 +11,7 @@ from __future__ import annotations
from datetime import datetime from datetime import datetime
from fastapi import APIRouter, Depends, Header, HTTPException from fastapi import APIRouter, Depends, Header, HTTPException, Request
from pydantic import BaseModel from pydantic import BaseModel
from sqlalchemy import or_, select from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession 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 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: def _is_outbound_revoke(payload: MomInboundPayload) -> bool:
"""payload 是否携带**显式**的撤回出库信号。 """payload 是否携带**显式**的撤回出库信号。
@ -150,6 +178,7 @@ async def _match_inbound_product(
@router.post("/mom-inbound") @router.post("/mom-inbound")
async def mom_inbound_webhook( async def mom_inbound_webhook(
payload: MomInboundPayload, payload: MomInboundPayload,
request: Request,
x_api_key: str | None = Header(default=None, alias="X-API-Key"), x_api_key: str | None = Header(default=None, alias="X-API-Key"),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
) -> dict: ) -> dict:
@ -173,6 +202,9 @@ async def mom_inbound_webhook(
if _is_foreign_company(payload.company_name): if _is_foreign_company(payload.company_name):
return {"ok": True, "matched": False, "reason": "ignored_company"} return {"ok": True, "matched": False, "reason": "ignored_company"}
# 归因到 MOM 侧实际扫码的人(必须在鉴权通过之后,见函数注释)
_attribute_audit_to_mom_operator(request, payload.operator)
explicit_revoke = _is_outbound_revoke(payload) explicit_revoke = _is_outbound_revoke(payload)
# ── 匹配产品 ── # ── 匹配产品 ──
@ -366,6 +398,7 @@ class MomOutboundPayload(BaseModel):
@router.post("/mom-outbound") @router.post("/mom-outbound")
async def mom_outbound_webhook( async def mom_outbound_webhook(
payload: MomOutboundPayload, payload: MomOutboundPayload,
request: Request,
x_api_key: str | None = Header(default=None, alias="X-API-Key"), x_api_key: str | None = Header(default=None, alias="X-API-Key"),
db: AsyncSession = Depends(get_db), db: AsyncSession = Depends(get_db),
) -> dict: ) -> dict:
@ -388,6 +421,9 @@ async def mom_outbound_webhook(
if _is_foreign_company(payload.company_name): if _is_foreign_company(payload.company_name):
return {"ok": True, "matched": False, "reason": "ignored_company"} return {"ok": True, "matched": False, "reason": "ignored_company"}
# 归因到 MOM 侧实际出库的人(必须在鉴权通过之后,见函数注释)
_attribute_audit_to_mom_operator(request, payload.operator)
# ── 按 serial_number(优先)或 sku 匹配"在仓库/已入库"的产品 ── # ── 按 serial_number(优先)或 sku 匹配"在仓库/已入库"的产品 ──
product = None product = None
where_cond = or_( where_cond = or_(