113 lines
4.7 KiB
Python
113 lines
4.7 KiB
Python
"""API 依赖：鉴权上下文（架构 §5.7 · 归属校验统一在此层）。
|
||
|
||
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 层签名接收此类型）。"""
|
||
|
||
actor_id: str = Field(..., description="操作者 ID：staff_id 或 customer_id")
|
||
roles: list[str] = Field(default_factory=list, description="角色集合，如 ['risk_officer','risk_demo']")
|
||
customer_id: str | None = Field(None, description="customer 角色时 = 本人 customer_id；其余为空")
|
||
|
||
def has_role(self, *roles: str) -> bool:
|
||
return any(r in self.roles for r in roles)
|
||
|
||
def is_customer(self) -> bool:
|
||
return "customer" in self.roles
|
||
|
||
|
||
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")
|