refractor: 拆分 rbac

Author: huan-yp

config.yaml

@@ -1,25 +1,25 @@
-root: '123456'
-bt_uin: '123456789'
-enable_webui_interaction: true
-debug: false
-github_proxy: null
-check_ncatbot_update: true
-skip_ncatbot_install_check: false
-websocket_timeout: 15
-napcat:
-  ws_uri: ws://localhost:16708
-  ws_token: napcat_ws
-  ws_listen_ip: localhost
-  webui_uri: http://localhost:6099
-  webui_token: (7Pyst9=}RO^^=|b
-  enable_webui: true
-  check_napcat_update: false
-  stop_napcat: false
-  remote_mode: false
-  report_self_message: false
-  report_forward_message_detail: true
-plugin:
-  plugins_dir: plugins
-  plugin_whitelist: []
-  plugin_blacklist: []
-  skip_plugin_load: true
+root: '123456'
+bt_uin: '123456789'
+enable_webui_interaction: true
+debug: false
+github_proxy: null
+check_ncatbot_update: true
+skip_ncatbot_install_check: false
+websocket_timeout: 15
+napcat:
+  ws_uri: ws://localhost:16708
+  ws_token: napcat_ws
+  ws_listen_ip: localhost
+  webui_uri: http://localhost:6099
+  webui_token: (7Pyst9=}RO^^=|b
+  enable_webui: true
+  check_napcat_update: false
+  stop_napcat: false
+  remote_mode: false
+  report_self_message: false
+  report_forward_message_detail: true
+plugin:
+  plugins_dir: plugins
+  plugin_whitelist: []
+  plugin_blacklist: []
+  skip_plugin_load: true

ncatbot/core/service/builtin/rbac/entity_manager.py

@@ -0,0 +1,201 @@
+"""
+RBAC 实体管理模块 - 处理权限、角色、用户的管理
+"""
+
+from typing import TYPE_CHECKING, List
+
+if TYPE_CHECKING:
+    from .service import RBACService
+
+
+class EntityManager:
+    """权限、角色、用户管理器"""
+
+    def __init__(self, service: "RBACService"):
+        self._service = service
+
+    # ==========================================================================
+    # 权限管理
+    # ==========================================================================
+
+    def add_permission(self, path: str) -> None:
+        """添加权限"""
+        if not self._service._permissions.exists(path, exact=True):
+            self._service._permissions.add(path)
+            self._service._clear_cache()
+
+    def remove_permission(self, path: str) -> None:
+        """移除权限"""
+        self._service._permissions.remove(path)
+        self._service._clear_cache()
+
+    def permission_exists(self, path: str) -> bool:
+        """检查权限是否存在"""
+        return self._service._permissions.exists(path, exact=True)
+
+    # ==========================================================================
+    # 角色管理
+    # ==========================================================================
+
+    def add_role(self, role: str, exist_ok: bool = False) -> None:
+        """添加角色"""
+        if role in self._service._roles:
+            if not exist_ok:
+                raise ValueError(f"角色 {role} 已存在")
+            return
+
+        self._service._roles[role] = {"whitelist": set(), "blacklist": set()}
+        self._service._role_users[role] = set()
+        self._service._role_inheritance[role] = []
+
+    def remove_role(self, role: str) -> None:
+        """移除角色"""
+        if role not in self._service._roles:
+            raise ValueError(f"角色 {role} 不存在")
+
+        # 从所有用户中移除该角色
+        for user_data in self._service._users.values():
+            if role in user_data["roles"]:
+                user_data["roles"].remove(role)
+
+        # 从继承链中移除
+        for parent_roles in self._service._role_inheritance.values():
+            if role in parent_roles:
+                parent_roles.remove(role)
+
+        del self._service._roles[role]
+        del self._service._role_users[role]
+        del self._service._role_inheritance[role]
+
+    def role_exists(self, role: str) -> bool:
+        """检查角色是否存在"""
+        return role in self._service._roles
+
+    def set_role_inheritance(self, role: str, parent: str) -> None:
+        """设置角色继承"""
+        if role not in self._service._roles:
+            raise ValueError(f"角色 {role} 不存在")
+        if parent not in self._service._roles:
+            raise ValueError(f"父角色 {parent} 不存在")
+        if role == parent:
+            raise ValueError("不能继承自身")
+        if self._would_create_cycle(role, parent):
+            raise ValueError(f"检测到循环继承: {role} -> {parent}")
+
+        if parent not in self._service._role_inheritance[role]:
+            self._service._role_inheritance[role].append(parent)
+            self._service._clear_cache()
+
+    def _would_create_cycle(self, role: str, new_parent: str) -> bool:
+        """检查是否会形成循环继承"""
+        visited = set()
+
+        def check(current: str) -> bool:
+            if current == role:
+                return True
+            if current in visited:
+                return False
+            visited.add(current)
+            for parent in self._service._role_inheritance.get(current, []):
+                if check(parent):
+                    return True
+            return False
+
+        return check(new_parent)
+
+    # ==========================================================================
+    # 用户管理
+    # ==========================================================================
+
+    def add_user(self, user: str, exist_ok: bool = False) -> None:
+        """添加用户"""
+        if user in self._service._users:
+            if not exist_ok:
+                raise ValueError(f"用户 {user} 已存在")
+            return
+
+        roles = [self._service._default_role] if self._service._default_role else []
+        self._service._users[user] = {"whitelist": set(), "blacklist": set(), "roles": roles}
+
+        # 添加到默认角色的用户列表
+        if self._service._default_role and self._service._default_role in self._service._role_users:
+            self._service._role_users[self._service._default_role].add(user)
+
+    def remove_user(self, user: str) -> None:
+        """移除用户"""
+        if user not in self._service._users:
+            raise ValueError(f"用户 {user} 不存在")
+
+        # 从所有角色的用户列表中移除
+        for role, users in self._service._role_users.items():
+            users.discard(user)
+
+        del self._service._users[user]
+        self._service._clear_cache()
+
+    def user_exists(self, user: str) -> bool:
+        """检查用户是否存在"""
+        return user in self._service._users
+
+    def user_has_role(self, user: str, role: str, create_user: bool = True) -> bool:
+        """检查用户是否拥有角色"""
+        if user not in self._service._users:
+            if create_user:
+                self.add_user(user, exist_ok=True)
+            else:
+                return False
+
+        return role in self._service._users[user]["roles"]
+
+    def assign_role(
+        self,
+        user: str,
+        role: str,
+        create_user: bool = True,
+    ) -> None:
+        """
+        分配角色给用户
+
+        Args:
+            user: 用户名
+            role: 角色名
+            create_user: 用户不存在时是否自动创建
+        """
+        if role not in self._service._roles:
+            raise ValueError(f"角色 {role} 不存在")
+
+        if user not in self._service._users:
+            if create_user:
+                self.add_user(user)
+            else:
+                raise ValueError(f"用户 {user} 不存在")
+
+        user_data = self._service._users[user]
+        if role not in user_data["roles"]:
+            user_data["roles"].append(role)
+
+        if user not in self._service._role_users[role]:
+            self._service._role_users[role].add(user)
+
+        self._service._clear_cache()
+
+    def unassign_role(self, user: str, role: str) -> None:
+        """
+        从用户移除角色
+
+        Args:
+            user: 用户名
+            role: 角色名
+        """
+        if user not in self._service._users:
+            raise ValueError(f"用户 {user} 不存在")
+
+        if role not in self._service._roles:
+            raise ValueError(f"角色 {role} 不存在")
+
+        user_data = self._service._users[user]
+        if role in user_data["roles"]:
+            user_data["roles"].remove(role)
+
+        self._service._role_users[role].discard(user)
+        self._service._clear_cache()

ncatbot/core/service/builtin/rbac/permission_assigner.py

@@ -0,0 +1,72 @@
+"""
+RBAC 权限分配模块
+"""
+
+from typing import Dict, Literal
+
+
+class PermissionAssigner:
+    """权限分配逻辑的抽象"""
+
+    def __init__(self, service):
+        self.service = service
+
+    def grant(
+        self,
+        target_type: Literal["user", "role"],
+        target: str,
+        permission: str,
+        mode: Literal["white", "black"] = "white",
+        create_permission: bool = True,
+    ) -> None:
+        """
+        授予权限
+
+        Args:
+            target_type: 目标类型 ("user" 或 "role")
+            target: 用户名或角色名
+            permission: 权限路径
+            mode: white=白名单, black=黑名单
+            create_permission: 权限不存在时是否自动创建
+        """
+        if not self.service.permission_exists(permission):
+            if create_permission:
+                self.service.add_permission(permission)
+            else:
+                raise ValueError(f"权限 {permission} 不存在")
+
+        if target_type == "user":
+            if target not in self.service._users:
+                raise ValueError(f"用户 {target} 不存在")
+            target_data = self.service._users[target]
+        else:
+            if target not in self.service._roles:
+                raise ValueError(f"角色 {target} 不存在")
+            target_data = self.service._roles[target]
+
+        list_key = "whitelist" if mode == "white" else "blacklist"
+        opposite_key = "blacklist" if mode == "white" else "whitelist"
+
+        target_data[list_key].add(permission)
+        target_data[opposite_key].discard(permission)  # 从相反列表移除
+        self.service._clear_cache()
+
+    def revoke(
+        self,
+        target_type: Literal["user", "role"],
+        target: str,
+        permission: str,
+    ) -> None:
+        """撤销权限"""
+        if target_type == "user":
+            if target not in self.service._users:
+                raise ValueError(f"用户 {target} 不存在")
+            target_data = self.service._users[target]
+        else:
+            if target not in self.service._roles:
+                raise ValueError(f"角色 {target} 不存在")
+            target_data = self.service._roles[target]
+
+        target_data["whitelist"].discard(permission)
+        target_data["blacklist"].discard(permission)
+        self.service._clear_cache()