Fix/command permission management

astrbot/dashboard/routes/__init__.py

@@ -10,6 +10,7 @@
 from .conversation import ConversationRoute
 from .file import FileRoute
 from .session_management import SessionManagementRoute
+from .command_permission import CommandPermissionRoute
 
 
 __all__ = [
@@ -25,4 +26,5 @@
     "ConversationRoute",
     "FileRoute",
     "SessionManagementRoute",
+    "CommandPermissionRoute",
 ]

astrbot/dashboard/routes/command_permission.py

@@ -0,0 +1,235 @@
+import traceback
+import time
+from typing import List, Dict, Any
+
+from .route import Route, Response, RouteContext
+from astrbot.core import logger
+from quart import request
+from astrbot.core.core_lifecycle import AstrBotCoreLifecycle
+from astrbot.core.star.star_handler import star_handlers_registry, StarHandlerMetadata
+from astrbot.core.star.filter.command import CommandFilter
+from astrbot.core.star.filter.command_group import CommandGroupFilter
+from astrbot.core.star.filter.permission import PermissionTypeFilter
+from astrbot.core.star.star import star_map
+from astrbot.core import DEMO_MODE
+from astrbot.core.utils.shared_preferences import SharedPreferences
+
+
+class CommandPermissionRoute(Route):
+    def __init__(
+        self,
+        context: RouteContext,
+        core_lifecycle: AstrBotCoreLifecycle,
+    ) -> None:
+        super().__init__(context)
+        self.routes = {
+            "/command_permission/get": ("GET", self.get_command_permissions),
+            "/command_permission/set": ("POST", self.set_command_permission),
+            "/command_permission/get_commands": ("GET", self.get_all_commands),
+        }
+        self.core_lifecycle = core_lifecycle
+        self._commands_cache = None
+        self._cache_timestamp = 0
+        self._cache_ttl = 60  # 缓存60秒
+        self.register_routes()
+
+    async def get_command_permissions(self) -> Response:
+        """获取所有指令的权限配置"""
+        try:
+            sp = SharedPreferences()
+            alter_cmd_cfg = sp.get("alter_cmd", {})
+
+            # 构建权限配置列表
+            permissions = []
+
+            # 遍历所有插件的权限配置
+            for plugin_name, plugin_config in alter_cmd_cfg.items():
+                for command_name, config in plugin_config.items():
+                    permission_type = config.get("permission", "member")
+                    permissions.append({
+                        "plugin_name": plugin_name,
+                        "command_name": command_name,
+                        "permission": permission_type,
+                        "id": f"{plugin_name}.{command_name}"
+                    })
+
+            return Response().ok({"permissions": permissions}).__dict__
+        except Exception:
+            logger.error(f"/api/command_permission/get: {traceback.format_exc()}")
+            return Response().error("获取指令权限配置失败").__dict__
+
+    def _get_permission_info(self, handler: StarHandlerMetadata, plugin) -> tuple[str, bool]:
+        """提取权限信息的辅助函数"""
+        sp = SharedPreferences()
+        alter_cmd_cfg = sp.get("alter_cmd", {})
+        current_permission = "member"  # 默认权限
+
+        if (plugin.name in alter_cmd_cfg and 
+            handler.handler_name in alter_cmd_cfg[plugin.name]):
+            current_permission = alter_cmd_cfg[plugin.name][handler.handler_name].get("permission", "member")
+
+        # 检查是否有默认的权限过滤器
+        has_default_admin = False
+        for f in handler.event_filters:
+            if isinstance(f, PermissionTypeFilter):
+                if f.permission_type.name == "ADMIN":
+                    has_default_admin = True
+                    if current_permission == "member":
+                        current_permission = "admin"
+                break
+
+        return current_permission, has_default_admin
+
+    def _is_cache_valid(self) -> bool:
+        """检查缓存是否有效"""
+        return (self._commands_cache is not None and 
+                time.time() - self._cache_timestamp < self._cache_ttl)
+
+    def _invalidate_cache(self):
+        """清空缓存"""
+        self._commands_cache = None
+        self._cache_timestamp = 0
+
+    def _build_commands_list(self) -> list:
+        """构建指令列表（无缓存）"""
+        commands = []
+
+        # 遍历所有注册的处理器
+        for handler in star_handlers_registry:
+            if not isinstance(handler, StarHandlerMetadata):
+                logger.warning(f"Handler {handler} is not an instance of StarHandlerMetadata, skipping.")
+                continue
+            plugin = star_map.get(handler.handler_module_path)
+
+            if not plugin:
+                continue
+
+            # 查找指令过滤器
+            for filter_ in handler.event_filters:
+                if isinstance(filter_, CommandFilter):
+                    current_permission, has_default_admin = self._get_permission_info(handler, plugin)
+
+                    commands.append({
+                        "command_name": filter_.command_name,
+                        "plugin_name": plugin.name,
+                        "handler_name": handler.handler_name,
+                        "description": getattr(handler, 'description', ''),
+                        "current_permission": current_permission,
+                        "has_default_admin": has_default_admin,
+                        "id": f"{plugin.name}.{handler.handler_name}"
+                    })
+                elif isinstance(filter_, CommandGroupFilter):
+                    current_permission, has_default_admin = self._get_permission_info(handler, plugin)
+
+                    commands.append({
+                        "command_name": filter_.group_name,
+                        "plugin_name": plugin.name,
+                        "handler_name": handler.handler_name,
+                        "description": getattr(handler, 'description', ''),
+                        "current_permission": current_permission,
+                        "has_default_admin": has_default_admin,
+                        "is_group": True,
+                        "id": f"{plugin.name}.{handler.handler_name}"
+                    })
+
+        return commands
+
+    async def get_all_commands(self) -> Response:
+        """获取所有可用的指令列表"""
+        try:
+            # 检查缓存
+            if self._is_cache_valid():
+                commands = self._commands_cache
+            else:
+                # 重新构建并缓存
+                commands = self._build_commands_list()
+                self._commands_cache = commands
+                self._cache_timestamp = time.time()
+
+            return Response().ok({"commands": commands}).__dict__
+        except Exception:
+            logger.error(f"/api/command_permission/get_commands: {traceback.format_exc()}")
+            return Response().error("获取指令列表失败").__dict__
+
+    async def set_command_permission(self) -> Response:
+        """设置指令权限"""
+        if DEMO_MODE:
+            return Response().error("演示模式下不允许修改配置").__dict__
+
+        try:
+            data = await request.get_json()
+            plugin_name = data.get("plugin_name")
+            handler_name = data.get("handler_name")
+            permission = data.get("permission")
+
+            missing_params = []
+            if not plugin_name:
+                missing_params.append("plugin_name")
+            if not handler_name:
+                missing_params.append("handler_name")
+            if not permission:
+                missing_params.append("permission")
+            if missing_params:
+                return Response().error(f"参数不完整，缺少: {', '.join(missing_params)}").__dict__
+
+            normalized_permission = permission.lower()
+            if normalized_permission not in ["admin", "member"]:
+                return Response().error("权限类型错误，只能是 admin 或 member").__dict__
+
+            # 使用规范化后的权限值
+            permission = normalized_permission
+
+            # 查找对应的处理器
+            found_handler = None
+            for handler in star_handlers_registry:
+                if (handler.handler_module_path in star_map and
+                    star_map[handler.handler_module_path].name == plugin_name and
+                    handler.handler_name == handler_name):
+                    found_handler = handler
+                    break
+
+            if not found_handler:
+                return Response().error("未找到指定的指令处理器").__dict__
+
+            # 更新配置
+            sp = SharedPreferences()
+            alter_cmd_cfg = sp.get("alter_cmd", {})
+
+            if plugin_name not in alter_cmd_cfg:
+                alter_cmd_cfg[plugin_name] = {}
+
+            if handler_name not in alter_cmd_cfg[plugin_name]:
+                alter_cmd_cfg[plugin_name][handler_name] = {}
+
+            alter_cmd_cfg[plugin_name][handler_name]["permission"] = permission
+            sp.put("alter_cmd", alter_cmd_cfg)
+
+            # 清空缓存，因为权限配置已经改变
+            self._invalidate_cache()
+
+            # 动态更新权限过滤器
+            from astrbot.core.star.filter.permission import PermissionType
+            found_permission_filter = False
+
+            # 更新第一个 PermissionTypeFilter（与项目其他部分保持一致）
+            for filter_ in found_handler.event_filters:
+                if isinstance(filter_, PermissionTypeFilter):
+                    if permission == "admin":
+                        filter_.permission_type = PermissionType.ADMIN
+                    else:
+                        filter_.permission_type = PermissionType.MEMBER
+                    found_permission_filter = True
+                    break
+
+            if not found_permission_filter:
+                # 如果没有权限过滤器，则添加一个
+                new_filter = PermissionTypeFilter(
+                    PermissionType.ADMIN if permission == "admin" else PermissionType.MEMBER
+                )
+                found_handler.event_filters.insert(0, new_filter)
+
+            return Response().ok({"message": f"已将 {handler_name} 权限设置为 {permission}"}).__dict__
+
+        except Exception:
+            logger.error(f"/api/command_permission/set: {traceback.format_exc()}")
+            return Response().error("设置指令权限失败").__dict__

astrbot/dashboard/server.py

@@ -60,6 +60,9 @@ def __init__(
         self.session_management_route = SessionManagementRoute(
             self.context, db, core_lifecycle
         )
+        self.command_permission_route = CommandPermissionRoute(
+            self.context, core_lifecycle
+        )
 
         self.app.add_url_rule(
             "/api/plug/<path:subpath>",

dashboard/src/i18n/locales/en-US/core/navigation.json

@@ -4,6 +4,7 @@
   "providers": "Providers",
   "toolUse": "MCP Tools",
   "config": "Config",
+  "commandPermission": "Command Permission",
   "extension": "Extensions",
   "extensionMarketplace": "Extension Market",
   "chat": "Chat",

dashboard/src/i18n/locales/zh-CN/core/navigation.json

@@ -4,6 +4,7 @@
   "providers": "服务提供商",
   "toolUse": "MCP",
   "config": "配置文件",
+  "commandPermission": "指令权限管理",
   "extension": "插件管理",
   "extensionMarketplace": "插件市场",
   "chat": "聊天",

dashboard/src/layouts/full/vertical-sidebar/sidebarItem.ts

@@ -43,6 +43,11 @@ const sidebarItem: menu[] = [
     icon: 'mdi-cog',
     to: '/config',
   },
+  {
+    title: 'core.navigation.commandPermission',
+    icon: 'mdi-shield-key',
+    to: '/command-permission',
+  },
   {
     title: 'core.navigation.extension',
     icon: 'mdi-puzzle',

dashboard/src/router/MainRoutes.ts

@@ -105,6 +105,11 @@ const MainRoutes = {
       name: 'About',
       path: '/about',
       component: () => import('@/views/AboutPage.vue')
+    },
+    {
+      name: 'CommandPermission',
+      path: '/command-permission',
+      component: () => import('@/views/CommandPermissionPage.vue')
     }
   ]
 };