Compare commits

..

3 Commits

Author SHA1 Message Date
700486d5d5 fix(webhook): 白名单判定容忍大小写,堵住「两边都丢弃」的静默黑洞
_belongs_to_this_org 此前是 (company_name or "").strip() == ORG_DEPARTMENT ——
只 strip 不 upper,而 IRIS 的 _is_foreign_company 做了 .upper()。两边口径不一致
的后果:

  company_name = "lica" / "Lica"
    → IRIS:.upper() 命中外来名单,拒掉
    → LICA:不等于 "LICA",白名单拒掉
    → 双方都回 200,消息彻底静默丢失

这正是本文档最该防的那种「漏」。改为 .strip().upper() 与 IRIS 对齐。

「严格」指的是白名单语义(只有本部门才收),不是逐字节比对 —— 空白值依然被
拒、未知部门名依然被拒,只是不再因为大小写写法白白丢消息。

实测 11 种取值:修复前 "lica" / "Lica" 是两个黑洞(两边都丢弃),修复后 0 个;
空白 / 全空格 / "IRIS" / "UNKNOWN" / None 仍然全部忽略,严格性未放松。

顺带修正 outbound_type 的取值注释:MOM 的 TransOutbound.outbound_type 实际是
SALES / USE / PRODUCTION(见 projects inventory-backend/app/models/outbound.py
第 136 行),原注释漏了 USE。
2026-09-22 13:46:12 +08:00
9778b8e4b2 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系统」。
2026-09-22 13:45:00 +08:00
754c25b477 fix(audit): qrcode 路径排除出审计采集(与 IRIS 实例对齐)
qrcode 端点刻意不加鉴权(前端走 <img src> 引用,无法携带 Authorization 头),
而 /api/v1/products/qrcode/{sn} 命中 _TRACKED_READ_PREFIXES 里的 /api/v1/products
前缀、又不是 bare list,会被判成「查看详情」逐条留痕 —— 列表页一次渲染就并发
拉几十张图,审计日志会被塞满,真正有价值的操作反而被淹没。

该端点不含业务数据(只把调用方给的序列号渲染成二维码图片),排除掉没有审计
价值损失。IRIS 实例今日已做同样处理,此处对齐,避免两边行为不一致。

实测:二维码正常加载;审计中不再出现 qrcode 记录。
2026-09-22 13:44:17 +08:00
3 changed files with 74 additions and 4 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
@ -63,8 +63,45 @@ def _belongs_to_this_org(company_name: str | None) -> bool:
配置,不收就彻底丢了。 配置,不收就彻底丢了。
LICA 没有兜底角色,空白值只可能来自"MOM 没判定出公司",那本就该由 LICA 没有兜底角色,空白值只可能来自"MOM 没判定出公司",那本就该由
IRIS 兜。**宁可漏,不可误收**:误收会把别的部门的设备状态改掉。 IRIS 兜。**宁可漏,不可误收**:误收会把别的部门的设备状态改掉。
⚠️ 大小写与首尾空白都容忍(.strip().upper()),与 IRIS 的判定口径对齐。
两边若一个容忍、一个不容忍,小写写法("lica")会同时被 IRIS 当外来户
拒掉、又被本实例的白名单拒掉 —— 双方都回 200,消息静默丢失,正是本文
档最该防的那种「漏」。「严格」指的是白名单语义,不是逐字节比对。
""" """
return (company_name or "").strip() == settings.ORG_DEPARTMENT return (company_name or "").strip().upper() == settings.ORG_DEPARTMENT.strip().upper()
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 里做子串匹配。
@ -156,6 +193,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 +220,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)
# ── 匹配产品 ── # ── 匹配产品 ──
@ -371,12 +412,13 @@ class MomOutboundPayload(BaseModel):
outbound_time: datetime | None = None # 出库时间 outbound_time: datetime | None = None # 出库时间
# ↓ 2026-09 新增:部门路由键,本实例只认 "LICA"(见 _belongs_to_this_org) # ↓ 2026-09 新增:部门路由键,本实例只认 "LICA"(见 _belongs_to_this_org)
company_name: str | None = None # IRIS / LICA company_name: str | None = None # IRIS / LICA
outbound_type: str | None = None # SALES / PRODUCTION(MOM 一直在发,此前被丢弃) outbound_type: str | None = None # SALES / USE / PRODUCTION(MOM 一直在发,此前被丢弃)
@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 +437,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_(

View File

@ -56,7 +56,19 @@ _TRACKED_READ_PREFIXES = (
) )
# 永久忽略的路径前缀 # 永久忽略的路径前缀
_IGNORED_PREFIXES = ("/health", "/docs", "/redoc", "/openapi.json") #
# 两类内容:
# 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] = { _PATH_MODULE: dict[str, str] = {