From e3fcf28b27f8f09468c90165f61d81af4f1c5bac Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 18:37:53 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E9=89=B4=E6=9D=83=E4=BE=9D=E8=B5=96(de?= =?UTF-8?q?bug=E5=A4=B4/=E5=BD=92=E5=B1=9E=E6=96=AD=E8=A8=80/=E8=B6=8A?= =?UTF-8?q?=E6=9D=83=E5=AE=A1=E8=AE=A1)=20+=20=E9=A3=8E=E6=8E=A74=20API(?= =?UTF-8?q?=E5=8F=B0=E8=B4=A6/=E5=A4=84=E7=BD=AE/=E6=A0=A1=E9=AA=8C/AML?= =?UTF-8?q?=E6=89=AB=E6=8F=8F,=20B6)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/deps.py | 94 ++++++++- app/api/risk.py | 131 ++++++++++++ app/service/risk/alert_service.py | 46 +++++ app/utils/exceptions.py | 25 ++- tests/test_risk_api.py | 326 ++++++++++++++++++++++++++++++ 5 files changed, 616 insertions(+), 6 deletions(-) create mode 100644 app/api/risk.py create mode 100644 tests/test_risk_api.py diff --git a/app/api/deps.py b/app/api/deps.py index 409f73e..7f40a70 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -1,14 +1,31 @@ """API 依赖：鉴权上下文（架构 §5.7 · 归属校验统一在此层）。 -AuthContext 模型在 A4 冻结（开发计划 v1.1）；get_auth_context 完整实现归 B6： -dev（app_env=development）从 X-Debug-Role / X-Debug-Actor 请求头构造， -非 dev 环境启动时检测 debug 依赖注册即拒绝；T-01 就绪后仅替换工厂内部为 JWT 解析。 +AuthContext 模型在 A4 冻结（开发计划 v1.1）；B6 落地 `get_auth_context()`： +dev（app_env=development）从 `X-Debug-Role` / `X-Debug-Actor` 请求头构造 +（customer 角色 customer_id=actor_id），非 dev 直接拒绝（启动期全局检查归 +B7 lifespan）；T-01 就绪后仅替换工厂内部为 JWT 解析，签名与调用方零改动。 + +归属断言 `assert_customer_access` 对齐 JWT 手册 §6.1/§6.2 与 PRD G-01： +customer 仅本人（AUTH_403_NOT_OWNER）、advisor 经 customer_advisor_rel +（AUTH_403_NOT_ASSIGNED）、risk_officer 全量；越权 403 + audit 留痕（A-9）。 """ from __future__ import annotations +from fastapi import HTTPException, Request +from fastapi.responses import JSONResponse from pydantic import BaseModel, Field +from app.config.settings import settings +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.risk_repository import RiskRepository +from app.utils.exceptions import PermissionDenied +from app.utils.trace import current_trace, new_trace + +STAFF_FULL_ACCESS_ROLES = ("risk_officer",) +DEBUG_ROLE_HEADER = "X-Debug-Role" +DEBUG_ACTOR_HEADER = "X-Debug-Actor" + class AuthContext(BaseModel): """统一鉴权上下文（全部 API 依赖层的产出；service 层签名接收此类型）。""" @@ -24,5 +41,72 @@ class AuthContext(BaseModel): return "customer" in self.roles -def get_auth_context() -> AuthContext: - raise NotImplementedError("implemented in B6 (X-Debug-* in dev / JWT in T-01)") +def get_auth_context(request: Request) -> AuthContext: + """鉴权工厂（B6）：dev 读 debug 头，非 dev 拒绝；T-01 后替换内部为 JWT 解析。""" + if settings.app_env != "development": + raise RuntimeError( + f"debug auth disabled outside development (app_env={settings.app_env})" + ) + roles = [r.strip() for r in request.headers.get(DEBUG_ROLE_HEADER, "").split(",") if r.strip()] + actor_id = request.headers.get(DEBUG_ACTOR_HEADER, "").strip() + if not roles or not actor_id: + raise HTTPException(status_code=401, detail="missing X-Debug-Role/X-Debug-Actor") + return AuthContext( + actor_id=actor_id, + roles=roles, + customer_id=actor_id if "customer" in roles else None, + ) + + +def permission_denied_handler(request: Request, exc: PermissionDenied) -> JSONResponse: + """FastAPI 异常 handler：PermissionDenied → 403 + 错误码（B7 注册进 main）。""" + return JSONResponse(status_code=403, content={"detail": str(exc), "code": exc.code}) + + +def _deny( + risk_repo: RiskRepository | None, auth: AuthContext, customer_id: str, code: str +) -> None: + """越权审计留痕（A-9：403 + audit）后抛 403。""" + if risk_repo is not None: + risk_repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "authz", + "agent_type": "risk", + "actor_id": auth.actor_id, + "customer_id": customer_id, + "rule_id": None, + "input_summary": {"roles": auth.roles, "code": code}, + "decision": "forbidden", + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + raise PermissionDenied(code, f"forbidden: {code} ({auth.actor_id} -> {customer_id})") + + +def assert_customer_access( + auth: AuthContext, + customer_id: str, + core_ro: CoreReadOnlyRepository | None = None, + risk_repo: RiskRepository | None = None, +) -> None: + """G-01 归属断言（customer/advisor/risk_officer；其他角色一律拒绝）。 + + customer 仅本人；advisor 需 customer_advisor_rel active；risk_officer 全量。 + compliance 不在客户业务数据访问白名单（仅审计类读，JWT 手册 §5.3）。 + """ + if auth.has_role(*STAFF_FULL_ACCESS_ROLES): + return + if "customer" in auth.roles: + if auth.customer_id == customer_id: + return + _deny(risk_repo, auth, customer_id, "AUTH_403_NOT_OWNER") + if "advisor" in auth.roles: + core = core_ro or CoreReadOnlyRepository() + if core.is_advisor_assigned(auth.actor_id, customer_id): + return + _deny(risk_repo, auth, customer_id, "AUTH_403_NOT_ASSIGNED") + _deny(risk_repo, auth, customer_id, "AUTH_403_SCOPE") diff --git a/app/api/risk.py b/app/api/risk.py new file mode 100644 index 0000000..9896f9e --- /dev/null +++ b/app/api/risk.py @@ -0,0 +1,131 @@ +"""风控 API（B6 · PRD FR-4 预警台账与人工处置 / FR-2 校验接口 / FR-5 手动扫描）。 + +鉴权：`Depends(get_auth_context)`（dev debug 头，T-01 后换 JWT，签名不变）。 +归属校验：`assert_customer_access`（G-01，越权 403 + audit，A-9）。 +权限矩阵：GET alerts = risk_officer 全量 / compliance 强制 aml（A-7）； +handle 仅 risk_officer；suitability/check 走 G-01；aml/scan 仅 risk_officer。 +handler_result 枚举由请求模型 Literal 校验（repo 不校验，开发计划 B6 备注）。 +挂载：B7 集成 main.py；统一响应外壳挂账 B7。 +""" + +from __future__ import annotations + +from dataclasses import asdict +from datetime import datetime +from typing import Literal + +from fastapi import APIRouter, Depends, HTTPException, Query +from pydantic import BaseModel, Field + +from app.api.deps import AuthContext, assert_customer_access, get_auth_context +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.risk_repository import RiskRepository +from app.service.risk.alert_service import handle_alert +from app.service.risk.aml_service import scan_all +from app.service.suitability import suitability_check +from app.utils.exceptions import StateConflict +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 点）。""" + return RiskRepository() + + +class HandleRequest(BaseModel): + handler_result: Literal["confirmed_normal", "confirmed_suspicious", "reported"] + handler_comment: str | None = Field(None, max_length=512) + + +class SuitabilityCheckRequest(BaseModel): + customer_id: str = Field(..., min_length=1) + product_id: str = Field(..., min_length=1) + + +@router.get("/alerts") +def list_alerts_api( + auth: AuthContext = Depends(get_auth_context), + status: str | None = None, + alert_type: str | None = None, + customer_id: str | None = None, + start: datetime | None = None, + end: datetime | None = None, + page: int = Query(1, ge=1), + page_size: int = Query(20, ge=1, le=100), +) -> dict: + """预警台账分页（FR-4）。compliance 强制 aml 过滤（A-7：仅返回 aml 单）。""" + if auth.has_role("risk_officer"): + pass + elif auth.has_role("compliance"): + alert_type = "aml" + else: + raise HTTPException(status_code=403, detail="risk_officer/compliance only") + rows, total = _repo().list_alerts( + status=status, + alert_type=alert_type, + customer_id=customer_id, + start=start, + end=end, + page=page, + page_size=page_size, + ) + return {"items": rows, "total": total, "page": page, "page_size": page_size} + + +@router.post("/alerts/{alert_id}/handle") +def handle_alert_api(alert_id: str, req: HandleRequest, auth: AuthContext = Depends(get_auth_context)) -> dict: + """人工处置（FR-4）：仅 risk_officer（A-7 compliance 403）；状态机 + 审计。""" + if not auth.has_role("risk_officer"): + raise HTTPException(status_code=403, detail="risk_officer only") + try: + return handle_alert( + alert_id, req.handler_result, auth.actor_id, req.handler_comment, risk_repo=_repo() + ) + except LookupError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + except StateConflict as exc: + raise HTTPException(status_code=409, detail=str(exc)) from exc + + +@router.post("/suitability/check") +def suitability_check_api(req: SuitabilityCheckRequest, auth: AuthContext = Depends(get_auth_context)) -> dict: + """适当性校验（FR-2/G-01）：customer 仅本人、advisor 名下、risk_officer 全量。""" + try: + core = CoreReadOnlyRepository() + assert_customer_access(auth, req.customer_id, core_ro=core, risk_repo=_repo()) + result = suitability_check( + req.customer_id, req.product_id, core_ro=core, risk_repo=_repo() + ) + except LookupError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc + return asdict(result) + + +@router.post("/aml/scan") +def aml_scan_api(auth: AuthContext = Depends(get_auth_context)) -> dict: + """手动全量 AML 扫描（FR-5 触发时机 2）：仅 risk_officer。""" + if not auth.has_role("risk_officer"): + raise HTTPException(status_code=403, detail="risk_officer only") + repo = _repo() + summary = scan_all(core_ro=CoreReadOnlyRepository(), risk_repo=repo) + repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "aml_scan", + "agent_type": "risk", + "actor_id": auth.actor_id, + "customer_id": None, + "rule_id": None, + "input_summary": dict(summary), + "decision": "scan_completed", + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + return summary diff --git a/app/service/risk/alert_service.py b/app/service/risk/alert_service.py index 27c418e..cd086a4 100644 --- a/app/service/risk/alert_service.py +++ b/app/service/risk/alert_service.py @@ -261,6 +261,52 @@ def record_suitability_alert( ) +def handle_alert( + alert_id: str, + handler_result: str, + handler_id: str, + handler_comment: str | None = None, + risk_repo: RiskRepository | None = None, +) -> dict[str, Any]: + """人工处置（FR-4；角色鉴权在依赖层，本层管状态机与审计）。 + + 仅 pending_review 可处置（禁止跳改已处置单，A-7）；处置动作全量写 + audit_log（agent_type='risk'，handler_* 三字段落审计行）。 + """ + repo = risk_repo or RiskRepository() + alert = repo.get_alert(alert_id) + if alert is None: + from app.utils.exceptions import NotFoundError + + raise NotFoundError(f"alert not found: {alert_id}") + if alert["status"] != "pending_review": + from app.utils.exceptions import StateConflict + + raise StateConflict(f"alert {alert_id} already handled (status={alert['status']})") + ok = repo.update_alert_status(alert_id, handler_result, handler_id, handler_comment) + if not ok: # 读后并发窗口：他人已抢先处置 + from app.utils.exceptions import StateConflict + + raise StateConflict(f"alert {alert_id} state changed concurrently") + repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "alert_handle", + "agent_type": "risk", + "actor_id": handler_id, + "customer_id": alert["customer_id"], + "rule_id": None, + "input_summary": {"alert_id": alert_id, "prev_status": alert["status"]}, + "decision": "alert_handled", + "risk_score": None, + "handler_id": handler_id, + "handler_result": handler_result, + "handler_comment": handler_comment, + } + ) + return repo.get_alert(alert_id) + + def record_aml_alert( customer_id: str, aml_detail: dict[str, Any], diff --git a/app/utils/exceptions.py b/app/utils/exceptions.py index a7f5d40..0a82bd0 100644 --- a/app/utils/exceptions.py +++ b/app/utils/exceptions.py @@ -1 +1,24 @@ -"""业务异常与鉴权错误码（对齐 JWT-RBAC 手册）。""" +"""业务异常与鉴权错误码（对齐 JWT-RBAC 手册）。 + +NotFoundError 继承 LookupError：服务层既有 `raise LookupError`（customer/product +not found）自动获得精确类型与 404 语义，调用方 except LookupError 无需改动 +（B5 评审 P3-1 的收敛锚点）。 +""" + +from __future__ import annotations + + +class NotFoundError(LookupError): + """资源不存在（HTTP 404）。""" + + +class PermissionDenied(PermissionError): + """归属/权限拒绝（HTTP 403）；code 对齐 JWT 手册 §6.1 错误码。""" + + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + + +class StateConflict(ValueError): + """状态机冲突（HTTP 409；如处置非 pending_review 的预警单）。""" diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py new file mode 100644 index 0000000..ca1bd09 --- /dev/null +++ b/tests/test_risk_api.py @@ -0,0 +1,326 @@ +"""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 create_engine, text +from sqlalchemy.pool import StaticPool + +from app.api import risk as risk_api +from app.api.deps import assert_customer_access, permission_denied_handler +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 alert_service +from app.utils.exceptions import PermissionDenied + + +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_engine( + "sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False} + ) + with engine.begin() as conn: + for ddl in [ + """CREATE TABLE core_customer ( + customer_id VARCHAR(64) PRIMARY KEY, display_name VARCHAR(128), age INTEGER, + occupation VARCHAR(64), open_date DATE, is_active TINYINT DEFAULT 1)""", + """CREATE TABLE core_customer_risk ( + customer_id VARCHAR(64), risk_code VARCHAR(8), evaluated_at TIMESTAMP)""", + """CREATE TABLE core_customer_advisor ( + advisor_id VARCHAR(64), customer_id VARCHAR(64), rel_status VARCHAR(16))""", + """CREATE TABLE core_product ( + product_id VARCHAR(64) PRIMARY KEY, product_name VARCHAR(128), + min_risk_code VARCHAR(8), product_type VARCHAR(32))""", + """CREATE TABLE risk_aml_list ( + id INTEGER PRIMARY KEY AUTOINCREMENT, list_id VARCHAR(64), list_type VARCHAR(16), + full_name VARCHAR(128), match_threshold REAL, source VARCHAR(64), + list_version VARCHAR(16), effective_date DATE, is_active TINYINT DEFAULT 1, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + """CREATE TABLE risk_alert ( + alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64), + trade_id VARCHAR(64), alert_type VARCHAR(16), triggered_rules TEXT, + risk_score INTEGER, status VARCHAR(24) DEFAULT 'pending_review', payload TEXT, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, handled_at TIMESTAMP)""", + """CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY, trace_id VARCHAR(64), event_type VARCHAR(64), + agent_type VARCHAR(16), actor_id VARCHAR(64), customer_id VARCHAR(64), + rule_id VARCHAR(64), input_summary TEXT, decision VARCHAR(64), risk_score INTEGER, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + """CREATE TABLE risk_suitability_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id VARCHAR(64), customer_id VARCHAR(64), + product_id VARCHAR(64), customer_risk_level VARCHAR(8), product_risk_level VARCHAR(8), + is_matched TINYINT, is_blocked TINYINT, block_reason VARCHAR(512), + request_ref VARCHAR(64), profile_l1_version VARCHAR(32), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + """CREATE TABLE customer_profile_l3 ( + customer_id VARCHAR(64) PRIMARY KEY, monitor_tier VARCHAR(16) NOT NULL, + risk_score INTEGER, score_dimensions TEXT, monitor_tags TEXT, + last_alert_id VARCHAR(64), computed_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + ]: + conn.execute(text(ddl)) + 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) VALUES" + " ('CUST-1001', 'C1', :t), ('CUST-3001', 'C3', :t)" + ), + {"t": datetime.now() - timedelta(days=30)}, + ) + conn.execute( + text( + "INSERT INTO core_customer_advisor VALUES ('ADV-01', 'CUST-3001', 'active')" + ) + ) + conn.execute( + text( + "INSERT INTO core_product 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)) + monkeypatch.setattr(risk_api, "CoreReadOnlyRepository", lambda: CoreReadOnlyRepository(engine=engine)) + app = FastAPI() + app.include_router(risk_router) + app.add_exception_handler(PermissionDenied, permission_denied_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") +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_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): + 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() + assert body["blocked"] is True and body["rule_id"] == "SUIT-001" # 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 + + +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 "AUTH_403_NOT_OWNER" in r.json()["detail"] + 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 "AUTH_403_NOT_ASSIGNED" in r.json()["detail"] + 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 "AUTH_403_SCOPE" in r.json()["detail"] + + +# ---------- 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, "hit_customers": 1, "alerts": body["alerts"]} # CUST-3001 命中 + assert len(body["alerts"]) == 1 + l3 = repo.get_l3("CUST-3001") + assert l3["monitor_tier"] == "high" # scan 命中标记 L3 + 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 + + +# ---------- 非 dev 环境拒绝 debug 鉴权 ---------- + + +def test_non_dev_rejects_debug_auth(client, monkeypatch): + from app.config.settings import settings + + monkeypatch.setattr(settings, "app_env", "production") + with pytest.raises(RuntimeError, match="debug auth disabled"): + client.get("/api/risk/alerts", headers=OFFICER) + + +# ---------- 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", risk_repo=repo) + assert ei.value.code == "AUTH_403_SCOPE"