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

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

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

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

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

本函数与 IRIS 实例(~/track)代码体逐行一致,只差 docstring —— 两侧审计口径
必须一样,否则排查时日志对不上。约定已记入 AGENTS.md。

实测:MOM 回调后审计记录显示实际操作人姓名;无 operator 时显示「MOM系统」。
This commit is contained in:
2026-09-22 13:45:00 +08:00
parent 754c25b477
commit 9778b8e4b2
2 changed files with 54 additions and 1 deletions

View File

@ -112,6 +112,19 @@
排查口径:**有 `reason` = 被部门校验拦下;无 `reason` = 通过了校验、只是产品没匹配上。** 排查口径:**有 `reason` = 被部门校验拦下;无 `reason` = 通过了校验、只是产品没匹配上。**
- **MOM 回执的审计归因**`endpoints/webhooks.py`2026-09 新增)
外部回调走 X-API-Key、没有 JWT审计中间件读到的 `request.state.audit_user`
永远是空 —— 操作审计里会出现一堆没有归属的「外部系统对接」记录,看不出是谁
扫的码。现在用载荷里的 `operator`MOM 侧实际扫码的那位)写进 `request.state`
归因到人;取不到操作人时写 `"MOM系统"`,而不是继续显示「未认证」。
⚠️ 调用位置必须在 X-API-Key 校验**之后** —— 密钥不对说明载荷本身就不可信,
此时写审计等于允许伪造人。放在两处部门校验**之后**同样有意为之:被拦下的
外来消息不该留下任何归属痕迹。
该函数与 IRIS 实例(`~/track`)代码体逐行一致,只差 docstring —— 改动请两边
同步,否则两个实例的审计口径会对不上。
## 本地起环境(关键,踩过的坑都在这) ## 本地起环境(关键,踩过的坑都在这)
1. **本机没有 Postgres 时需要先装**(容器内 `sudo` 可用): 1. **本机没有 Postgres 时需要先装**(容器内 `sudo` 可用):

View File

@ -22,7 +22,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,38 @@ def _belongs_to_this_org(company_name: str | None) -> bool:
return (company_name or "").strip() == settings.ORG_DEPARTMENT return (company_name or "").strip() == settings.ORG_DEPARTMENT
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 写进审计等于允许伪造人。放在部门校验之后同样有意
为之 —— 被拦下的外来消息不该留下任何归属痕迹。
⚠️ 本函数与 IRIS 实例(~/track逐字一致改动请两边同步MOM 回执的
审计口径必须一样,否则两个实例的日志对不上。
"""
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
# 「撤回出库」信号词 —— 只在 action / event 里做子串匹配。 # 「撤回出库」信号词 —— 只在 action / event 里做子串匹配。
# MOM 侧的字段命名尚未冻结故刻意宽松revoke_outbound / outbound.revoked / # MOM 侧的字段命名尚未冻结故刻意宽松revoke_outbound / outbound.revoked /
# rollback_outbound 都能命中,避免因对方改个词就整条链路失联。 # rollback_outbound 都能命中,避免因对方改个词就整条链路失联。
@ -156,6 +188,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:
@ -182,6 +215,9 @@ async def mom_inbound_webhook(
if not _belongs_to_this_org(payload.company_name): if not _belongs_to_this_org(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)
# ── 匹配产品 ── # ── 匹配产品 ──
@ -377,6 +413,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:
@ -395,6 +432,9 @@ async def mom_outbound_webhook(
if not _belongs_to_this_org(payload.company_name): if not _belongs_to_this_org(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_(