GaoYiYuan_0626
8c226f0c6d
背景：远端 main 新提交 3995cb4（09-07 17:20，交接文档漏记）为 Wave 0 鉴权/ chat/防护平行实现，与已完工 T-01/T-02/T-03/T-06 同名不同路径；试合并实测 20 文件冲突（原记 9 个），另有 13 个 main 新增文件不报冲突会静默并入。 拍板：风控 Agent 按独立封装模块自治，与宿主耦合收敛到 4 个接缝。 1. 《风控Agent模块边界与合并接缝标注》入库存档：A~D 四类文件归属表； 4 接缝（S1 挂载点 main.py / S2 AuthContext / S3 settings / S4 引擎工厂）； 20 冲突文件逐个裁决（core_ro、model/suitability、conftest、02-seed-base 以模块版为准；chat/main/settings/agent_service 等公共层以 main 为主）； 三处硬伤处置：issuer 不一致改为适配器映射不统一、STAFF-90001 必保、 main infer_roles 未知 actor 默认 analyst（fail-open）记宿主侧 P1。 2. app/api/auth_adapter.py：S2 接缝适配器预制件（当前未接线，AL-09 接入）。 鸭子类型读宿主 ctx 故不依赖宿主文件；sub→actor_id、trace_id→contextvar、 perm_matches 兼容宿主 `前缀:*` 通配；缺主体即 HostAuthAdapterError， fail-closed 不静默降级。 3. tests/test_module_boundary.py：边界防呆 4 类断言——模块私有文件存在、 禁止跨层 import 宿主私有实现（gateway.*/config.database/middleware.*/ utils.input_guard/model.schemas）、AuthContext 契约完整（actor_id 与 has_role 多参）、settings 私有字段与 AGENT_TYPES 四值不漂移。 4. tests/test_auth_adapter.py：适配器 11 例（映射/回退/fail-closed/trace 绑定/通配）。 基线：406 → 436 全绿（演示库已按演练 SOP §2 重灌：AML 8 条 / 演示 7 行 / sync 33 rows）。
98 lines
3.9 KiB
Python
98 lines
3.9 KiB
Python
"""宿主 AuthContext → 模块 AuthContext 适配器（S2 接缝，AL-09 接线用）。
|
||
|
||
状态：**预制件，当前未接线**。模块内部继续直接使用 `app.api.deps.get_auth_context`。
|
||
AL-09 合并 main 后，在模块 API 入口把宿主的 `AuthContext` 经 `from_host_auth` 转换即可，
|
||
模块其余代码零改动。
|
||
|
||
设计取舍：
|
||
- **不 import 宿主模块**（main 的 `app/model/schemas.py` 等），改用鸭子类型读取字段，
|
||
保证本文件在当前分支（宿主文件尚不存在）也能被导入与单测；
|
||
- 只做字段映射与语义补齐，不做任何权限判定（判定仍在 `deps.py` 的矩阵内）；
|
||
- 权限原样透传，另提供 `perm_matches` 兼容宿主的 `前缀:*` 通配语义，
|
||
避免模块侧 `has_permission` 精确匹配漏判通配权限。
|
||
|
||
字段映射（宿主 → 模块）：
|
||
sub → actor_id （字段名差异，模块全量代码依赖 actor_id）
|
||
trace_id → contextvar 绑定 （模块不存字段，走 utils.trace.current_trace）
|
||
agent_type → 不入模，仅返回 （模块按 agent_type 单独传参做准入判定）
|
||
roles/permissions/tenant_id/jti/customer_id → 原样透传
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from typing import Any
|
||
|
||
from app.api.deps import AuthContext
|
||
from app.utils.trace import set_trace
|
||
|
||
__all__ = ["from_host_auth", "perm_matches", "HostAuthAdapterError"]
|
||
|
||
|
||
class HostAuthAdapterError(ValueError):
|
||
"""宿主上下文字段缺失且无法安全兜底时抛出（fail-closed，不静默降级）。"""
|
||
|
||
|
||
def perm_matches(perm: str, permissions: list[str] | tuple[str, ...]) -> bool:
|
||
"""权限匹配：精确命中，或命中宿主的 `前缀:*` 通配写法。
|
||
|
||
例：`risk:alert:read` 在权限集含 `risk:*` 时返回 True。
|
||
"""
|
||
if not perm:
|
||
return False
|
||
if perm in permissions:
|
||
return True
|
||
prefix = perm.split(":", 1)[0]
|
||
return any(p == f"{prefix}:*" or p.endswith(":*") and p.startswith(f"{prefix}:") for p in permissions)
|
||
|
||
|
||
def from_host_auth(
|
||
host_ctx: Any,
|
||
*,
|
||
agent_type: str | None = None,
|
||
bind_trace: bool = True,
|
||
) -> AuthContext:
|
||
"""把宿主 AuthContext 转成模块 AuthContext。
|
||
|
||
Args:
|
||
host_ctx: 宿主的 `AuthContext`（含 sub/roles/token_type/... 的任意对象）。
|
||
agent_type: 显式指定请求目标 agent；缺省时取宿主的 `agent_type` 字段。
|
||
bind_trace: 是否把宿主的 `trace_id` 绑定到模块 trace contextvar。
|
||
|
||
Raises:
|
||
HostAuthAdapterError: 连主体标识（sub/actor_id）都取不到时抛出。
|
||
"""
|
||
actor_id = getattr(host_ctx, "sub", None) or getattr(host_ctx, "actor_id", None)
|
||
if not actor_id:
|
||
# fail-closed：拿不到主体就拒绝，不匿名放行
|
||
raise HostAuthAdapterError("宿主 AuthContext 缺少主体标识（sub/actor_id），拒绝放行")
|
||
|
||
roles = list(getattr(host_ctx, "roles", None) or [])
|
||
permissions = list(getattr(host_ctx, "permissions", None) or [])
|
||
token_type = getattr(host_ctx, "token_type", None) or "staff"
|
||
customer_id = getattr(host_ctx, "customer_id", None)
|
||
|
||
# customer 类 token 若未显式带 customer_id，回退为主体自身
|
||
if customer_id is None and token_type == "customer":
|
||
customer_id = actor_id
|
||
|
||
if bind_trace:
|
||
trace_id = getattr(host_ctx, "trace_id", None)
|
||
if trace_id:
|
||
set_trace(str(trace_id))
|
||
|
||
return AuthContext(
|
||
actor_id=str(actor_id),
|
||
roles=roles,
|
||
customer_id=customer_id,
|
||
token_type=str(token_type),
|
||
permissions=permissions,
|
||
tenant_id=getattr(host_ctx, "tenant_id", None),
|
||
jti=getattr(host_ctx, "jti", None),
|
||
)
|
||
|
||
|
||
def resolve_agent_type(host_ctx: Any, default: str = "risk") -> str:
|
||
"""取请求目标 agent 类型；宿主无该字段时回退 default。"""
|
||
value = getattr(host_ctx, "agent_type", None)
|
||
return str(value) if value else default
|