258 lines
11 KiB
Python
258 lines
11 KiB
Python
"""API 依赖：鉴权上下文（架构 §5.7 · 归属校验统一在此层）。
|
||
|
||
AuthContext 模型在 A4 冻结（开发计划 v1.1）；B6 落地 `get_auth_context()`；
|
||
T-01（2026-09-07）工厂内部替换为 JWT 解析（auth_service.verify_token，
|
||
手册 §4/§8），`Authorization: Bearer` 为所有环境首选通道：
|
||
- JWT 通道：验签 + claims + jti 吊销 + `X-Agent-Type` 交叉校验（手册 §4.6
|
||
必填、§5.4 准入矩阵；缺失 AUTH_401_MISSING_AGENT_TYPE、不符
|
||
AUTH_403_AGENT_MISMATCH）；
|
||
- debug 头兜底：仅 development 无 Bearer 时从 `X-Debug-Role`/`X-Debug-Actor`
|
||
构造（B6 过渡口径，演示 SOP 与既有测试依赖此通道；不校验 X-Agent-Type）；
|
||
非 dev 无 Bearer 一律 401 留痕（AUTH_401_MISSING_BEARER）。
|
||
|
||
归属断言 `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）；响应体为统一错误结构（utils/response，手册 §10，B7 挂账④）；
|
||
多角色按 fail-closed 口径固化（customer 分支优先，命中 deny 即拒，不再并集
|
||
放宽——评审 P3-1②）。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from fastapi import Request
|
||
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.service.auth_service import Claims, TokenInvalid, verify_token
|
||
from app.utils.exceptions import ApiError, 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"
|
||
AGENT_TYPE_HEADER = "X-Agent-Type"
|
||
AGENT_TYPES = ("customer", "advisor", "analyst", "risk")
|
||
|
||
# 手册 §5.4 Agent × 角色/token_type 准入矩阵（JWT 通道强制；debug 头通道
|
||
# 维持 B6 路由内角色口径——演示 SOP 的 compliance 台账只读属其扩展）
|
||
AGENT_ACCESS_MATRIX: dict[str, dict[str, tuple[str, ...]]] = {
|
||
"customer": {"token_types": ("customer",), "roles": ("customer",)},
|
||
"advisor": {"token_types": ("staff",), "roles": ("advisor", "compliance", "ops")},
|
||
"analyst": {"token_types": ("staff",), "roles": ("analyst", "compliance")},
|
||
"risk": {"token_types": ("staff", "service"), "roles": ("risk_officer", "service_risk")},
|
||
}
|
||
|
||
# T-01 完成：JWT 已接入，debug 头降级为 dev 兜底（main.lifespan 据此放行
|
||
# 非 dev 启动，改为校验 jwt_ready）。
|
||
AUTH_FACTORY_IS_DEBUG = False
|
||
|
||
|
||
class AuthContext(BaseModel):
|
||
"""统一鉴权上下文（全部 API 依赖层的产出；service 层签名接收此类型）。
|
||
|
||
T-01 扩展（手册 §8.1）：token_type/permissions/tenant_id/jti；debug 头
|
||
通道 token_type 按角色推断（customer 角色 → customer，其余 staff），
|
||
permissions/tenant_id/jti 为空（过渡口径）。
|
||
"""
|
||
|
||
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；其余为空")
|
||
token_type: str = Field("staff", description="customer | staff | service（手册 §3）")
|
||
permissions: list[str] = Field(default_factory=list, description="细粒度权限（JWT claims）")
|
||
tenant_id: str | None = Field(None, description="租户/法人主体")
|
||
jti: str | None = Field(None, description="Token 唯一 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 has_permission(self, permission: str) -> bool:
|
||
return permission in self.permissions
|
||
|
||
|
||
def _authz_audit(
|
||
risk_repo: RiskRepository,
|
||
auth: AuthContext | None,
|
||
customer_id: str | None,
|
||
code: str,
|
||
agent_type: str = "risk",
|
||
) -> None:
|
||
"""鉴权失败审计（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": agent_type,
|
||
"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,
|
||
agent_type: str = "risk",
|
||
) -> None:
|
||
"""越权出口：审计 + 403（全部 403 必经此函数，保证留痕与错误码）。"""
|
||
_authz_audit(risk_repo, auth, customer_id, code, agent_type)
|
||
raise PermissionDenied(code, message or f"forbidden: {code}")
|
||
|
||
|
||
def _unauthenticated_audit(
|
||
risk_repo: RiskRepository,
|
||
actor_id: str,
|
||
code: str,
|
||
agent_type: str = "risk",
|
||
) -> None:
|
||
"""401 留痕（P-05；B6 debug 通道既有口径，T-01 推广到 JWT 通道）。"""
|
||
risk_repo.insert_audit_log(
|
||
{
|
||
"trace_id": current_trace() or new_trace(),
|
||
"event_type": "authz",
|
||
"agent_type": agent_type,
|
||
"actor_id": actor_id or "anonymous",
|
||
"customer_id": None,
|
||
"rule_id": None,
|
||
"input_summary": {"code": code},
|
||
"decision": "unauthenticated",
|
||
"risk_score": None,
|
||
"handler_id": None,
|
||
"handler_result": None,
|
||
"handler_comment": None,
|
||
}
|
||
)
|
||
|
||
|
||
def _claims_to_auth(claims: Claims) -> AuthContext:
|
||
"""JWT claims → AuthContext（手册 §8.1；customer_id 仅 customer token）。"""
|
||
return AuthContext(
|
||
actor_id=claims.sub,
|
||
roles=claims.roles,
|
||
customer_id=claims.customer_id if claims.token_type == "customer" else None,
|
||
token_type=claims.token_type,
|
||
permissions=claims.permissions,
|
||
tenant_id=claims.tenant_id,
|
||
jti=claims.jti,
|
||
)
|
||
|
||
|
||
def assert_agent_access(
|
||
auth: AuthContext,
|
||
agent_type: str,
|
||
risk_repo: RiskRepository | None = None,
|
||
) -> None:
|
||
"""Agent 入口准入（手册 §5.4 矩阵；chat 等显式入口调用）。
|
||
|
||
不匹配 → AUTH_403_AGENT_MISMATCH + 审计（deny）。risk_repo 传 None 时
|
||
仅抛异常不留痕（仅限 JWT 通道已审计的交叉校验复用路径）。
|
||
"""
|
||
rule = AGENT_ACCESS_MATRIX.get(agent_type)
|
||
if rule is None:
|
||
raise ApiError(400, "BAD_REQUEST", f"unknown agent_type: {agent_type}")
|
||
if auth.token_type in rule["token_types"] and set(auth.roles) & set(rule["roles"]):
|
||
return
|
||
if risk_repo is not None:
|
||
deny(auth, "AUTH_403_AGENT_MISMATCH", risk_repo, message=f"not allowed for agent '{agent_type}'")
|
||
raise PermissionDenied("AUTH_403_AGENT_MISMATCH", f"not allowed for agent '{agent_type}'")
|
||
|
||
|
||
def get_auth_context(request: Request) -> AuthContext:
|
||
"""鉴权工厂（T-01）：Bearer JWT 优先（全环境），dev 无 Bearer 时 debug 头兜底。
|
||
|
||
JWT 通道强制 `X-Agent-Type` 交叉校验（手册 §4.6/§5.4）：缺失 401、
|
||
值域外 400、与 token 不符 403（均留痕）。debug 通道保持 B6 行为。
|
||
"""
|
||
auth_header = request.headers.get("Authorization", "")
|
||
if auth_header.startswith("Bearer "):
|
||
token = auth_header[len("Bearer "):].strip()
|
||
if not token:
|
||
_unauthenticated_audit(RiskRepository(), "anonymous", "AUTH_401_INVALID_TOKEN")
|
||
raise ApiError(401, "AUTH_401_INVALID_TOKEN", "empty bearer token")
|
||
try:
|
||
claims = verify_token(token)
|
||
except TokenInvalid as exc:
|
||
_unauthenticated_audit(RiskRepository(), "anonymous", exc.code)
|
||
raise ApiError(401, exc.code, exc.message) from exc
|
||
auth = _claims_to_auth(claims)
|
||
|
||
agent_type = request.headers.get(AGENT_TYPE_HEADER, "").strip()
|
||
if not agent_type:
|
||
_unauthenticated_audit(RiskRepository(), auth.actor_id, "AUTH_401_MISSING_AGENT_TYPE")
|
||
raise ApiError(401, "AUTH_401_MISSING_AGENT_TYPE", f"missing {AGENT_TYPE_HEADER} header")
|
||
if agent_type not in AGENT_TYPES:
|
||
raise ApiError(400, "BAD_REQUEST", f"invalid {AGENT_TYPE_HEADER}: {agent_type}")
|
||
# 交叉校验失败也走 deny 全量审计（fail-closed；agent_type 归请求目标）
|
||
assert_agent_access(auth, agent_type, risk_repo=RiskRepository())
|
||
request.state.auth = auth
|
||
return auth
|
||
|
||
if settings.app_env != "development":
|
||
_unauthenticated_audit(RiskRepository(), "anonymous", "AUTH_401_MISSING_BEARER")
|
||
raise ApiError(401, "AUTH_401_MISSING_BEARER", "missing Authorization bearer token")
|
||
|
||
# dev debug 头兜底（B6 过渡口径；演示 SOP 与既有权限矩阵测试依赖此通道）
|
||
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:
|
||
# 401 也留痕（P1-1）；debug 通道仅 dev，生产等价流量由 JWT 中间件拒绝
|
||
_unauthenticated_audit(
|
||
RiskRepository(), actor_id, "AUTH_401_MISSING_DEBUG_HEADERS"
|
||
)
|
||
raise ApiError(
|
||
401, "AUTH_401_MISSING_DEBUG_HEADERS", "missing X-Debug-Role/X-Debug-Actor headers"
|
||
)
|
||
return AuthContext(
|
||
actor_id=actor_id,
|
||
roles=roles,
|
||
customer_id=actor_id if "customer" in roles else None,
|
||
token_type="customer" if "customer" in roles else "staff",
|
||
)
|
||
|
||
|
||
def assert_customer_access(
|
||
auth: AuthContext,
|
||
customer_id: str,
|
||
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(auth, "AUTH_403_NOT_OWNER", risk_repo, customer_id)
|
||
if "advisor" in auth.roles:
|
||
if core_ro.is_advisor_assigned(auth.actor_id, customer_id):
|
||
return
|
||
deny(auth, "AUTH_403_NOT_ASSIGNED", risk_repo, customer_id)
|
||
deny(auth, "AUTH_403_SCOPE", risk_repo, customer_id)
|