diff --git a/astrbot/core/provider/func_tool_manager.py b/astrbot/core/provider/func_tool_manager.py index ab6dd037f..f3b3cb77c 100644 --- a/astrbot/core/provider/func_tool_manager.py +++ b/astrbot/core/provider/func_tool_manager.py @@ -210,6 +210,65 @@ async def _quick_test_mcp_connection(config: dict) -> tuple[bool, str]: return False, f"{e!s}" +class _PermissionGuardedTool(FunctionTool): + """Transparent proxy that checks per-tool permissions before delegating. + + Only wraps non-builtin tools. Builtin tools are added to the tool set + without wrapping, so their existing hardcoded permission logic + (``check_admin_permission`` / ``_is_restricted_env``) is unaffected. + + The ``handler`` field is intentionally kept ``None`` so that + ``FunctionToolExecutor._execute_local`` falls through to the + ``is_override_call`` branch and invokes our ``call()`` instead of + calling the raw handler directly. This ensures the permission + check runs for *every* invocation path. + """ + + def __init__( + self, + tool: FunctionTool, + manager: FunctionToolManager, + ) -> None: + # Do NOT pass handler to the parent — keep self.handler = None + # so the tool executor always routes through our call(). + super().__init__( + name=tool.name, + description=tool.description, + parameters=getattr(tool, "parameters", {}), + ) + self._wrapped = tool + self._mgr = manager + # Mirror mutable state from the underlying tool + self.active = getattr(tool, "active", True) + self.handler_module_path = getattr(tool, "handler_module_path", None) + + async def call(self, context: Any, **kwargs: Any) -> Any: + import inspect as _inspect + + error = self._mgr._check_tool_permission(self.name, context) + if error is not None: + return error + + # Delegate to handler first (plugin tools). + if self._wrapped.handler is not None: + result = self._wrapped.handler(context, **kwargs) + if _inspect.isasyncgen(result): + last: Any = None + async for item in result: + last = item + return last + if _inspect.isawaitable(result): + return await result + return result + + # Fall back to overridden call() on subclasses (e.g. MCPTool). + call_override = getattr(type(self._wrapped), "call", None) + if call_override is not None and call_override is not FunctionTool.call: + return await self._wrapped.call(context, **kwargs) + + return "error: tool has no callable handler" + + class FunctionToolManager: def __init__(self) -> None: self.func_list: list[FuncTool] = [] @@ -374,6 +433,51 @@ class FunctionToolManager: ensure_builtin_tools_loaded() return get_builtin_tool_class(name) is not None + def _default_permission(self, tool_name: str) -> str: + """Compute the fallback permission for a non-builtin tool. + + All non-builtin tools default to ``"member"`` (no restriction). + Builtin tools are never routed through this method.""" + return "member" + + def _check_tool_permission( + self, + tool_name: str, + context: Any, + ) -> str | None: + """Return an error string if the caller lacks permission, or None. + + Only non-builtin tools are guarded. Permission is resolved from + ``tool_permissions`` in SharedPreferences (``_default`` key). When + no explicit entry exists the tool inherits the fallback + ``_default_permission``.""" + try: + perms_raw = sp.get( + "tool_permissions", {}, scope="global", scope_id="global" + ) + except Exception: + perms_raw = {} + defaults = perms_raw.get("_default", {}) if isinstance(perms_raw, dict) else {} + effective = defaults.get(tool_name) + if effective is None: + effective = self._default_permission(tool_name) + + if effective != "admin": + return None # member or unknown → pass + + try: + event = context.context.event + except AttributeError: + event = None + if event is None or not event.is_admin(): + sender_id = getattr(event, "get_sender_id", lambda: "unknown")() + return ( + f"error: Permission denied. The tool '{tool_name}' requires admin " + f"privileges. Your ID: {sender_id}. " + "Ask admin to configure in WebUI → Extension → Components." + ) + return None + def get_full_tool_set(self) -> ToolSet: """获取完整工具集 @@ -383,10 +487,14 @@ class FunctionToolManager: 因此,后加载的 inactive 工具不会覆盖已激活的工具; 同时,MCP 工具在需要时仍可覆盖被禁用的内置工具。 + + Non-builtin tools are wrapped with ``_PermissionGuardedTool`` so that + every invocation checks the per-tool permission configured via the + dashboard. """ tool_set = ToolSet() for tool in self.func_list: - tool_set.add_tool(tool) + tool_set.add_tool(_PermissionGuardedTool(tool, self)) return tool_set @staticmethod diff --git a/astrbot/dashboard/routes/tools.py b/astrbot/dashboard/routes/tools.py index 157b4d75b..2ad80687e 100644 --- a/astrbot/dashboard/routes/tools.py +++ b/astrbot/dashboard/routes/tools.py @@ -2,7 +2,7 @@ import traceback from quart import request -from astrbot.core import logger +from astrbot.core import logger, sp from astrbot.core.agent.mcp_client import MCPTool, validate_mcp_stdio_config from astrbot.core.core_lifecycle import AstrBotCoreLifecycle from astrbot.core.star import star_map @@ -55,11 +55,27 @@ class ToolsRoute(Route): "/tools/mcp/test": ("POST", self.test_mcp_connection), "/tools/list": ("GET", self.get_tool_list), "/tools/toggle-tool": ("POST", self.toggle_tool), + "/tools/permission": ("POST", self.update_tool_permission), "/tools/mcp/sync-provider": ("POST", self.sync_provider), } self.register_routes() self.tool_mgr = self.core_lifecycle.provider_manager.llm_tools + @staticmethod + def _get_tool_permission(tool_name: str) -> tuple[str, bool]: + """Return (effective_permission, configured) for a tool. + + ``configured`` is True when the permission was explicitly set via the + dashboard rather than being a fallback default. + """ + perms_store = sp.get("tool_permissions", {}, scope="global", scope_id="global") + defaults = ( + perms_store.get("_default", {}) if isinstance(perms_store, dict) else {} + ) + if tool_name in defaults: + return defaults[tool_name], True + return "member", False + def _rollback_mcp_server(self, name: str) -> bool: try: rollback_config = self.tool_mgr.load_mcp_config() @@ -504,6 +520,10 @@ class ToolsRoute(Route): "builtin_config_statuses": builtin_config_statuses, "builtin_config_tags": builtin_config_tags, } + if not readonly: + perm, configured = self._get_tool_permission(tool.name) + tool_info["permission"] = perm + tool_info["permission_configured"] = configured tools_dict.append(tool_info) return Response().ok(data=tools_dict).__dict__ except Exception as e: @@ -569,3 +589,56 @@ class ToolsRoute(Route): except Exception as e: logger.error(traceback.format_exc()) return Response().error(f"Sync failed: {e!s}").__dict__ + + async def update_tool_permission(self): + """Set or remove the permission level of a registered tool.""" + try: + data = await request.json + tool_name = data.get("name") + permission = data.get("permission") # "admin" | "member" + + if not tool_name or permission not in ("admin", "member"): + return ( + Response() + .error("name and permission (admin or member) are required") + .__dict__ + ) + + if self.tool_mgr.is_builtin_tool(tool_name): + return ( + Response() + .error( + "Builtin tools do not support per-tool permission configuration." + ) + .__dict__ + ) + + # Verify the tool is known + if not any(t.name == tool_name for t in self.tool_mgr.func_list): + return Response().error(f"Tool '{tool_name}' not found").__dict__ + + perms_store = sp.get( + "tool_permissions", {}, scope="global", scope_id="global" + ) + if not isinstance(perms_store, dict): + perms_store = {} + defaults = perms_store.get("_default", {}) + if not isinstance(defaults, dict): + defaults = {} + defaults[tool_name] = permission + perms_store["_default"] = defaults + sp.put( + "tool_permissions", + perms_store, + scope="global", + scope_id="global", + ) + + return ( + Response() + .ok(None, f"Tool '{tool_name}' permission set to {permission}") + .__dict__ + ) + except Exception as e: + logger.error(traceback.format_exc()) + return Response().error(f"Failed to update tool permission: {e!s}").__dict__ diff --git a/dashboard/src/components/extension/componentPanel/components/ToolTable.vue b/dashboard/src/components/extension/componentPanel/components/ToolTable.vue index 8a42f33fe..75e59f30a 100644 --- a/dashboard/src/components/extension/componentPanel/components/ToolTable.vue +++ b/dashboard/src/components/extension/componentPanel/components/ToolTable.vue @@ -12,14 +12,16 @@ const props = defineProps<{ const emit = defineEmits<{ (e: 'toggle-tool', tool: ToolItem): void; + (e: 'update-permission', tool: ToolItem, permission: 'admin' | 'member'): void; }>(); const toolHeaders = computed(() => [ { title: tmTool('functionTools.title'), key: 'name', minWidth: '320px' }, { title: tmTool('functionTools.description'), key: 'description' }, - { title: tmTool('functionTools.table.origin'), key: 'origin', sortable: false, width: '120px' }, - { title: tmTool('functionTools.table.originName'), key: 'origin_name', sortable: false, width: '160px' }, - { title: tmTool('functionTools.table.actions'), key: 'actions', sortable: false, width: '120px' } + { title: tmTool('functionTools.table.origin'), key: 'origin', sortable: false, width: '100px' }, + { title: tmTool('functionTools.table.originName'), key: 'origin_name', sortable: false, width: '140px' }, + { title: tmTool('functionTools.table.permission'), key: 'permission', sortable: false, width: '110px' }, + { title: tmTool('functionTools.table.actions'), key: 'actions', sortable: false, width: '100px' } ]); const parameterEntries = (tool: ToolItem) => Object.entries(tool.parameters?.properties || {}); @@ -69,6 +71,24 @@ const enabledConfigTags = (tool: ToolItem): BuiltinToolConfigTag[] => { if (tool.origin !== 'builtin') return []; return (tool.builtin_config_tags || []).filter(tag => tag.enabled); }; + +const getPermissionColor = (permission?: string): string => { + switch (permission) { + case 'admin': + return 'error'; + default: + return 'success'; + } +}; + +const getPermissionLabel = (permission?: string): string => { + switch (permission) { + case 'admin': + return tmTool('functionTools.table.permissionAdmin'); + default: + return tmTool('functionTools.table.permissionEveryone'); + } +}; + + + + diff --git a/dashboard/src/components/extension/componentPanel/types.ts b/dashboard/src/components/extension/componentPanel/types.ts index 4784cc8ee..cb96f33a7 100644 --- a/dashboard/src/components/extension/componentPanel/types.ts +++ b/dashboard/src/components/extension/componentPanel/types.ts @@ -37,6 +37,13 @@ export interface CommandSummary { conflicts: number; } +/** 工具摘要统计 */ +export interface ToolSummary { + total: number; + active: number; + inactive: number; +} + /** 过滤器状态 */ export interface FilterState { searchQuery: string; @@ -119,4 +126,8 @@ export interface ToolItem { origin_name?: string; builtin_config_statuses?: BuiltinToolConfigTag[]; builtin_config_tags?: BuiltinToolConfigTag[]; + /** Per-tool permission level ("admin" | "member"). Builtin tools omit this. */ + permission?: 'admin' | 'member'; + /** True when permission was explicitly configured rather than a fallback default. */ + permission_configured?: boolean; } diff --git a/dashboard/src/i18n/locales/en-US/features/tool-use.json b/dashboard/src/i18n/locales/en-US/features/tool-use.json index 858f50272..52c27bbfe 100644 --- a/dashboard/src/i18n/locales/en-US/features/tool-use.json +++ b/dashboard/src/i18n/locales/en-US/features/tool-use.json @@ -39,6 +39,9 @@ "description": "Function Description", "parameters": "Parameter List", "noParameters": "This tool has no parameters", + "filter": { + "showBuiltin": "Show builtin tools" + }, "table": { "paramName": "Parameter Name", "type": "Type", @@ -47,6 +50,11 @@ "origin": "Origin", "originName": "Origin Name", "readonly": "Read-only", + "permission": "Permission", + "permissionAdmin": "Admin", + "permissionMember": "All Users", + "permissionEveryone": "All Users", + "permissionBuiltin": "System Builtin", "actions": "Actions" }, "configTags": { @@ -57,6 +65,11 @@ "in": "{key} matched {expected}", "fallback": "{key} is currently {actual}" } + }, + "summary": { + "total": "Total", + "active": "Active", + "inactive": "Inactive" } }, "marketplace": { @@ -167,6 +180,9 @@ "toggleToolSuccess": "Tool status toggled successfully!", "toggleToolReadonly": "Builtin tools are read-only and cannot be enabled or disabled.", "toggleToolError": "Failed to toggle tool status: {error}", + "updateToolPermissionSuccess": "Permission for {name} updated", + "updateToolPermissionFailed": "Failed to update tool permission", + "updateToolPermissionBuiltin": "Builtin tools do not support per-tool permission configuration.", "testError": "Test connection failed: {error}" } } diff --git a/dashboard/src/i18n/locales/ru-RU/features/tool-use.json b/dashboard/src/i18n/locales/ru-RU/features/tool-use.json index f945d5356..a4e2a583e 100644 --- a/dashboard/src/i18n/locales/ru-RU/features/tool-use.json +++ b/dashboard/src/i18n/locales/ru-RU/features/tool-use.json @@ -39,6 +39,9 @@ "description": "Описание функции", "parameters": "Параметры", "noParameters": "У этого инструмента нет параметров", + "filter": { + "showBuiltin": "Показать встроенные" + }, "table": { "paramName": "Параметр", "type": "Тип", @@ -47,6 +50,11 @@ "origin": "Источник", "originName": "Имя источника", "readonly": "Только чтение", + "permission": "Доступ", + "permissionAdmin": "Админ", + "permissionMember": "Все", + "permissionEveryone": "Все", + "permissionBuiltin": "Встроенный", "actions": "Действия" }, "configTags": { @@ -57,6 +65,11 @@ "in": "{key} соответствует {expected}", "fallback": "Текущее значение {key}: {actual}" } + }, + "summary": { + "total": "Всего", + "active": "Активно", + "inactive": "Неактивно" } }, "marketplace": { @@ -167,6 +180,9 @@ "toggleToolSuccess": "Статус инструмента изменен!", "toggleToolReadonly": "Встроенные инструменты доступны только для чтения и не могут быть включены или выключены.", "toggleToolError": "Не удалось изменить статус: {error}", + "updateToolPermissionSuccess": "Доступ к {name} обновлён", + "updateToolPermissionFailed": "Не удалось обновить доступ к инструменту", + "updateToolPermissionBuiltin": "Встроенные инструменты не поддерживают настройку доступа.", "testError": "Ошибка теста связи: {error}" }, "syncProvider": { diff --git a/dashboard/src/i18n/locales/zh-CN/features/tool-use.json b/dashboard/src/i18n/locales/zh-CN/features/tool-use.json index 7b717d06c..a2f140261 100644 --- a/dashboard/src/i18n/locales/zh-CN/features/tool-use.json +++ b/dashboard/src/i18n/locales/zh-CN/features/tool-use.json @@ -39,6 +39,9 @@ "description": "功能描述", "parameters": "参数列表", "noParameters": "此工具没有参数", + "filter": { + "showBuiltin": "显示内置工具" + }, "table": { "paramName": "参数名", "type": "类型", @@ -47,6 +50,11 @@ "origin": "来源", "originName": "来源名称", "readonly": "只读", + "permission": "权限", + "permissionAdmin": "管理员", + "permissionMember": "全部用户", + "permissionEveryone": "全部用户", + "permissionBuiltin": "系统内置", "actions": "操作" }, "configTags": { @@ -57,6 +65,11 @@ "in": "{key} 命中了 {expected}", "fallback": "{key} 当前值为 {actual}" } + }, + "summary": { + "total": "总数", + "active": "已启用", + "inactive": "未启用" } }, "marketplace": { @@ -167,6 +180,9 @@ "toggleToolSuccess": "工具状态切换成功!", "toggleToolReadonly": "内置工具为只读,无法进行启用或停用操作。", "toggleToolError": "工具状态切换失败: {error}", + "updateToolPermissionSuccess": "工具 {name} 权限已更新", + "updateToolPermissionFailed": "工具权限更新失败", + "updateToolPermissionBuiltin": "内置工具不支持单独配置权限", "testError": "测试连接失败: {error}" } } diff --git a/tests/unit/test_tool_conflict_resolution.py b/tests/unit/test_tool_conflict_resolution.py index 7fd04d448..8146ad887 100644 --- a/tests/unit/test_tool_conflict_resolution.py +++ b/tests/unit/test_tool_conflict_resolution.py @@ -180,16 +180,20 @@ class TestFunctionToolManagerGetFullToolSet: assert web_search is not None assert web_search.active is True - def test_no_deepcopy_preserves_identity(self): - """Should not deep copy tools, preserving object identity.""" + def test_wrapping_preserves_tool_name_and_description(self): + """_PermissionGuardedTool wrapping should preserve name and description.""" + from astrbot.core.provider.func_tool_manager import _PermissionGuardedTool + manager = FunctionToolManager() tool = make_tool("web_search") manager.func_list = [tool] toolset = manager.get_full_tool_set() - # Same object reference (no deepcopy) - assert toolset.tools[0] is tool + wrapped = toolset.tools[0] + assert wrapped.name == "web_search" + assert wrapped.description == "Test tool web_search" + assert isinstance(wrapped, _PermissionGuardedTool) def test_mcp_tool_overrides_disabled_builtin(self): """ diff --git a/tests/unit/test_tool_permission.py b/tests/unit/test_tool_permission.py new file mode 100644 index 000000000..1a3a8a376 --- /dev/null +++ b/tests/unit/test_tool_permission.py @@ -0,0 +1,434 @@ +"""Tests for per-tool permission management.""" + +import asyncio +import json +from typing import Any +from unittest.mock import AsyncMock, MagicMock, patch + +import pytest + +from astrbot.core import sp +from astrbot.core.agent.tool import FunctionTool, ToolSet +from astrbot.core.provider.func_tool_manager import ( + FunctionToolManager, + _PermissionGuardedTool, +) +from astrbot.core.tools.computer_tools.shell import ExecuteShellTool + + +# ── helpers ────────────────────────────────────────────────────────── + + +def _make_coro(value: object): + """Return a fresh coroutine object that resolves to ``value``. + + Used to mock Quart's ``await request.json`` where ``json`` is an + async property — assigning a coroutine lets ``await`` work directly.""" + + async def _inner(): + return value + + return _inner() + + +def _make_context(role: str = "member", sender_id: str = "user_123"): + """Return a mock context object suitable for tool permission checks.""" + + class FakeEvent: + unified_msg_origin = "aiocqhttp:GroupMessage:g1" + + def is_admin(self) -> bool: + return role == "admin" + + def get_sender_id(self) -> str: + return sender_id + + class FakeConfig: + def get_config(self, umo: str | None = None): + return {} + + class FakeAstrContext: + context = FakeConfig() + event = FakeEvent() + + class FakeWrapper: + context = FakeAstrContext() + + return FakeWrapper() + + +def _dummy_tool(name: str = "test_tool") -> FunctionTool: + return FunctionTool( + name=name, + description="A test tool", + parameters={"type": "object", "properties": {}}, + handler=None, + ) + + +def _clear_tool_permissions() -> None: + sp.put("tool_permissions", {}, scope="global", scope_id="global") + + +# ── _default_permission ────────────────────────────────────────────── + + +def test_default_permission_is_member(): + mgr = FunctionToolManager() + assert mgr._default_permission("any_mcp_tool") == "member" + + +# ── _check_tool_permission ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_check_permission_passes_when_no_config(): + _clear_tool_permissions() + mgr = FunctionToolManager() + context = _make_context(role="member") + + error = mgr._check_tool_permission("no_such_tool", context) + assert error is None + + +@pytest.mark.asyncio +async def test_check_permission_passes_for_admin_with_admin_tool(): + sp.put( + "tool_permissions", + {"_default": {"dangerous_tool": "admin"}}, + scope="global", + scope_id="global", + ) + try: + mgr = FunctionToolManager() + context = _make_context(role="admin", sender_id="admin_001") + error = mgr._check_tool_permission("dangerous_tool", context) + assert error is None + finally: + _clear_tool_permissions() + + +@pytest.mark.asyncio +async def test_check_permission_denies_member_for_admin_tool(): + sp.put( + "tool_permissions", + {"_default": {"dangerous_tool": "admin"}}, + scope="global", + scope_id="global", + ) + try: + mgr = FunctionToolManager() + context = _make_context(role="member", sender_id="user_999") + error = mgr._check_tool_permission("dangerous_tool", context) + assert error is not None + assert "dangerous_tool" in str(error) + assert "admin" in str(error).lower() + assert "user_999" in str(error) + finally: + _clear_tool_permissions() + + +@pytest.mark.asyncio +async def test_check_permission_denies_when_no_event(): + sp.put( + "tool_permissions", + {"_default": {"dangerous_tool": "admin"}}, + scope="global", + scope_id="global", + ) + try: + mgr = FunctionToolManager() + + class FakeWrapper: + pass # no .context.event + + error = mgr._check_tool_permission("dangerous_tool", FakeWrapper()) + assert error is not None + assert "admin" in str(error).lower() + finally: + _clear_tool_permissions() + + +@pytest.mark.asyncio +async def test_check_permission_passes_for_member_when_configured_member(): + sp.put( + "tool_permissions", + {"_default": {"safe_tool": "member"}}, + scope="global", + scope_id="global", + ) + try: + mgr = FunctionToolManager() + context = _make_context(role="member") + error = mgr._check_tool_permission("safe_tool", context) + assert error is None + finally: + _clear_tool_permissions() + + +# ── _PermissionGuardedTool ─────────────────────────────────────────── + + +@pytest.mark.asyncio +async def test_guarded_tool_delegates_when_permission_passes(): + _clear_tool_permissions() + mgr = FunctionToolManager() + + called = False + + async def handler(ctx, **kw): + nonlocal called + called = True + return "ok" + + wrapped = FunctionTool( + name="delegated", + description="desc", + parameters={}, + handler=handler, + ) + guarded = _PermissionGuardedTool(wrapped, mgr) + context = _make_context(role="member") + + result = await guarded.call(context) + assert called + assert result == "ok" + + +@pytest.mark.asyncio +async def test_guarded_tool_blocks_when_permission_denied(): + sp.put( + "tool_permissions", + {"_default": {"blocked_tool": "admin"}}, + scope="global", + scope_id="global", + ) + try: + mgr = FunctionToolManager() + called = False + + async def handler(ctx, **kw): + nonlocal called + called = True + return "should not reach" + + wrapped = FunctionTool( + name="blocked_tool", + description="desc", + parameters={}, + handler=handler, + ) + guarded = _PermissionGuardedTool(wrapped, mgr) + context = _make_context(role="member") + + result = await guarded.call(context) + assert not called + assert isinstance(result, str) + assert "Permission denied" in result + finally: + _clear_tool_permissions() + + +@pytest.mark.asyncio +async def test_guarded_tool_delegates_to_wrapped_call(): + _clear_tool_permissions() + mgr = FunctionToolManager() + + class CallableTool(FunctionTool): + async def call(self, context, **kwargs): + return "from call()" + + wrapped = CallableTool( + name="has_call", + description="desc", + parameters={}, + ) + guarded = _PermissionGuardedTool(wrapped, mgr) + context = _make_context() + + result = await guarded.call(context) + assert result == "from call()" + + +@pytest.mark.asyncio +async def test_guarded_tool_handles_async_generator_handler(): + _clear_tool_permissions() + mgr = FunctionToolManager() + + async def gen_handler(ctx, **kw): # type: ignore[misc] + yield "A" + yield "B" + yield "C" + + wrapped = FunctionTool( + name="gen_tool", + description="desc", + parameters={}, + handler=gen_handler, + ) + guarded = _PermissionGuardedTool(wrapped, mgr) + context = _make_context() + + result = await guarded.call(context) + # should return the last yielded value + assert result == "C" + + +# ── get_full_tool_set ──────────────────────────────────────────────── + + +def test_get_full_tool_set_excludes_builtin_tools(): + """Builtin tools are added separately by astr_main_agent.py, not through + get_full_tool_set().""" + mgr = FunctionToolManager() + tool_set = mgr.get_full_tool_set() + + names = {t.name for t in tool_set.tools} + # Builtin tools are injected individually by the agent builder — + # they must NOT appear in the generic tool set. + assert "astrbot_execute_shell" not in names + + +def test_get_full_tool_set_wraps_non_builtin(): + mgr = FunctionToolManager() + _clear_tool_permissions() + + mgr.func_list.append(_dummy_tool("my_plugin_tool")) + tool_set = mgr.get_full_tool_set() + + plugin_tools = [t for t in tool_set.tools if t.name == "my_plugin_tool"] + assert plugin_tools + assert isinstance( + plugin_tools[0], _PermissionGuardedTool + ), "non-builtin tools must be wrapped" + + +# ── API: get_tool_list permission fields ────────────────────────────── + + +class TestGetToolListPermission: + @pytest.mark.asyncio + async def test_list_includes_permission_fields_for_non_builtin(self): + from astrbot.dashboard.routes.tools import ToolsRoute + + # Minimal stubs to avoid full core lifecycle init + route = ToolsRoute.__new__(ToolsRoute) + route.core_lifecycle = MagicMock() + route.core_lifecycle.astrbot_config_mgr = MagicMock() + route.core_lifecycle.astrbot_config_mgr.get_conf_list.return_value = [] + route.core_lifecycle.astrbot_config_mgr.confs = {} + route.tool_mgr = FunctionToolManager() + + sp.put( + "tool_permissions", + {"_default": {"my_plugin_tool": "admin"}}, + scope="global", + scope_id="global", + ) + try: + route.tool_mgr.func_list.append(_dummy_tool("my_plugin_tool")) + resp = await route.get_tool_list() + data = json.loads(json.dumps(resp)) # simulate json serialisation + tools = data["data"] + + target = next(t for t in tools if t["name"] == "my_plugin_tool") + assert target["permission"] == "admin" + assert target["permission_configured"] is True + assert target["readonly"] is False + finally: + _clear_tool_permissions() + + @pytest.mark.asyncio + async def test_list_no_permission_fields_for_builtin(self): + from astrbot.dashboard.routes.tools import ToolsRoute + + route = ToolsRoute.__new__(ToolsRoute) + route.core_lifecycle = MagicMock() + route.core_lifecycle.astrbot_config_mgr = MagicMock() + route.core_lifecycle.astrbot_config_mgr.get_conf_list.return_value = [] + route.core_lifecycle.astrbot_config_mgr.confs = {} + route.tool_mgr = FunctionToolManager() + + resp = await route.get_tool_list() + data = json.loads(json.dumps(resp)) + tools = data["data"] + + target = next(t for t in tools if t["name"] == "astrbot_execute_shell") + assert "permission" not in target + assert "permission_configured" not in target + assert target["readonly"] is True + + +# ── API: update_tool_permission ────────────────────────────────────── + + +class TestUpdateToolPermission: + @pytest.mark.asyncio + async def test_set_admin_permission(self): + from astrbot.dashboard.routes.tools import ToolsRoute + + route = ToolsRoute.__new__(ToolsRoute) + route.core_lifecycle = MagicMock() + route.tool_mgr = FunctionToolManager() + route.tool_mgr.func_list.append(_dummy_tool("target_tool")) + _clear_tool_permissions() + + mock_req = MagicMock() + mock_req.json = _make_coro({"name": "target_tool", "permission": "admin"}) + with patch("astrbot.dashboard.routes.tools.request", mock_req): + resp = await route.update_tool_permission() + data = json.loads(json.dumps(resp)) + assert data["status"] == "ok" + + stored = sp.get( + "tool_permissions", {}, scope="global", scope_id="global" + ) + assert stored["_default"]["target_tool"] == "admin" + + @pytest.mark.asyncio + async def test_reject_builtin_tool(self): + from astrbot.dashboard.routes.tools import ToolsRoute + + route = ToolsRoute.__new__(ToolsRoute) + route.core_lifecycle = MagicMock() + route.tool_mgr = FunctionToolManager() + + mock_req = MagicMock() + mock_req.json = _make_coro({"name": "astrbot_execute_shell", "permission": "admin"}) + with patch("astrbot.dashboard.routes.tools.request", mock_req): + resp = await route.update_tool_permission() + data = json.loads(json.dumps(resp)) + assert data["status"] == "error" + assert "builtin" in str(data["message"]).lower() + + @pytest.mark.asyncio + async def test_reject_unknown_tool(self): + from astrbot.dashboard.routes.tools import ToolsRoute + + route = ToolsRoute.__new__(ToolsRoute) + route.core_lifecycle = MagicMock() + route.tool_mgr = FunctionToolManager() + + mock_req = MagicMock() + mock_req.json = _make_coro({"name": "ghost_tool", "permission": "admin"}) + with patch("astrbot.dashboard.routes.tools.request", mock_req): + resp = await route.update_tool_permission() + data = json.loads(json.dumps(resp)) + assert data["status"] == "error" + assert "not found" in str(data["message"]).lower() + + @pytest.mark.asyncio + async def test_reject_invalid_permission_value(self): + from astrbot.dashboard.routes.tools import ToolsRoute + + route = ToolsRoute.__new__(ToolsRoute) + route.core_lifecycle = MagicMock() + route.tool_mgr = FunctionToolManager() + route.tool_mgr.func_list.append(_dummy_tool("target_tool")) + + mock_req = MagicMock() + mock_req.json = _make_coro({"name": "target_tool", "permission": "everyone"}) + with patch("astrbot.dashboard.routes.tools.request", mock_req): + resp = await route.update_tool_permission() + data = json.loads(json.dumps(resp)) + assert data["status"] == "error"