GaoYiYuan_0626
90d0660b44
- 仓储：新增 list_pending_alerts_all / update_alert_escalation（统一读改写 payload， 升级标记独占写入，status/handler_* 列不碰；_parse_alert 补 created_at/handled_at 字符串→datetime 解析，兼容 sqlite 原生 DDL） - escalation_service.scan_and_escalate：扫描判级（普通 4h/24h、AML 1h/4h 短通道）、 幂等闸门（仅升不降）、先持久化再推送、按 (customer_id,level) 降噪合并一次推送、 通知链累积（L1 含 risk_manager / L2 含 compliance）、逐单+任务级审计 - scripts/cron/escalation_scan.py：15min 定时扫描壳（sys.path 引导 + new_trace + JSON 摘要） - 对话线 query_overdue_alerts Tool + 注册表 + 意图词（置于 alert_query 之前）+ summarize 分支 - C5 前置：seed STAFF-31001/31002(risk_manager) + deps 矩阵放行 + chat.py 显式 deny + risk.py 台账全量只读分支；JWT 手册 §5.3/§5.4/§6.1 增补 risk_manager - 测试：conftest 回拨 fixture + test_escalation_service(9) + risk_api/manager(5) + chat deny(1) + chat_tools overdue(2)；全量 470 绿（453+17）
433 lines
17 KiB
Python
433 lines
17 KiB
Python
"""risk API 权限矩阵测试（B6 · A-7 处置/仅 aml、A-9 越权 403+audit、debug 头边界）。
|
||
|
||
TestClient 独立挂 router（main 挂载归 B7）；仓储经 monkeypatch 注入 sqlite。
|
||
"""
|
||
|
||
from datetime import datetime, timedelta
|
||
|
||
import pytest
|
||
from fastapi import FastAPI
|
||
from fastapi.testclient import TestClient
|
||
from sqlalchemy import text
|
||
|
||
from _ddl import create_sqlite_engine, seed_suitability_matrix
|
||
from app.api import risk as risk_api
|
||
from app.api.deps import assert_customer_access
|
||
from app.api.risk import router as risk_router
|
||
from app.repository.core_ro import CoreReadOnlyRepository
|
||
from app.repository.risk_repository import RiskRepository
|
||
from app.service.risk import redis_gateway
|
||
from app.utils.exceptions import PermissionDenied
|
||
from app.utils.response import register_error_handlers
|
||
|
||
|
||
class FakePublisher:
|
||
"""aml/scan 推送隔离（复审 P2：测试不得依赖本机 Redis 状态）。"""
|
||
|
||
def __init__(self):
|
||
self.messages = []
|
||
self.deletes = []
|
||
|
||
def publish(self, channel, payload):
|
||
self.messages.append((channel, payload))
|
||
|
||
def delete(self, *keys):
|
||
self.deletes.append(keys)
|
||
|
||
|
||
def _alert(alert_id, customer, atype, status="pending_review", score=70):
|
||
return {
|
||
"alert_id": alert_id,
|
||
"trace_id": f"trc-{alert_id.lower()}",
|
||
"customer_id": customer,
|
||
"trade_id": None,
|
||
"alert_type": atype,
|
||
"triggered_rules": ["RISK-001"],
|
||
"risk_score": score,
|
||
"status": status,
|
||
"payload": {"events": [], "customer_context": {}},
|
||
}
|
||
|
||
|
||
@pytest.fixture()
|
||
def env():
|
||
engine = create_sqlite_engine() # DDL 单一事实源（B4 评审 P3-12）
|
||
seed_suitability_matrix(engine) # AL-05：check_suitability 以矩阵表为 L0 权威
|
||
with engine.begin() as conn:
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer (customer_id, display_name, age, is_active) VALUES"
|
||
" ('CUST-1001', '客户·王**', 28, 1), ('CUST-3001', '客户·孙**', 45, 1)"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at, expires_at) VALUES"
|
||
" ('CUST-1001', 'C1', :t, :exp), ('CUST-3001', 'C3', :t, :exp)"
|
||
),
|
||
{"t": datetime.now() - timedelta(days=30), "exp": datetime.now() + timedelta(days=300)},
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_customer_advisor VALUES ('ADV-01', 'CUST-3001', 'active')"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO core_product (product_id, product_name, min_risk_code, product_type) VALUES"
|
||
" ('PROD-161725', '科技成长主题', 'R4', 'mixed'),"
|
||
" ('PROD-510300', '沪深300指数', 'R3', 'index')"
|
||
)
|
||
)
|
||
conn.execute(
|
||
text(
|
||
"INSERT INTO risk_aml_list (list_id, list_type, full_name, match_threshold,"
|
||
" source, list_version, is_active) VALUES"
|
||
" ('SAN-1', 'sanction', '客户·孙**', 0.85, 'mock', 'v1', 1)"
|
||
)
|
||
)
|
||
repo = RiskRepository(engine=engine)
|
||
repo.insert_alert(_alert("ALT-E1", "CUST-3001", "large_amount"))
|
||
repo.insert_alert(_alert("ALT-A1", "CUST-1001", "aml", score=95))
|
||
repo.insert_alert(_alert("ALT-D1", "CUST-3001", "freq_trade", status="confirmed_normal", score=50))
|
||
yield repo, engine
|
||
engine.dispose()
|
||
|
||
|
||
@pytest.fixture()
|
||
def client(env, monkeypatch):
|
||
repo, engine = env
|
||
monkeypatch.setattr(risk_api, "_repo", lambda: repo)
|
||
monkeypatch.setattr(risk_api, "CoreReadOnlyRepository", lambda: CoreReadOnlyRepository(engine=engine))
|
||
# 401/越权审计经 deps 内 RiskRepository 兜底构造,统一注入 sqlite(B6 评审 P3-4)
|
||
from app.api import deps as deps_mod
|
||
|
||
monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo)
|
||
# aml 命中广播 Pub/Sub + L3 缓存 DEL:注入 fake,不触真 Redis(复审 P2)
|
||
monkeypatch.setattr(redis_gateway, "_gateway", FakePublisher())
|
||
app = FastAPI()
|
||
app.include_router(risk_router)
|
||
register_error_handlers(app) # 统一错误体(手册 §10,与 main 同一 handler 集)
|
||
with TestClient(app) as c:
|
||
yield c
|
||
|
||
|
||
def _h(role="", actor=""):
|
||
return {"X-Debug-Role": role, "X-Debug-Actor": actor} if role else {}
|
||
|
||
|
||
OFFICER = _h("risk_officer", "STAFF-90001")
|
||
COMPLIANCE = _h("compliance", "STAFF-40001")
|
||
MANAGER = _h("risk_manager", "STAFF-31001") # C5 前置（PRD 4A.1）：HTTP 台账全量只读，无处置权
|
||
CUST_1001 = _h("customer", "CUST-1001")
|
||
ADV_01 = _h("advisor", "ADV-01")
|
||
|
||
|
||
def _counts(engine, table, where="1=1"):
|
||
with engine.connect() as conn:
|
||
return conn.execute(text(f"SELECT COUNT(*) FROM {table} WHERE {where}")).scalar_one()
|
||
|
||
|
||
# ---------- GET /alerts ----------
|
||
|
||
|
||
def test_officer_sees_all_compliance_forced_aml(client):
|
||
r = client.get("/api/risk/alerts", headers=OFFICER)
|
||
assert r.status_code == 200
|
||
assert r.json()["total"] == 3 # 全量（含已处置）
|
||
r = client.get("/api/risk/alerts", headers=COMPLIANCE)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["total"] == 1 and body["items"][0]["alert_type"] == "aml" # A-7 强制过滤
|
||
r = client.get("/api/risk/alerts", params={"alert_type": "large_amount"}, headers=COMPLIANCE)
|
||
assert r.json()["items"][0]["alert_type"] == "aml" # 传入类型被覆盖
|
||
|
||
|
||
def test_other_roles_cannot_list_alerts(client):
|
||
for headers in (CUST_1001, ADV_01):
|
||
r = client.get("/api/risk/alerts", headers=headers)
|
||
assert r.status_code == 403
|
||
|
||
|
||
def test_manager_sees_all_alerts_readonly(client):
|
||
"""C5 前置（PRD 4A.1）：risk_manager 经矩阵放行 HTTP 通道，GET 全量 200（同 officer 查询路径）。"""
|
||
r = client.get("/api/risk/alerts", headers=MANAGER)
|
||
assert r.status_code == 200
|
||
assert r.json()["total"] == 3 # 全量（含已处置 + aml），无处置过滤
|
||
|
||
|
||
def test_manager_handle_403(client, env):
|
||
"""manager 无 risk_officer 角色 → handle 端点白名单自然 403（零代码改动）。"""
|
||
repo, _ = env
|
||
r = client.post(
|
||
"/api/risk/alerts/ALT-E1/handle",
|
||
json={"handler_result": "confirmed_suspicious"},
|
||
headers=MANAGER,
|
||
)
|
||
assert r.status_code == 403
|
||
assert repo.get_alert("ALT-E1")["status"] == "pending_review" # 未改动
|
||
|
||
|
||
def test_manager_suitability_check_403(client):
|
||
"""评审 P2-5 回归：新增端点仅依赖矩阵放行时，manager 仍须被 G-01 挡在客户业务数据外。"""
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-3001", "product_id": "PROD-510300"},
|
||
headers=MANAGER,
|
||
)
|
||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SCOPE"
|
||
|
||
|
||
def test_manager_aml_scan_403(client):
|
||
"""评审 P2-5 回归：aml/scan 仅 risk_officer，manager 触碰即 403。"""
|
||
assert client.post("/api/risk/aml/scan", headers=MANAGER).status_code == 403
|
||
|
||
|
||
def test_alerts_date_filter_and_pagination(client):
|
||
"""复审 P3:start_date/end_date 过滤与分页边界回归保护。
|
||
|
||
DDL 收敛后 created_at 为 localtime（_ddl.py，B5 评审 P3-4），窗口用本地时间构造。
|
||
"""
|
||
now = datetime.now()
|
||
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
|
||
|
||
|
||
# ---------- POST /alerts/{id}/handle ----------
|
||
|
||
|
||
def test_officer_handle_success_state_machine_and_audit(client, env):
|
||
repo, engine = env
|
||
r = client.post(
|
||
"/api/risk/alerts/ALT-E1/handle",
|
||
json={"handler_result": "confirmed_suspicious", "handler_comment": "确认可疑"},
|
||
headers=OFFICER,
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["status"] == "confirmed_suspicious" and body["handler_id"] == "STAFF-90001"
|
||
assert _counts(engine, "audit_log", "event_type='alert_handle' AND decision='alert_handled'") == 1
|
||
# 状态机：已处置单禁止跳改
|
||
r2 = client.post(
|
||
"/api/risk/alerts/ALT-E1/handle",
|
||
json={"handler_result": "confirmed_normal"},
|
||
headers=OFFICER,
|
||
)
|
||
assert r2.status_code == 409
|
||
|
||
|
||
def test_compliance_handle_403(client, env):
|
||
repo, _ = env
|
||
r = client.post(
|
||
"/api/risk/alerts/ALT-E1/handle",
|
||
json={"handler_result": "confirmed_suspicious"},
|
||
headers=COMPLIANCE,
|
||
)
|
||
assert r.status_code == 403 # A-7
|
||
assert repo.get_alert("ALT-E1")["status"] == "pending_review"
|
||
|
||
|
||
def test_handle_invalid_result_422(client):
|
||
r = client.post(
|
||
"/api/risk/alerts/ALT-E1/handle",
|
||
json={"handler_result": "auto_frozen"}, # 非法枚举（依赖层校验）
|
||
headers=OFFICER,
|
||
)
|
||
assert r.status_code == 422
|
||
|
||
|
||
def test_handle_missing_alert_404(client):
|
||
r = client.post(
|
||
"/api/risk/alerts/ALT-XXXX/handle",
|
||
json={"handler_result": "confirmed_normal"},
|
||
headers=OFFICER,
|
||
)
|
||
assert r.status_code == 404
|
||
|
||
|
||
# ---------- POST /suitability/check（G-01 + A-9） ----------
|
||
|
||
|
||
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"},
|
||
headers=CUST_1001,
|
||
)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
# AL-06 main 契约：机器码断言改 block_response_code（SUIT-001 退役）
|
||
assert body["blocked"] is True and body["block_response_code"] == "SUIT_RISK_MISMATCH" # C1+R4
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-3001", "product_id": "PROD-510300"},
|
||
headers=OFFICER,
|
||
)
|
||
assert r.status_code == 200 and r.json()["is_matched"] is True
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-3001", "product_id": "PROD-510300"},
|
||
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):
|
||
repo, engine = env
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-3001", "product_id": "PROD-510300"},
|
||
headers=CUST_1001, # 查他人
|
||
)
|
||
assert r.status_code == 403
|
||
assert r.json()["error_code"] == "AUTH_403_NOT_OWNER"
|
||
assert _counts(engine, "audit_log", "event_type='authz' AND decision='forbidden'") == 1
|
||
|
||
|
||
def test_a9_advisor_not_assigned_403_with_audit(client, env):
|
||
repo, engine = env
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-1001", "product_id": "PROD-161725"},
|
||
headers=ADV_01, # 非名下
|
||
)
|
||
assert r.status_code == 403
|
||
assert r.json()["error_code"] == "AUTH_403_NOT_ASSIGNED"
|
||
assert _counts(engine, "audit_log", "event_type='authz' AND decision='forbidden'") == 1
|
||
|
||
|
||
def test_compliance_suitability_check_403(client):
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-3001", "product_id": "PROD-510300"},
|
||
headers=COMPLIANCE, # 审计角色不在客户业务数据白名单
|
||
)
|
||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_SCOPE"
|
||
|
||
|
||
# ---------- POST /aml/scan ----------
|
||
|
||
|
||
def test_aml_scan_officer_only_with_audit(client, env):
|
||
repo, engine = env
|
||
r = client.post("/api/risk/aml/scan", headers=OFFICER)
|
||
assert r.status_code == 200
|
||
body = r.json()
|
||
assert body["scanned"] == 2 and body["hit_customers"] == 1 # CUST-3001 命中
|
||
assert len(body["alerts"]) == 1
|
||
assert body["skipped_existing"] == [] # 首扫无既有单
|
||
assert body["disclaimer"] == risk_api.ALERT_DISCLAIMER # B9b 核查单①
|
||
l3 = repo.get_l3("CUST-3001")
|
||
assert l3["monitor_tier"] == "high" # scan 命中标记 L3
|
||
# L3 写侧缓存 DEL 钩子（B7 挂账②）：命中客户 upsert 后失效读缓存
|
||
fake = redis_gateway.get_gateway()
|
||
assert ("profile:l3:CUST-3001",) in fake.deletes
|
||
assert _counts(engine, "audit_log", "event_type='aml_scan' AND decision='scan_completed'") == 1
|
||
for headers in (COMPLIANCE, CUST_1001):
|
||
assert client.post("/api/risk/aml/scan", headers=headers).status_code == 403
|
||
|
||
|
||
def test_aml_scan_idempotent_same_day(client, env):
|
||
"""B9b 核查单②（B6 评审 P3-6）：同日重扫不重复出单，重复点击防护。"""
|
||
repo, engine = env
|
||
first = client.post("/api/risk/aml/scan", headers=OFFICER).json()
|
||
second = client.post("/api/risk/aml/scan", headers=OFFICER).json()
|
||
assert second["alerts"] == []
|
||
assert second["skipped_existing"] == first["alerts"]
|
||
# aml 单总数不变（env 预置 ALT-A1 一张 + scan 新出一张；重扫零新增）
|
||
assert _counts(engine, "risk_alert", "alert_type='aml'") == 2
|
||
|
||
|
||
def test_alert_disclaimer_on_alert_apis(client):
|
||
"""B9b 核查单①（B6 复审 P3-7）：预警类 API 响应体固定 disclaimer（规则表 §5）。"""
|
||
r = client.get("/api/risk/alerts", headers=OFFICER)
|
||
body = r.json()
|
||
assert r.status_code == 200 and body["disclaimer"] == risk_api.ALERT_DISCLAIMER
|
||
pending = next(i for i in body["items"] if i["status"] == "pending_review")
|
||
r = client.post(
|
||
f"/api/risk/alerts/{pending['alert_id']}/handle",
|
||
json={"handler_result": "confirmed_normal"},
|
||
headers=OFFICER,
|
||
)
|
||
assert r.status_code == 200 and r.json()["disclaimer"] == risk_api.ALERT_DISCLAIMER
|
||
|
||
|
||
# ---------- 非 dev 环境拒绝 debug 鉴权 ----------
|
||
|
||
|
||
def test_non_dev_rejects_debug_auth(client, monkeypatch):
|
||
"""T-01 后：非 dev 无 Bearer 一律 401 留痕（debug 头通道不存在于生产）。"""
|
||
from app.config.settings import settings
|
||
|
||
monkeypatch.setattr(settings, "app_env", "production")
|
||
r = client.get("/api/risk/alerts", headers=OFFICER)
|
||
assert r.status_code == 401
|
||
assert r.json()["error_code"] == "AUTH_401_MISSING_BEARER"
|
||
|
||
|
||
# ---------- deps 单元：compliance 不放行客户业务数据 ----------
|
||
|
||
|
||
def test_assert_customer_access_scope_denial(env):
|
||
repo, _ = env
|
||
from app.api.deps import AuthContext
|
||
|
||
auth = AuthContext(actor_id="STAFF-40001", roles=["compliance"])
|
||
with pytest.raises(PermissionDenied) as ei:
|
||
assert_customer_access(
|
||
auth, "CUST-1001", core_ro=CoreReadOnlyRepository(engine=env[1]), risk_repo=repo
|
||
)
|
||
assert ei.value.code == "AUTH_403_SCOPE"
|
||
|
||
|
||
def test_authz_denials_are_audited(client, env):
|
||
"""B6 评审 P1-1:403/401 全部留痕(AUTH_403_ROLE / AUTH_401_MISSING_DEBUG_HEADERS)。"""
|
||
repo, engine = env
|
||
assert client.get("/api/risk/alerts").status_code == 401 # 无 debug 头
|
||
assert client.post("/api/risk/aml/scan", headers=ADV_01).status_code == 403
|
||
assert client.post(
|
||
"/api/risk/alerts/ALT-E1/handle",
|
||
json={"handler_result": "confirmed_normal"},
|
||
headers=COMPLIANCE,
|
||
).status_code == 403
|
||
assert _counts(engine, "audit_log", "decision='unauthenticated' AND input_summary LIKE '%AUTH_401%'") == 1
|
||
assert _counts(engine, "audit_log", "decision='forbidden' AND input_summary LIKE '%AUTH_403_ROLE%'") == 2
|
||
|
||
|
||
def test_multi_role_combinations_fixed_behavior(client):
|
||
"""B6 评审 P3-1:多角色口径固化——并集权限、归属 fail-closed。"""
|
||
# customer+compliance:借 compliance 角色看 aml 台账(审计类跨客户,固化允许)
|
||
r = client.get("/api/risk/alerts", headers=_h("customer,compliance", "CUST-1001"))
|
||
assert r.status_code == 200 and r.json()["items"][0]["alert_type"] == "aml"
|
||
# customer+advisor 查非本人客户:customer 分支 deny 即终止(最窄范围 fail-closed)
|
||
r = client.post(
|
||
"/api/risk/suitability/check",
|
||
json={"customer_id": "CUST-1001", "product_id": "PROD-161725"},
|
||
headers=_h("customer,advisor", "CUST-3001"),
|
||
)
|
||
assert r.status_code == 403 and r.json()["error_code"] == "AUTH_403_NOT_OWNER"
|