diff --git a/app/api/deps.py b/app/api/deps.py index 7f40a70..fa22f04 100644 --- a/app/api/deps.py +++ b/app/api/deps.py @@ -8,6 +8,9 @@ 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）。 +所有 403/401 一律经 `deny`/`unauthenticated` 审计（手册 P-05 全局铁律， +B6 评审 P1-1）；多角色按 fail-closed 口径固化（customer 分支优先，命中 deny +即拒，不再并集放宽——评审 P3-1②，T-01 引入 token_type 后收紧）。 """ from __future__ import annotations @@ -41,6 +44,48 @@ class AuthContext(BaseModel): return "customer" in self.roles +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 _authz_audit( + risk_repo: RiskRepository, + auth: AuthContext | None, + customer_id: str | None, + code: str, +) -> None: + """鉴权失败审计（event_type='authz'；手册 P-05，B6 评审 P1-1）。""" + risk_repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "authz", + "agent_type": "risk", + "actor_id": auth.actor_id if auth else "anonymous", + "customer_id": customer_id, + "rule_id": None, + "input_summary": {"roles": auth.roles if auth else [], "code": code}, + "decision": "forbidden", + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + + +def deny( + auth: AuthContext, + code: str, + risk_repo: RiskRepository, + customer_id: str | None = None, + message: str | None = None, +) -> None: + """越权出口：审计 + 403（全部 403 必经此函数，保证留痕与错误码）。""" + _authz_audit(risk_repo, auth, customer_id, code) + raise PermissionDenied(code, message or f"forbidden: {code}") + + def get_auth_context(request: Request) -> AuthContext: """鉴权工厂（B6）：dev 读 debug 头，非 dev 拒绝；T-01 后替换内部为 JWT 解析。""" if settings.app_env != "development": @@ -50,7 +95,25 @@ def get_auth_context(request: Request) -> AuthContext: 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") + # 401 也留痕（P1-1）；debug 通道仅 dev，生产等价流量由 JWT 中间件拒绝 + repo = RiskRepository() + repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "authz", + "agent_type": "risk", + "actor_id": actor_id or "anonymous", + "customer_id": None, + "rule_id": None, + "input_summary": {"roles": roles, "code": "AUTH_401_MISSING_DEBUG_HEADERS"}, + "decision": "unauthenticated", + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + raise HTTPException(status_code=401, detail="AUTH_401_MISSING_DEBUG_HEADERS") return AuthContext( actor_id=actor_id, roles=roles, @@ -58,55 +121,27 @@ def get_auth_context(request: Request) -> AuthContext: ) -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, + core_ro: CoreReadOnlyRepository, + risk_repo: RiskRepository, ) -> None: """G-01 归属断言（customer/advisor/risk_officer；其他角色一律拒绝）。 customer 仅本人；advisor 需 customer_advisor_rel active；risk_officer 全量。 compliance 不在客户业务数据访问白名单（仅审计类读，JWT 手册 §5.3）。 + core_ro/risk_repo 必传（评审 P3-3/P3-4：审计与归属查询不得静默降级）。 + 多角色 fail-closed：customer 分支 deny 即终止（P3-1② 固化口径）。 """ 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") + deny(auth, "AUTH_403_NOT_OWNER", risk_repo, customer_id) if "advisor" in auth.roles: - core = core_ro or CoreReadOnlyRepository() - if core.is_advisor_assigned(auth.actor_id, customer_id): + if core_ro.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") + deny(auth, "AUTH_403_NOT_ASSIGNED", risk_repo, customer_id) + deny(auth, "AUTH_403_SCOPE", risk_repo, customer_id) diff --git a/app/api/risk.py b/app/api/risk.py index 9896f9e..4a0ac3b 100644 --- a/app/api/risk.py +++ b/app/api/risk.py @@ -1,11 +1,15 @@ """风控 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）。 +归属校验：`assert_customer_access`（G-01，越权 403 + audit，A-9）；全部 403/401 +经 deps.deny/依赖层审计（手册 P-05，B6 评审 P1-1）。 权限矩阵：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。 +直调 suitability/check 每次落 audit（request_ref='api:suitability_check'， +评审 P2-3/P3-9）；aml/scan 幂等防护挂账 B9b 前（评审 P3-6）。 +挂载：B7 集成 main.py；统一响应外壳（含错误体结构对齐手册 §10、预警类响应 +附 disclaimer）挂账 B7（评审 P3-2/P3-7）。 """ from __future__ import annotations @@ -17,7 +21,7 @@ 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.api.deps import AuthContext, assert_customer_access, deny, 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 @@ -52,8 +56,8 @@ def list_alerts_api( status: str | None = None, alert_type: str | None = None, customer_id: str | None = None, - start: datetime | None = None, - end: datetime | None = None, + start_date: datetime | None = None, + end_date: datetime | None = None, page: int = Query(1, ge=1), page_size: int = Query(20, ge=1, le=100), ) -> dict: @@ -63,13 +67,13 @@ def list_alerts_api( elif auth.has_role("compliance"): alert_type = "aml" else: - raise HTTPException(status_code=403, detail="risk_officer/compliance only") + deny(auth, "AUTH_403_ROLE", _repo(), message="risk_officer/compliance only") rows, total = _repo().list_alerts( status=status, alert_type=alert_type, customer_id=customer_id, - start=start, - end=end, + start=start_date, + end=end_date, page=page, page_size=page_size, ) @@ -80,7 +84,7 @@ def list_alerts_api( 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") + deny(auth, "AUTH_403_ROLE", _repo(), message="risk_officer only") try: return handle_alert( alert_id, req.handler_result, auth.actor_id, req.handler_comment, risk_repo=_repo() @@ -93,15 +97,41 @@ def handle_alert_api(alert_id: str, req: HandleRequest, auth: AuthContext = Depe @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 全量。""" + """适当性校验（FR-2/G-01）：customer 仅本人、advisor 名下、risk_officer 全量。 + + 直调路径每次校验补 audit（PRD §7.3 全量留痕；网关路径由 trade_request + 审计兜底，评审 P2-3）。 + """ + repo = _repo() try: core = CoreReadOnlyRepository() - assert_customer_access(auth, req.customer_id, core_ro=core, risk_repo=_repo()) + 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() + req.customer_id, req.product_id, core_ro=core, risk_repo=repo, + request_ref="api:suitability_check", ) except LookupError as exc: raise HTTPException(status_code=404, detail=str(exc)) from exc + repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "suitability_check", + "agent_type": "risk", + "actor_id": auth.actor_id, + "customer_id": req.customer_id, + "rule_id": result.rule_id, + "input_summary": { + "product_id": req.product_id, + "request_ref": "api:suitability_check", + "reasons": list(result.reasons), + }, + "decision": "suitability_blocked" if result.blocked else "suitability_passed", + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) return asdict(result) @@ -109,7 +139,7 @@ def suitability_check_api(req: SuitabilityCheckRequest, auth: AuthContext = Depe 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") + deny(auth, "AUTH_403_ROLE", _repo(), message="risk_officer only") repo = _repo() summary = scan_all(core_ro=CoreReadOnlyRepository(), risk_repo=repo) repo.insert_audit_log( diff --git a/app/api/simulate.py b/app/api/simulate.py index 85ecde4..dbec936 100644 --- a/app/api/simulate.py +++ b/app/api/simulate.py @@ -1,7 +1,8 @@ """模拟交易网关路由（PRD FR-1 · 薄路由，不含业务）。 -鉴权：依赖 T-01 JWT（risk_demo 或客户本人 customer_id == subject）； -`get_auth_context` B6 落地后在此挂 `Depends`，本路由签名不变（开发计划 B5/B6 边界）。 +鉴权：`Depends(get_auth_context)`（B6 回挂，评审 P2-2）——一期接受 +risk_demo 演示账号或客户本人（auth.customer_id == 请求 customer_id， +PRD FR-1 §鉴权）；越权经 deps.deny 审计后 403。T-01 后工厂内部换 JWT。 trace：B7 中间件贯通；B7 前由 service 层 ensure_trace 兜底。 挂载：B7 集成 main.py（当前仅 TestClient 独立挂 router 验证）。 统一响应外壳：utils/response.py 为占位（P1 任务），落地点挂账 B7（届时 @@ -12,14 +13,21 @@ from __future__ import annotations from decimal import Decimal -from fastapi import APIRouter, HTTPException +from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field +from app.api.deps import AuthContext, deny, get_auth_context from app.gateway.trade_gateway import UnsupportedTradeType, submit_trade +from app.repository.risk_repository import RiskRepository router = APIRouter(prefix="/api/simulate", tags=["simulate"]) +def _repo() -> RiskRepository: + """审计仓储（deny 留痕用；测试 monkeypatch 点）。""" + return RiskRepository() + + class TradeRequest(BaseModel): customer_id: str = Field(..., min_length=1) product_id: str = Field(..., min_length=1) @@ -28,11 +36,16 @@ class TradeRequest(BaseModel): @router.post("/trade") -def submit_trade_api(req: TradeRequest) -> dict: +def submit_trade_api(req: TradeRequest, auth: AuthContext = Depends(get_auth_context)) -> dict: """模拟交易（FR-1）：适当性阻断或放行+引擎判定，返回 blocked + trade_id。""" + if not (auth.has_role("risk_demo") or (auth.is_customer() and auth.customer_id == req.customer_id)): + deny( + auth, "AUTH_403_ROLE", _repo(), + customer_id=req.customer_id, message="risk_demo or owner customer only", + ) try: return submit_trade(req.model_dump()) except UnsupportedTradeType as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - except LookupError as exc: # KeyError 亦为其子类；B6 统一 NotFoundError 时收敛为精确类型 + except LookupError as exc: # NotFoundError 亦为其子类；已统一（B6 评审 P3-5） raise HTTPException(status_code=404, detail=str(exc)) from exc diff --git a/app/service/risk/alert_service.py b/app/service/risk/alert_service.py index cd086a4..e607534 100644 --- a/app/service/risk/alert_service.py +++ b/app/service/risk/alert_service.py @@ -16,6 +16,7 @@ from uuid import uuid4 from app.config.settings import settings from app.repository.risk_repository import EVENT_ALERT_TYPES, RiskRepository from app.service.risk.rules import RuleHit +from app.utils.exceptions import NotFoundError, StateConflict from app.utils.trace import current_trace, new_trace logger = logging.getLogger(__name__) @@ -276,17 +277,11 @@ def handle_alert( 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( { diff --git a/app/service/suitability.py b/app/service/suitability.py index 580affb..4644bbf 100644 --- a/app/service/suitability.py +++ b/app/service/suitability.py @@ -16,6 +16,7 @@ from typing import Any 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 NotFoundError from app.utils.trace import current_trace, new_trace TIER_ORDER = {"C1": 1, "C2": 2, "C3": 3, "C4": 4, "C5": 5} @@ -152,10 +153,10 @@ def suitability_check( customer = core_ro.get_customer_l0(customer_id) if customer is None: - raise LookupError(f"customer not found: {customer_id}") + raise NotFoundError(f"customer not found: {customer_id}") product = core_ro.get_product(product_id) if product is None: - raise LookupError(f"product not found: {product_id}") + raise NotFoundError(f"product not found: {product_id}") result = check_core(customer, product, valid_days=valid_days, today=today) diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py index ca1bd09..d358a74 100644 --- a/tests/test_risk_api.py +++ b/tests/test_risk_api.py @@ -126,7 +126,10 @@ 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)) + # 401/越权审计经 deps 内 RiskRepository 兜底构造,统一注入 sqlite(B6 评审 P3-4) + from app.api import deps as deps_mod + + monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) app = FastAPI() app.include_router(risk_router) app.add_exception_handler(PermissionDenied, permission_denied_handler) @@ -322,5 +325,35 @@ def test_assert_customer_access_scope_denial(env): auth = AuthContext(actor_id="STAFF-40001", roles=["compliance"]) with pytest.raises(PermissionDenied) as ei: - assert_customer_access(auth, "CUST-1001", risk_repo=repo) + 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 "AUTH_403_NOT_OWNER" in r.json()["detail"] diff --git a/tests/test_trade_gateway.py b/tests/test_trade_gateway.py index f7612a1..6160faa 100644 --- a/tests/test_trade_gateway.py +++ b/tests/test_trade_gateway.py @@ -9,6 +9,8 @@ from decimal import Decimal import pytest from fastapi import FastAPI + +from app.api.deps import PermissionDenied, permission_denied_handler from fastapi.testclient import TestClient from sqlalchemy import create_engine, text from sqlalchemy.pool import StaticPool @@ -234,20 +236,30 @@ def client(env, monkeypatch): monkeypatch.setattr(tg, "CoreReadOnlyRepository", lambda: core) monkeypatch.setattr(tg, "RiskRepository", lambda: repo) monkeypatch.setattr(tg, "GatewayRepository", lambda: writer) + # 401/越权审计经 deps/simulate 内仓储构造点,统一注入 sqlite(B6 评审 P3-4) + from app.api import deps as deps_mod + from app.api import simulate as simulate_mod + + monkeypatch.setattr(deps_mod, "RiskRepository", lambda: repo) + monkeypatch.setattr(simulate_mod, "_repo", lambda: repo) app = FastAPI() + app.add_exception_handler(PermissionDenied, permission_denied_handler) app.include_router(simulate_router) with TestClient(app) as c: yield c +DEMO = {"X-Debug-Role": "risk_demo", "X-Debug-Actor": "STAFF-90001"} + + def test_api_convert_returns_400(client): - r = client.post("/api/simulate/trade", json=_req(ttype="convert")) + r = client.post("/api/simulate/trade", json=_req(ttype="convert"), headers=DEMO) assert r.status_code == 400 assert "转换交易暂不支持" in r.json()["detail"] def test_api_blocked_returns_200_with_blocked_true(client): - r = client.post("/api/simulate/trade", json=_req()) # C1 买 R4 + r = client.post("/api/simulate/trade", json=_req(), headers=DEMO) # C1 买 R4 assert r.status_code == 200 body = r.json() assert body["blocked"] is True and "SUIT-001" in body["block_reason"] @@ -257,6 +269,7 @@ def test_api_accepted_returns_200_with_trade_id(client): r = client.post( "/api/simulate/trade", json=_req(customer="CUST-3001", product="PROD-510300", amount="600000"), + headers=DEMO, ) assert r.status_code == 200 body = r.json() @@ -264,11 +277,30 @@ def test_api_accepted_returns_200_with_trade_id(client): assert body["triggered_rules"] == ["RISK-001", "RISK-002"] +def test_api_customer_owner_allowed_other_denied(client, env): + """B6 评审 P2-2:客户本人放行进业务(适当性阻断与否由业务层决定),查他人 403+审计。""" + _, repo, _, _, engine = env + # CUST-1001(C1)买 R3:鉴权通过进入业务,业务层适当性阻断(200 blocked=true) + r = client.post( + "/api/simulate/trade", + json=_req(customer="CUST-1001", product="PROD-510300", amount="1000"), + headers={"X-Debug-Role": "customer", "X-Debug-Actor": "CUST-1001"}, + ) + assert r.status_code == 200 and r.json()["blocked"] is True # 业务响应,非 403 + r = client.post( + "/api/simulate/trade", + json=_req(customer="CUST-3001", product="PROD-510300", amount="1000"), + 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 + + def test_api_non_positive_amount_returns_422(client): - r = client.post("/api/simulate/trade", json=_req(amount="0")) + r = client.post("/api/simulate/trade", json=_req(amount="0"), headers=DEMO) assert r.status_code == 422 def test_api_unknown_customer_returns_404(client): - r = client.post("/api/simulate/trade", json=_req(customer="CUST-9999")) + r = client.post("/api/simulate/trade", json=_req(customer="CUST-9999"), headers=DEMO) assert r.status_code == 404