311 lines
13 KiB
Python
311 lines
13 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""⑤.7 配置台三级 RBAC 权限模型 —— issue #62 / PRD ⑤.7。
|
||
|
||
配置台面向**多角色协作**：实施工程师配模板，行业工程师调参数，运维/管理者
|
||
发布上线。直接对所有人开放写权限会带来误改与不可溯源风险。本模块用三级
|
||
RBAC（基于角色的访问控制）锁定"谁能对哪类配置做什么"，并把每次权限判定
|
||
的**理由**一并返回，对齐 PRD「可解释可溯源」。
|
||
|
||
三级角色（由低到高，后者继承前者全部权限）：
|
||
|
||
- ``readonly`` （只读）：查看配置 / 预览 / 历史版本，不可写；
|
||
- ``engineer`` （行业工程师）：只读权限 + 编辑/校验/导入配置（模型超参 /
|
||
RAG / 布局 / 点位字典），但**不能发布与回滚**；
|
||
- ``admin`` （管理员）：工程师权限 + 发布 / 回滚 / 推送内核 / 用户管理。
|
||
|
||
权限判定核心为 ``has_permission(user, resource, action)``，返回
|
||
``PermissionDecision``（allow + reason），便于配置台前端把"为什么拒绝"
|
||
直接展示给操作者，而不是一个干瘪的 403。
|
||
|
||
零运行时依赖：仅用 dataclass / Enum / 标准库。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from enum import Enum
|
||
from typing import Dict, List, Optional, Set
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 权限维度：资源 × 动作
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class Resource(str, Enum):
|
||
"""配置台可管控的资源（对齐 #63~#67 子模块）。"""
|
||
|
||
POINT_DICT = "point_dict" # 点位字典（#63）
|
||
MODEL_PARAM = "model_param" # 模型超参配置（#64）
|
||
RAG_CONFIG = "rag_config" # RAG 知识库配置（#64）
|
||
LAYOUT = "layout" # 驾驶舱布局配置（#64/#65）
|
||
PREVIEW = "preview" # 预览（#65）
|
||
RELEASE = "release" # 版本发布/回滚（#66）
|
||
PUSH = "push" # 配置推送内核（#67）
|
||
USER = "user" # 用户/角色管理
|
||
|
||
|
||
class Action(str, Enum):
|
||
"""对资源可执行的动作。"""
|
||
|
||
VIEW = "view" # 查看 / 预览 / 列表
|
||
EDIT = "edit" # 新增 / 修改 / 删除 / 导入 / 校验
|
||
PUBLISH = "publish" # 发布版本 / 回滚 / 推送内核
|
||
MANAGE = "manage" # 用户与角色管理
|
||
|
||
|
||
class RoleKind(str, Enum):
|
||
"""三级角色枚举（值即配置资产中的角色标识）。"""
|
||
|
||
READONLY = "readonly"
|
||
ENGINEER = "engineer"
|
||
ADMIN = "admin"
|
||
|
||
|
||
# 各资源的「写」动作等价集合：EDIT 含新增/修改/删除/导入/校验。
|
||
# PUBLISH 含发布/回滚/推送。这样配置台前端只需关心粗粒度动作。
|
||
_WRITE_ACTIONS: Set[Action] = {Action.EDIT, Action.PUBLISH, Action.MANAGE}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 权限模型
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass(frozen=True)
|
||
class Permission:
|
||
"""一条权限授予（角色 → 资源 → 动作）。
|
||
|
||
``meaning`` 解释该权限的业务含义，用于审计日志与配置台权限矩阵展示。
|
||
注意：权限**匹配**基于 ``resource:action``（资源×动作），与授予角色无关——
|
||
这正是角色继承能生效的关键（admin 继承 engineer 的 edit，匹配键相同）。
|
||
``role`` 仅作为审计元数据，记录"是谁授予的"。
|
||
"""
|
||
|
||
role: RoleKind
|
||
resource: Resource
|
||
action: Action
|
||
meaning: str = ""
|
||
|
||
def key(self) -> str:
|
||
"""权限匹配键（资源:动作）—— 角色继承据此累计。"""
|
||
return f"{self.resource.value}:{self.action.value}"
|
||
|
||
def audit_key(self) -> str:
|
||
"""审计唯一键（角色/资源/动作三元组，含授予者）。"""
|
||
return f"{self.role.value}:{self.resource.value}:{self.action.value}"
|
||
|
||
|
||
@dataclass
|
||
class Role:
|
||
"""一个角色：权限集合 + 继承的父角色。"""
|
||
|
||
kind: RoleKind
|
||
label: str # 中文展示名
|
||
permissions: List[Permission] = field(default_factory=list)
|
||
inherits: Optional[RoleKind] = None # 继承的低一级角色
|
||
description: str = "" # 角色职责说明（可解释性）
|
||
|
||
def permission_keys(self) -> Set[str]:
|
||
"""本角色直接授予的权限键集合。"""
|
||
return {p.key() for p in self.permissions}
|
||
|
||
|
||
@dataclass
|
||
class User:
|
||
"""配置台用户。"""
|
||
|
||
username: str
|
||
role: RoleKind
|
||
display_name: str = ""
|
||
# 可选资源级收窄：即便角色允许，列表中的资源也会被额外限制为只读。
|
||
# 用于"只允许工程师改某几类配置"的细粒度场景。
|
||
restricted_to_readonly: List[Resource] = field(default_factory=list)
|
||
|
||
|
||
@dataclass
|
||
class PermissionDecision:
|
||
"""``has_permission`` 的判定结果（带理由，可解释）。"""
|
||
|
||
allow: bool
|
||
reason: str # 人类可读的判定理由（允许/拒绝原因）
|
||
role: RoleKind
|
||
resource: Resource
|
||
action: Action
|
||
source: str = "explicit" # explicit（本角色直接授予）/ inherited（继承自父角色）
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 角色注册表：三级权限矩阵（对齐 PRD ⑤.7「三级 RBAC」）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _build_role_registry() -> Dict[RoleKind, Role]:
|
||
"""构建三级角色及其权限矩阵。
|
||
|
||
权限设计依据（PRD ⑤.7）：
|
||
- readonly：可查看所有配置/预览/历史，但不能改、不能发；
|
||
- engineer：在 readonly 基础上，可编辑/校验/导入四类业务配置，
|
||
但**发布/回滚/推送/用户管理仍归 admin**（避免未经评审上线）；
|
||
- admin：在 engineer 基础上，可发布/回滚/推送 + 管理用户角色。
|
||
"""
|
||
ro = Role(
|
||
kind=RoleKind.READONLY,
|
||
label="只读",
|
||
description="实施/运维只读角色：查看配置、预览、历史版本，不可写。",
|
||
permissions=[
|
||
Permission(RoleKind.READONLY, Resource.POINT_DICT, Action.VIEW,
|
||
"查看点位字典与校验报告"),
|
||
Permission(RoleKind.READONLY, Resource.MODEL_PARAM, Action.VIEW,
|
||
"查看模型超参配置"),
|
||
Permission(RoleKind.READONLY, Resource.RAG_CONFIG, Action.VIEW,
|
||
"查看 RAG 知识库配置"),
|
||
Permission(RoleKind.READONLY, Resource.LAYOUT, Action.VIEW,
|
||
"查看驾驶舱布局配置"),
|
||
Permission(RoleKind.READONLY, Resource.PREVIEW, Action.VIEW,
|
||
"查看配置预览"),
|
||
Permission(RoleKind.READONLY, Resource.RELEASE, Action.VIEW,
|
||
"查看历史发布版本"),
|
||
],
|
||
)
|
||
|
||
engineer = Role(
|
||
kind=RoleKind.ENGINEER,
|
||
label="行业工程师",
|
||
inherits=RoleKind.READONLY,
|
||
description="行业工程师：编辑/校验/导入业务配置，但不能发布与推送。",
|
||
permissions=[
|
||
Permission(RoleKind.ENGINEER, Resource.POINT_DICT, Action.EDIT,
|
||
"导入/编辑/校验点位字典 CSV"),
|
||
Permission(RoleKind.ENGINEER, Resource.MODEL_PARAM, Action.EDIT,
|
||
"调整模型超参配置"),
|
||
Permission(RoleKind.ENGINEER, Resource.RAG_CONFIG, Action.EDIT,
|
||
"编辑 RAG 知识库配置"),
|
||
Permission(RoleKind.ENGINEER, Resource.LAYOUT, Action.EDIT,
|
||
"编辑驾驶舱布局配置"),
|
||
Permission(RoleKind.ENGINEER, Resource.PREVIEW, Action.VIEW,
|
||
"预览配置效果（编辑后必看）"),
|
||
],
|
||
)
|
||
|
||
admin = Role(
|
||
kind=RoleKind.ADMIN,
|
||
label="管理员",
|
||
inherits=RoleKind.ENGINEER,
|
||
description="管理员：在工程师基础上负责发布/回滚/推送与用户管理。",
|
||
permissions=[
|
||
Permission(RoleKind.ADMIN, Resource.RELEASE, Action.PUBLISH,
|
||
"发布新版本与回滚到历史版本"),
|
||
Permission(RoleKind.ADMIN, Resource.PUSH, Action.PUBLISH,
|
||
"把已发布配置推送给内核"),
|
||
Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE,
|
||
"管理用户与角色分配"),
|
||
Permission(RoleKind.ADMIN, Resource.POINT_DICT, Action.PUBLISH,
|
||
"确认点位字典上线（审批环节）"),
|
||
Permission(RoleKind.ADMIN, Resource.MODEL_PARAM, Action.PUBLISH,
|
||
"确认模型超参上线"),
|
||
Permission(RoleKind.ADMIN, Resource.LAYOUT, Action.PUBLISH,
|
||
"确认布局上线"),
|
||
],
|
||
)
|
||
|
||
return {RoleKind.READONLY: ro, RoleKind.ENGINEER: engineer, RoleKind.ADMIN: admin}
|
||
|
||
|
||
_ROLES: Dict[RoleKind, Role] = _build_role_registry()
|
||
|
||
|
||
def get_role(kind: RoleKind) -> Role:
|
||
"""获取角色定义。"""
|
||
return _ROLES[kind]
|
||
|
||
|
||
def all_roles() -> List[Role]:
|
||
"""全部角色（按权限由低到高）。"""
|
||
return [_ROLES[RoleKind.READONLY], _ROLES[RoleKind.ENGINEER], _ROLES[RoleKind.ADMIN]]
|
||
|
||
|
||
def effective_permissions(kind: RoleKind) -> Set[str]:
|
||
"""角色有效权限键（含继承链）。
|
||
|
||
继承解析：admin 继承 engineer 继承 readonly，递归向上累计权限键。
|
||
"""
|
||
role = _ROLES[kind]
|
||
keys: Set[str] = set(role.permission_keys())
|
||
if role.inherits is not None:
|
||
keys |= effective_permissions(role.inherits)
|
||
return keys
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 判定 API
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def has_permission(
|
||
user: User,
|
||
resource: Resource,
|
||
action: Action,
|
||
) -> PermissionDecision:
|
||
"""判定用户对某资源执行某动作是否被允许（带理由）。
|
||
|
||
判定顺序：
|
||
1. 计算角色有效权限（含继承），命中即允许并标注来源（本角色/继承）；
|
||
2. 命中后若该资源在用户 ``restricted_to_readonly`` 列表且动作是写动作，
|
||
则降级拒绝（细粒度收窄）；
|
||
3. 未命中则拒绝，理由标注缺失的权限三元组。
|
||
|
||
Args:
|
||
user: 配置台用户；
|
||
resource: 目标资源；
|
||
action: 目标动作。
|
||
|
||
Returns:
|
||
PermissionDecision：allow + reason（可直接展示给操作者）。
|
||
"""
|
||
target = f"{resource.value}:{action.value}"
|
||
eff = effective_permissions(user.role)
|
||
|
||
# 细粒度收窄：即便角色允许，特定资源也被限制为只读
|
||
if resource in user.restricted_to_readonly and action in _WRITE_ACTIONS:
|
||
return PermissionDecision(
|
||
allow=False,
|
||
reason=(f"用户 '{user.username}' 对资源 '{resource.value}' 被收窄为只读，"
|
||
f"禁止执行 '{action.value}' 动作"),
|
||
role=user.role, resource=resource, action=action, source="restricted",
|
||
)
|
||
|
||
if target in eff:
|
||
# 判定来源：本角色直接授予 or 继承自父角色
|
||
own = get_role(user.role).permission_keys()
|
||
source = "explicit" if target in own else "inherited"
|
||
src_label = "本角色直接授予" if source == "explicit" else "继承自低级角色"
|
||
return PermissionDecision(
|
||
allow=True,
|
||
reason=(f"用户 '{user.username}'（{get_role(user.role).label}）"
|
||
f"允许对 '{resource.value}' 执行 '{action.value}'（{src_label}）"),
|
||
role=user.role, resource=resource, action=action, source=source,
|
||
)
|
||
|
||
return PermissionDecision(
|
||
allow=False,
|
||
reason=(f"用户 '{user.username}'（{get_role(user.role).label}）缺少权限 "
|
||
f"{user.role.value}:{resource.value}:{action.value}；"
|
||
f"该动作需更高角色或审批"),
|
||
role=user.role, resource=resource, action=action, source="denied",
|
||
)
|
||
|
||
|
||
def can_publish(user: User) -> bool:
|
||
"""便捷判定：用户是否具备发布（发布/回滚/推送）能力。"""
|
||
return has_permission(user, Resource.RELEASE, Action.PUBLISH).allow
|
||
|
||
|
||
def user_summary(user: User) -> Dict[str, object]:
|
||
"""用户权限概览（供配置台用户卡片/审计日志展示）。"""
|
||
role = get_role(user.role)
|
||
return {
|
||
"username": user.username,
|
||
"display_name": user.display_name or user.username,
|
||
"role": user.role.value,
|
||
"role_label": role.label,
|
||
"description": role.description,
|
||
"effective_permission_count": len(effective_permissions(user.role)),
|
||
"restricted_to_readonly": [r.value for r in user.restricted_to_readonly],
|
||
}
|