From a78e9d4d3184dd10cf2c36167e5babfccc26dbb8 Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 19:21:15 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20B6=20=E5=A4=8D=E5=AE=A1=E6=94=B6?= =?UTF-8?q?=E5=8F=A3=E2=80=94=E2=80=94=E6=B5=8B=E8=AF=95=20Redis=20?= =?UTF-8?q?=E9=9A=94=E7=A6=BB/=E7=BD=91=E5=85=B3=E5=AE=A1=E8=AE=A1?= =?UTF-8?q?=E5=8F=A3=E5=BE=84=20platform/=E6=AD=BB=E5=B8=B8=E9=87=8F?= =?UTF-8?q?=E6=B8=85=E7=90=86/=E5=9B=9E=E5=BD=92=E6=96=AD=E8=A8=80?= =?UTF-8?q?=E8=A1=A5=E5=BC=BA?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - P2: test_risk_api env 注入 FakePublisher,aml 推送不触真 Redis - P3: deny/_authz_audit 增 agent_type 参数,simulate 越权审计传 platform (与网关放行审计同口径);删 HANDLE_RESULTS 死常量(Literal 单处定义) - P3: 补 start_date/end_date 过滤+分页 422 边界+suitability api 层审计断言; 补 app/service/risk/__init__.py - 挂账: B7 行新增④项(启动期拒绝/引擎工厂覆盖三实例化点/handle 原子性/ input_guard_log);B9b 行核查单(disclaimer/scan 幂等/Swagger 手测/analyst); TODO.md T-30/T-31/T-32 进度同步 --- app/api/deps.py | 12 +++++-- app/api/risk.py | 2 -- app/api/simulate.py | 1 + app/service/risk/__init__.py | 1 + docs/memory/TODO.md | 6 ++-- docs/项目框架设计/开发计划-风控模块.md | 4 +-- tests/test_risk_api.py | 44 +++++++++++++++++++++++++- tests/test_trade_gateway.py | 3 +- 8 files changed, 61 insertions(+), 12 deletions(-) create mode 100644 app/service/risk/__init__.py diff --git a/app/api/deps.py b/app/api/deps.py index fa22f04..92caa1d 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -54,13 +54,18 @@ def _authz_audit( auth: AuthContext | None, customer_id: str | None, code: str, + agent_type: str = "risk", ) -> None: - """鉴权失败审计（event_type='authz'；手册 P-05，B6 评审 P1-1）。""" + """鉴权失败审计（event_type='authz'；手册 P-05，B6 评审 P1-1）。 + + agent_type 按路由归属传入（网关路由传 'platform'，复审 P3：与放行审计 + 同口径，避免按模块检索审计时漏网关越权事件）。 + """ risk_repo.insert_audit_log( { "trace_id": current_trace() or new_trace(), "event_type": "authz", - "agent_type": "risk", + "agent_type": agent_type, "actor_id": auth.actor_id if auth else "anonymous", "customer_id": customer_id, "rule_id": None, @@ -80,9 +85,10 @@ def deny( risk_repo: RiskRepository, customer_id: str | None = None, message: str | None = None, + agent_type: str = "risk", ) -> None: """越权出口：审计 + 403（全部 403 必经此函数，保证留痕与错误码）。""" - _authz_audit(risk_repo, auth, customer_id, code) + _authz_audit(risk_repo, auth, customer_id, code, agent_type) raise PermissionDenied(code, message or f"forbidden: {code}") diff --git a/app/api/risk.py b/app/api/risk.py index 4a0ac3b..0b8ec77 100644 --- a/app/api/risk.py +++ b/app/api/risk.py @@ -32,8 +32,6 @@ from app.utils.trace import current_trace, new_trace router = APIRouter(prefix="/api/risk", tags=["risk"]) -HANDLE_RESULTS = ("confirmed_normal", "confirmed_suspicious", "reported") - def _repo() -> RiskRepository: """仓储入口（测试 monkeypatch 点）。""" diff --git a/app/api/simulate.py b/app/api/simulate.py index dbec936..28f66bf 100644 --- a/app/api/simulate.py +++ b/app/api/simulate.py @@ -42,6 +42,7 @@ def submit_trade_api(req: TradeRequest, auth: AuthContext = Depends(get_auth_con deny( auth, "AUTH_403_ROLE", _repo(), customer_id=req.customer_id, message="risk_demo or owner customer only", + agent_type="platform", # 网关越权与放行审计同口径（复审 P3） ) try: return submit_trade(req.model_dump()) diff --git a/app/service/risk/__init__.py b/app/service/risk/__init__.py new file mode 100644 index 0000000..51d7440 --- /dev/null +++ b/app/service/risk/__init__.py @@ -0,0 +1 @@ +"""风控服务子包（规则纯函数 / 预警聚合 / AML / L3 画像 / 引擎编排）。""" diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md index 287b131..fc26ba1 100644 --- a/docs/memory/TODO.md +++ b/docs/memory/TODO.md @@ -20,9 +20,9 @@ ### 风控模块（PRD v1.0 已冻结 · `docs/PRD/PRD-风控监测Agent.md`，事件驱动线不依赖 T-07 可先行） -- [ ] T-30 风控事件线：`app/gateway/` 交易网关 + `service/risk/` 规则引擎（RISK-001~005）+ 预警单聚合 + L3 最小写入 + `risk:pub:alert` 推送 + AML（含 `risk_aml_list` 种子）+ 演示数据脚本验收 A-1~A-5/A-9 —— **进度（2026-09-06）：B1 规则纯函数 / B2 预警服务 / B3 L3 写入（risk_score 一期不写）/ B4 AML+引擎编排 / B5 交易网关（convert 400、阻断不落 trade、引擎异常审计降级）均已完成并经独立 AI 评审闭环；剩 B6 鉴权+API、B7 main 集成、B8 conftest+集成测试、B9a 脚本、B9b 演示走查** -- [ ] T-31 `service/suitability.py` 公共校验（SUIT-001~008）+ `POST /api/risk/suitability/check` + 单测验收 A-8 —— **代码已完成（A1~A4 · 77 测试绿）；MySQL 手工 SQL 对照挂账至 B9b 执行（阶段 A 评审 P2-8）** -- [ ] T-32 预警台账与人工处置 API（`GET /alerts`、`POST /handle`，risk_officer/compliance 权限）+ 对话线（依赖 T-01/T-03/T-07）验收 A-6/A-7 +- [ ] T-30 风控事件线：`app/gateway/` 交易网关 + `service/risk/` 规则引擎（RISK-001~005）+ 预警单聚合 + L3 最小写入 + `risk:pub:alert` 推送 + AML（含 `risk_aml_list` 种子）+ 演示数据脚本验收 A-1~A-5/A-9 —— **进度（2026-09-06）：B1 规则纯函数 / B2 预警服务 / B3 L3 写入（risk_score 一期不写）/ B4 AML+引擎编排 / B5 交易网关 / B6 鉴权+4 API+处置编排（含独立 AI 复审通过：P0/P1=0，P2 测试隔离已修，agent_type 口径等 P3 已修或挂账 B7/B9b，173 测试绿）均已完成并经独立 AI 评审闭环；剩 B7 main 集成、B8 conftest+集成测试、B9a 脚本、B9b 演示走查** +- [ ] T-31 `service/suitability.py` 公共校验（SUIT-001~008）+ `POST /api/risk/suitability/check` + 单测验收 A-8 —— **代码已完成（A1~A4 · 77 测试绿；B6 起该 API 带鉴权+直调审计）；MySQL 手工 SQL 对照挂账至 B9b 执行（阶段 A 评审 P2-8）** +- [ ] T-32 预警台账与人工处置 API（`GET /alerts`、`POST /handle`，risk_officer/compliance 权限）+ 对话线（依赖 T-01/T-03/T-07）验收 A-6/A-7 —— **API 部分已由 B6 覆盖并复审通过（A-7/A-9 用例绿）；对话线仍依赖 T-01/T-03/T-07** ## 已完成 diff --git a/docs/项目框架设计/开发计划-风控模块.md b/docs/项目框架设计/开发计划-风控模块.md index 4b260e2..17bea34 100644 --- a/docs/项目框架设计/开发计划-风控模块.md +++ b/docs/项目框架设计/开发计划-风控模块.md @@ -26,10 +26,10 @@ | B4 | `aml_service.py`（归一化+相似度匹配、scan_all）+ `engine.py`（process_trade_event 组装 + 预留客户事件钩子）+ **`scoring.py` 占位签名（FR-7 预留）** | 引擎完整 | 单测：AML 阈值边界；引擎集成冒烟 | B1、B2、B3。**B4 评审修复（2026-09-06）：FR-4 payload 客户上下文（L0+近 30 天统计）补组装；core_ro 增 list_trades_range（升序）/list_active_customers；种子名单 3 条改名维持唯一演示命中；core_cash_flow 上下文一期未接（PRD「如有」），B8 前评估；真实名单数据接入时 payload/审计出口接 utils/desensitize（B9b 前核查项）；复审观察项：30 天窗口口径（现 31 自然日）/申赎混合求和方向，R-05 接入前统一** | | B5 | `app/gateway/`（trade_gateway + gateway_repository 仅 INSERT core_trade）+ `api/simulate.py` 薄路由 | 网关 | 集成：convert 400、阻断不落 trade | A4、B4 | | B6 | **`app/api/deps.py`：`AuthContext`（actor_id/roles/customer_id，字段按 JWT 手册冻结）+ `get_auth_context()` 工厂**——dev 模式从 `X-Debug-Role`/`X-Debug-Actor` 请求头构造、`app_env != development` 启动时检测 debug 头直接拒绝；T-01 就绪后仅替换工厂内部为 JWT 解析，签名不变。另：`api/risk.py` 4 个 API（GET alerts / POST handle / POST suitability/check / POST aml/scan）+ 归属校验（含 compliance 强制 aml 过滤）。**备注：依赖层须校验 handler_result 枚举（repo 不校验）；本阶段顺手统一 `NotFoundError` 异常（utils/exceptions.py 现为占位）**（评审 P2-7①②） | 鉴权依赖 + 4 个 API | Swagger 手测 + **权限矩阵（按 debug 头切换角色/身份执行 A-7/A-9 用例）** | A4、B2、**B4**（aml/scan 依赖 scan_all） | -| B7 | `main.py` 集成：路由挂载 + lifespan（双 Engine 单例注入 + Redis 单例 + trace 中间件）。**备注：顺手提取 `utils/db.py` 引擎工厂收敛 core_ro/risk_repository 双份 _default_engine**（评审 P2-6）；**B3 挂账（B3 评审 P2-5）：① `_run_locked` 锁原语公共化（alert_service/profile_l3 现复用私有实现）② L3 写侧 Redis 缓存 DEL 钩子（PRD §5.1 `profile:l3:{customer_id}` 更新时 DEL，`profile_l3.upsert_profile_l3` 已留痕）③ 删除 core_ro.list_trades 死代码（B4 改用 list_trades_range 后无调用方，复审 N3）④ 统一响应外壳落地（utils/response.py 现占位，simulate/risk 路由届时一并包裹，B5 评审 P2-2）** | 可运行应用 | `uvicorn` 启动 + `/health` + 全路由可达 | B5、B6 | +| B7 | `main.py` 集成：路由挂载 + lifespan（双 Engine 单例注入 + Redis 单例 + trace 中间件）。**备注：顺手提取 `utils/db.py` 引擎工厂收敛 core_ro/risk_repository 双份 _default_engine**（评审 P2-6）；**B3 挂账（B3 评审 P2-5）：① `_run_locked` 锁原语公共化（alert_service/profile_l3 现复用私有实现）② L3 写侧 Redis 缓存 DEL 钩子（PRD §5.1 `profile:l3:{customer_id}` 更新时 DEL，`profile_l3.upsert_profile_l3` 已留痕）③ 删除 core_ro.list_trades 死代码（B4 改用 list_trades_range 后无调用方，复审 N3）④ 统一响应外壳落地（utils/response.py 现占位，simulate/risk 路由届时一并包裹，B5 评审 P2-2；错误体对齐手册 §10 error_code/message/trace_id，B6 评审 P3-2）**；**B6 挂账（B6 复审）：⑤ lifespan 启动期检测 `app_env != development` + debug 头依赖直接拒绝启动（现为请求时 RuntimeError，B6 评审遗漏①）⑥ 引擎工厂须覆盖 deps/simulate/risk 三处每请求 `RiskRepository()`/`CoreReadOnlyRepository()` 实例化点并 dispose（现复用 _default_engine 不释放，B6 复审 P3）⑦ `handle_alert` 的 update_alert_status 与 insert_audit_log 两事务非原子——统一事务或补偿记录（B6 复审 P3）⑧ input_guard_log 双写缺口（手册 P-05 要求 audit_log+input_guard_log，表归 T-03 底座，B7 接 T-02 审计中间件时统一补）** | 可运行应用 | `uvicorn` 启动 + `/health` + 全路由可达 | B5、B6 | | B8 | **`tests/conftest.py`**：a) session fixture 启动校验演示数据就位（CUST-4001 测评 <365 天、risk_aml_list ≥8），缺失则中止并提示先跑 FLOW §0 ③④；b) fixture 幂等代跑 `prepare_risk_demo.sql`；c) teardown 按 `TRD-TEST-` 清 core_trade + 关联 risk_alert/risk_suitability_log/audit_log + 还原 L3 行。**顺手集中 sqlite 测试 DDL 为单一事实源（B4 评审 P3-12，各测试文件手写 DDL 收敛；CURRENT_TIMESTAMP 改 localtime 或 fixture 固定时间，防 UTC/本地日界错位——B5 评审 P3-4）**；集成测试交易统一走 `trade_id_factory` 注入 `TRD-TEST-` 前缀（trade_gateway 已留参数，B5 评审 P3-2）。集成测试：A-1~A-5、A-7（状态机/compliance 403/GET 强制 aml）、A-9 越权、**trace 一致性断言**；补 platform 审计 input_summary 的 JSON 解析断言（含引擎输出/阻断 reasons，B5 复审 L1） | 测试套件 + fixture | `pytest` 全绿 | B7 | | B9a | 演示/运维脚本开发：`scripts/demo/subscribe_alerts.py`（订阅演示）+ `scripts/demo/rebuild_alerts.py`（按 trade_id 幂等重放补偿） | 2 个脚本 | 手工执行验证 | B2、B4（可与 B5~B8 并行） | -| B9b | 演示链路走查：`reset.ps1` → `prepare_risk_demo.sql` → agent 库建表 → `seed-aml-list.sql` → Swagger 逐条过 **A-1~A-5、A-7~A-9（A-6 归 M3）** | 演示 SOP | 按 PRD §8 验收表逐条打勾 | B8、B9a | +| B9b | 演示链路走查：`reset.ps1` → `prepare_risk_demo.sql` → agent 库建表 → `seed-aml-list.sql` → Swagger 逐条过 **A-1~A-5、A-7~A-9（A-6 归 M3）**。**B6 挂账核查单：① 预警类 API 响应体含固定 disclaimer「本预警由系统自动生成，最终判定需经风控专员人工审核」（PRD §6/规则表 §5，B6 复审 P3-7）② aml/scan 幂等防护（重复扫描同命中客户重复出单，演示点击即复现，B6 评审 P3-6）③ B6 时代码以 TestClient 独立挂 router 等价验证，本次补一次真 Swagger 手测（B6 复审遗漏⑤）④ analyst 台账只读权限扩展待 Wave 3 分析 Agent 接入时定权限矩阵（手册 §5.3 有 risk:alert:read，现 fail-closed 拒绝，B6 复审观察③）** | 演示 SOP | 按 PRD §8 验收表逐条打勾 | B8、B9a | ## 阶段 C · 对话线（依赖 Wave 0 的 T-01 JWT / T-03 输入防护 / T-07 LangGraph） diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py index d358a74..eab4ff1 100644 --- a/tests/test_risk_api.py +++ b/tests/test_risk_api.py @@ -20,6 +20,16 @@ from app.service.risk import alert_service from app.utils.exceptions import PermissionDenied +class FakePublisher: + """aml/scan 推送隔离（复审 P2：测试不得依赖本机 Redis 状态）。""" + + def __init__(self): + self.messages = [] + + def publish(self, channel, payload): + self.messages.append((channel, payload)) + + def _alert(alert_id, customer, atype, status="pending_review", score=70): return { "alert_id": alert_id, @@ -130,6 +140,8 @@ def client(env, monkeypatch): from app.api import deps as deps_mod monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) + # aml 命中会广播 Pub/Sub:注入 fake,不触真 Redis(复审 P2) + monkeypatch.setattr(alert_service, "_publisher", FakePublisher()) app = FastAPI() app.include_router(risk_router) app.add_exception_handler(PermissionDenied, permission_denied_handler) @@ -173,6 +185,25 @@ def test_other_roles_cannot_list_alerts(client): assert r.status_code == 403 +def test_alerts_date_filter_and_pagination(client): + """复审 P3:start_date/end_date 过滤与分页边界回归保护。 + + sqlite CURRENT_TIMESTAMP 为 UTC,窗口用 utcnow 构造(localtime 收敛挂账 B8)。 + """ + now = datetime.utcnow() + r = client.get("/api/risk/alerts", params={"start_date": (now - timedelta(hours=1)).isoformat()}, headers=OFFICER) + assert r.json()["total"] == 3 # 窗口内全命中 + r = client.get("/api/risk/alerts", params={"start_date": (now + timedelta(hours=1)).isoformat()}, headers=OFFICER) + assert r.json()["total"] == 0 # 未来起点 → 空 + r = client.get("/api/risk/alerts", params={"end_date": (now - timedelta(hours=1)).isoformat()}, headers=OFFICER) + assert r.json()["total"] == 0 # 过去终点 → 空 + r = client.get("/api/risk/alerts", params={"page": 2, "page_size": 2}, headers=OFFICER) + body = r.json() + assert body["total"] == 3 and len(body["items"]) == 1 and body["page"] == 2 + assert client.get("/api/risk/alerts", params={"page": 0}, headers=OFFICER).status_code == 422 + assert client.get("/api/risk/alerts", params={"page_size": 101}, headers=OFFICER).status_code == 422 + + def test_missing_debug_headers_401(client): assert client.get("/api/risk/alerts").status_code == 401 @@ -232,7 +263,8 @@ def test_handle_missing_alert_404(client): # ---------- POST /suitability/check（G-01 + A-9） ---------- -def test_suitability_check_by_owner_officer_and_assigned_advisor(client): +def test_suitability_check_by_owner_officer_and_assigned_advisor(client, env): + repo, engine = env r = client.post( "/api/risk/suitability/check", json={"customer_id": "CUST-1001", "product_id": "PROD-161725"}, @@ -253,6 +285,16 @@ def test_suitability_check_by_owner_officer_and_assigned_advisor(client): headers=ADV_01, # ADV-01 名下 CUST-3001 ) assert r.status_code == 200 + # 直调路径每次校验补 api 层审计（评审 P2-3；复审 P3 回归保护） + assert _counts(engine, "audit_log", "event_type='suitability_check'") == 3 + assert ( + _counts( + engine, + "audit_log", + "event_type='suitability_check' AND decision='suitability_blocked'", + ) + == 1 + ) def test_a9_customer_other_403_with_audit(client, env): diff --git a/tests/test_trade_gateway.py b/tests/test_trade_gateway.py index 6160faa..14c407b 100644 --- a/tests/test_trade_gateway.py +++ b/tests/test_trade_gateway.py @@ -293,7 +293,8 @@ def test_api_customer_owner_allowed_other_denied(client, env): headers={"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-1001"}, ) assert r.status_code == 403 - assert _counts(engine, "audit_log", "event_type='authz' AND decision='forbidden'") == 1 + # 网关越权审计与放行同口径 agent_type='platform'(复审 P3) + assert _counts(engine, "audit_log", "event_type='authz' AND decision='forbidden' AND agent_type='platform'") == 1 def test_api_non_positive_amount_returns_422(client):