Compare commits
3 Commits
24a15126d0
...
main
| Author | SHA1 | Date | |
|---|---|---|---|
| 700486d5d5 | |||
| 9778b8e4b2 | |||
| 754c25b477 |
13
AGENTS.md
13
AGENTS.md
@ -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` 可用):
|
||||||
|
|||||||
@ -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_(
|
||||||
|
|||||||
@ -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] = {
|
||||||
|
|||||||
Reference in New Issue
Block a user