diff --git a/core/template-console/__init__.py b/core/template-console/__init__.py new file mode 100644 index 0000000..ca1afec --- /dev/null +++ b/core/template-console/__init__.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +"""⑤.7 模板配置台（Template Console）内核引擎 —— EPIC #9。 + +配置台是**跨模板通用的内核能力**：为实施工程师提供一个无代码的配置驱动 +界面，把"模型超参 / RAG / 布局"三类配置 + 点位字典 + 版本发布统一编排， +并将发布的配置**推送给内核**（edge-gateway / rag-kb / model-framework）。 + +本包拆为 6 个子模块，对应 6 个 issue（同一 feature 分支承载，单 PR 关联）： + +- ``rbac`` (#62) 三级 RBAC 权限（admin / engineer / readonly）； +- ``point_importer`` (#63) 点位字典 CSV 导入 + 自动校验页面（复用 + ``core/edge-gateway/point_dict`` 校验器，增加配置台 + 级结果聚合 + OPC 节点格式校验 + 模板选择）； +- ``config_store`` (#64) 配置项 CRUD（模型超参 / RAG / 布局三类，文件系统 + 版本化 JSON 存储）； +- ``preview`` (#65) 预览渲染引擎（布局/告警/查询 → 可预览结构化输出， + 对齐 iAOP-cockpit-layout-v1 widget 类型）； +- ``release`` (#66) 版本发布 + 回滚点（基于 config_store 快照，semver）； +- ``push_channel`` (#67) 配置台↔内核配置推送契约（JSON manifest + 校验和 + + 幂等性）。 + +设计原则（对齐 PRD「可解释可溯源」与既有内核范式）： +- 纯标准库零运行时依赖（无 pyyaml/numpy/pandas），YAML 子集用内置解析器； +- dataclass + Enum + 类型注解 + 中文 docstring； +- 关键决策均带 ``meaning`` / ``reason`` 字段，便于审计与可解释性。 +""" +from __future__ import annotations + +from .rbac import ( + Action, + Permission, + Role, + RoleKind, + User, + has_permission, +) + +__all__ = [ + "Action", + "Permission", + "Role", + "RoleKind", + "User", + "has_permission", +] + +#: 本包版本（对齐 EPIC #9 模板配置台交付节奏） +__version__ = "1.0.0" diff --git a/core/template-console/rbac.py b/core/template-console/rbac.py new file mode 100644 index 0000000..9ee3749 --- /dev/null +++ b/core/template-console/rbac.py @@ -0,0 +1,310 @@ +# -*- 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], + } diff --git a/core/template-console/tests/_bootstrap.py b/core/template-console/tests/_bootstrap.py new file mode 100644 index 0000000..bcff38b --- /dev/null +++ b/core/template-console/tests/_bootstrap.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +"""测试引导：把 `core/template-console` 以包名 `template_console` 挂载到 sys.modules。 + +目录名 `template-console` 含连字符，无法直接以包名 import；挂载后模块内 +相对导入（`from .rbac import ...`）在 unittest 发现机制下可正常解析。 + +同时把兄弟内核目录 `core/edge-gateway` 加入 sys.path，使 point_importer +可复用其 `point_dict` 子包（loader/validator/schema），避免重复造轮子。 +""" +import os +import sys +import types + +# 1) 挂载 core/template-console 为 template_console 包 +CONSOLE_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +sys.path.insert(0, CONSOLE_DIR) +if "template_console" not in sys.modules: + pkg = types.ModuleType("template_console") + pkg.__path__ = [CONSOLE_DIR] + sys.modules["template_console"] = pkg + +# 2) 暴露兄弟内核 edge-gateway/point_dict（#63 复用其校验器） +CORE_DIR = os.path.dirname(CONSOLE_DIR) +EDGE_GW_DIR = os.path.join(CORE_DIR, "edge-gateway") +if os.path.isdir(EDGE_GW_DIR) and EDGE_GW_DIR not in sys.path: + sys.path.insert(0, EDGE_GW_DIR) diff --git a/core/template-console/tests/test_rbac.py b/core/template-console/tests/test_rbac.py new file mode 100644 index 0000000..4a26e4c --- /dev/null +++ b/core/template-console/tests/test_rbac.py @@ -0,0 +1,181 @@ +# -*- coding: utf-8 -*- +"""三级 RBAC 权限模型测试（issue #62）。 + +覆盖： +1. 三级角色权限矩阵正确（readonly/engineer/admin）； +2. 角色继承（admin 继承 engineer 继承 readonly）； +3. has_permission 允许/拒绝判定 + 理由可解释； +4. 细粒度收窄（restricted_to_readonly 把写动作降级拒绝）； +5. 便捷判定 can_publish / 用户概览。 +""" +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 + +from template_console.rbac import ( # noqa: E402 + Action, + Permission, + Resource, + Role, + RoleKind, + User, + all_roles, + can_publish, + effective_permissions, + get_role, + has_permission, + user_summary, +) + + +class RoleRegistryTest(unittest.TestCase): + """三级角色注册表。""" + + def test_three_roles_present(self): + roles = {r.kind for r in all_roles()} + self.assertEqual(roles, {RoleKind.READONLY, RoleKind.ENGINEER, RoleKind.ADMIN}) + + def test_role_labels_in_chinese(self): + self.assertEqual(get_role(RoleKind.READONLY).label, "只读") + self.assertEqual(get_role(RoleKind.ENGINEER).label, "行业工程师") + self.assertEqual(get_role(RoleKind.ADMIN).label, "管理员") + + def test_role_descriptions_explainable(self): + # 可解释性：每个角色都有职责说明 + for role in all_roles(): + self.assertTrue(role.description, f"{role.kind} 缺少 description") + + def test_inheritance_chain(self): + self.assertEqual(get_role(RoleKind.ADMIN).inherits, RoleKind.ENGINEER) + self.assertEqual(get_role(RoleKind.ENGINEER).inherits, RoleKind.READONLY) + self.assertIsNone(get_role(RoleKind.READONLY).inherits) + + def test_permission_key_format(self): + p = Permission(RoleKind.ADMIN, Resource.USER, Action.MANAGE) + # 匹配键为资源:动作（角色无关，便于继承）；审计键含授予角色 + self.assertEqual(p.key(), "user:manage") + self.assertEqual(p.audit_key(), "admin:user:manage") + + +class EffectivePermissionTest(unittest.TestCase): + """继承后的有效权限集合。""" + + def test_admin_inherits_engineer_and_readonly(self): + eff = effective_permissions(RoleKind.ADMIN) + # 匹配键为 resource:action：admin 拥有自身的 user:manage， + # 也继承 engineer 的 model_param:edit 与 readonly 的 layout:view + self.assertIn("user:manage", eff) + self.assertIn("model_param:edit", eff) + self.assertIn("layout:view", eff) + + def test_engineer_cannot_publish(self): + eff = effective_permissions(RoleKind.ENGINEER) + # 工程师不能发布/推送/管用户 + self.assertNotIn("release:publish", eff) + self.assertNotIn("push:publish", eff) + self.assertNotIn("user:manage", eff) + + def test_readonly_has_no_write(self): + eff = effective_permissions(RoleKind.READONLY) + for key in eff: + # 只读权限只能以 :view 结尾 + self.assertTrue(key.endswith(":view"), f"readonly 不应有写/发布权限: {key}") + + +class HasPermissionTest(unittest.TestCase): + """has_permission 判定 + 理由。""" + + def setUp(self): + self.ro = User("viewer", RoleKind.READONLY, "查看员") + self.eng = User("li_engineer", RoleKind.ENGINEER, "李工") + self.admin = User("root_admin", RoleKind.ADMIN, "管理员甲") + + def test_readonly_view_allowed(self): + d = has_permission(self.ro, Resource.LAYOUT, Action.VIEW) + self.assertTrue(d.allow) + self.assertEqual(d.source, "explicit") + + def test_readonly_edit_denied(self): + d = has_permission(self.ro, Resource.LAYOUT, Action.EDIT) + self.assertFalse(d.allow) + self.assertIn("缺少权限", d.reason) + + def test_engineer_edit_allowed_inherited_view(self): + # 工程师编辑是本角色权限（explicit） + d_edit = has_permission(self.eng, Resource.LAYOUT, Action.EDIT) + self.assertTrue(d_edit.allow) + self.assertEqual(d_edit.source, "explicit") + # 工程师查看布局是继承自 readonly（inherited） + d_view = has_permission(self.eng, Resource.LAYOUT, Action.VIEW) + self.assertTrue(d_view.allow) + self.assertEqual(d_view.source, "inherited") + + def test_engineer_publish_denied(self): + d = has_permission(self.eng, Resource.RELEASE, Action.PUBLISH) + self.assertFalse(d.allow) + + def test_admin_publish_allowed(self): + d = has_permission(self.admin, Resource.RELEASE, Action.PUBLISH) + self.assertTrue(d.allow) + self.assertEqual(d.source, "explicit") + + def test_admin_inherited_engineer_edit(self): + d = has_permission(self.admin, Resource.MODEL_PARAM, Action.EDIT) + self.assertTrue(d.allow) + self.assertEqual(d.source, "inherited") + + def test_decision_carries_reason(self): + # 可解释性：无论允许/拒绝，reason 非空且含用户名与资源 + for user in (self.ro, self.eng, self.admin): + d = has_permission(user, Resource.PUSH, Action.PUBLISH) + self.assertIn(user.username, d.reason) + self.assertIn(Resource.PUSH.value, d.reason) + + +class RestrictedUserTest(unittest.TestCase): + """细粒度收窄：restricted_to_readonly。""" + + def test_restricted_engineer_cannot_edit_that_resource(self): + # 工程师本可编辑布局，但被收窄为只读后应拒绝 + u = User("limited", RoleKind.ENGINEER, "受限工程师", + restricted_to_readonly=[Resource.LAYOUT]) + d = has_permission(u, Resource.LAYOUT, Action.EDIT) + self.assertFalse(d.allow) + self.assertEqual(d.source, "restricted") + + def test_restricted_engineer_can_still_view(self): + u = User("limited", RoleKind.ENGINEER, "受限工程师", + restricted_to_readonly=[Resource.LAYOUT]) + d = has_permission(u, Resource.LAYOUT, Action.VIEW) + self.assertTrue(d.allow) + + def test_restricted_only_affects_named_resource(self): + u = User("limited", RoleKind.ENGINEER, "受限工程师", + restricted_to_readonly=[Resource.LAYOUT]) + # 模型超参未被收窄，仍可编辑 + d = has_permission(u, Resource.MODEL_PARAM, Action.EDIT) + self.assertTrue(d.allow) + + +class ConvenienceTest(unittest.TestCase): + """便捷判定与用户概览。""" + + def test_can_publish(self): + self.assertFalse(can_publish(User("v", RoleKind.READONLY))) + self.assertFalse(can_publish(User("e", RoleKind.ENGINEER))) + self.assertTrue(can_publish(User("a", RoleKind.ADMIN))) + + def test_user_summary(self): + s = user_summary(User("li", RoleKind.ENGINEER, "李工")) + self.assertEqual(s["username"], "li") + self.assertEqual(s["role"], "engineer") + self.assertEqual(s["role_label"], "行业工程师") + self.assertGreater(s["effective_permission_count"], 0) + self.assertEqual(s["restricted_to_readonly"], []) + + +if __name__ == "__main__": + unittest.main()