feat(dashboard): add per-tool permission management for function tools (#8693)

* feat(func-tool): add per-tool permission management

* feat(dashboard): add permission column to function tool table

* refactor(dashboard): encapsulate tool actions into composable and unify permission UI

* style: format code

* fix: guard sync handler, simplify sp access, skip builtin instantiation
This commit is contained in:
Ruochen Pan
2026-06-09 14:47:41 +08:00
committed by GitHub
parent 0fb3f5eb93
commit ae44b912fc
11 changed files with 936 additions and 46 deletions

View File

@@ -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

View File

@@ -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__

View File

@@ -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');
}
};
</script>
<template>
@@ -133,11 +153,54 @@ const enabledConfigTags = (tool: ToolItem): BuiltinToolConfigTag[] => {
</template>
<template #item.origin_name="{ item }">
<div class="text-body-2 text-medium-emphasis" style="max-width: 200px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap;" :title="item.origin_name">
<div class="text-body-2 text-medium-emphasis" style="max-width: 180px; overflow: hidden; text-overflow: ellipsis; white-space: nowrap;" :title="item.origin_name">
{{ item.origin_name || '-' }}
</div>
</template>
<template #item.permission="{ item }">
<!-- Builtin tools: non-clickable badge -->
<v-chip
v-if="item.origin === 'builtin'"
size="small"
variant="tonal"
class="font-weight-medium text-medium-emphasis"
>
{{ tmTool('functionTools.table.permissionBuiltin') }}
</v-chip>
<!-- Other tools: clickable dropdown -->
<v-menu v-else location="bottom">
<template v-slot:activator="{ props: menuProps }">
<v-chip
v-bind="menuProps"
:color="getPermissionColor(item.permission)"
size="small"
class="font-weight-medium cursor-pointer"
link
>
{{ getPermissionLabel(item.permission) }}
<v-icon end size="14">mdi-chevron-down</v-icon>
</v-chip>
</template>
<v-list density="compact">
<v-list-item
:value="'member'"
@click="emit('update-permission', item, 'member')"
:active="item.permission !== 'admin'"
>
<v-list-item-title>{{ tmTool('functionTools.table.permissionEveryone') }}</v-list-item-title>
</v-list-item>
<v-list-item
:value="'admin'"
@click="emit('update-permission', item, 'admin')"
:active="item.permission === 'admin'"
>
<v-list-item-title>{{ tmTool('functionTools.table.permissionAdmin') }}</v-list-item-title>
</v-list-item>
</v-list>
</v-menu>
</template>
<template #item.actions="{ item }">
<span v-if="item.readonly" class="text-medium-emphasis">-</span>
<v-switch

View File

@@ -0,0 +1,127 @@
/**
* Tool actions composable
*/
import { ref, computed, type Ref } from 'vue';
import axios from 'axios';
import { normalizeTextInput } from '@/utils/inputValue';
import type { ToolItem, ToolSummary } from '../types';
export function useToolActions(
tools: Ref<ToolItem[]>,
toast: (message: string, color?: string) => void
) {
const toolSearch = ref('');
const showBuiltinTools = ref(true);
const filteredTools = computed(() => {
let result = tools.value;
// Filter builtin tools
if (!showBuiltinTools.value) {
result = result.filter(tool => tool.origin !== 'builtin');
}
// Filter by search query
const query = normalizeTextInput(toolSearch.value).trim().toLowerCase();
if (query) {
result = result.filter(
tool =>
tool.name?.toLowerCase().includes(query) ||
tool.description?.toLowerCase().includes(query)
);
}
return result;
});
const toolSummary = computed<ToolSummary>(() => {
const all = tools.value;
return {
total: all.length,
active: all.filter(t => t.active).length,
inactive: all.filter(t => !t.active).length,
};
});
/**
* Toggle a tool's active state (optimistic update).
*/
const toggleTool = async (
tool: ToolItem,
readonlyMessage: string,
successMessage: string,
errorMessage: string
) => {
if (tool.readonly) {
toast(readonlyMessage, 'info');
return;
}
const previous = tool.active;
tool.active = !tool.active;
try {
const res = await axios.post('/api/tools/toggle-tool', {
name: tool.name,
activate: tool.active,
});
if (res.data.status === 'ok') {
toast(res.data.message || successMessage);
} else {
tool.active = previous;
toast(res.data.message || errorMessage, 'error');
}
} catch (error: any) {
tool.active = previous;
toast(
error?.response?.data?.message ||
error?.message ||
errorMessage,
'error'
);
}
};
/**
* Update a tool's permission level.
*/
const updateToolPermission = async (
tool: ToolItem,
permission: 'admin' | 'member',
successMessage: string,
builtinMessage: string,
errorMessage: string
) => {
if (tool.origin === 'builtin') {
toast(builtinMessage, 'info');
return;
}
try {
const res = await axios.post('/api/tools/permission', {
name: tool.name,
permission,
});
if (res.data.status === 'ok') {
tool.permission = permission;
tool.permission_configured = true;
toast(successMessage, 'success');
} else {
toast(res.data.message || errorMessage, 'error');
}
} catch (error: any) {
toast(
error?.response?.data?.message ||
error?.message ||
errorMessage,
'error'
);
}
};
return {
toolSearch,
showBuiltinTools,
filteredTools,
toolSummary,
toggleTool,
updateToolPermission,
};
}

View File

@@ -12,15 +12,14 @@
* - components/RenameDialog.vue: 重命名对话框
* - components/DetailsDialog.vue: 详情对话框
*/
import { computed, onActivated, onMounted, ref, watch} from 'vue';
import axios from 'axios';
import { onActivated, onMounted, ref, watch} from 'vue';
import { useModuleI18n } from '@/i18n/composables';
import { normalizeTextInput } from '@/utils/inputValue';
// Composables
import { useComponentData } from './composables/useComponentData';
import { useCommandFilters } from './composables/useCommandFilters';
import { useCommandActions } from './composables/useCommandActions';
import { useToolActions } from './composables/useToolActions';
// Components
import CommandFilters from './components/CommandFilters.vue';
@@ -41,7 +40,6 @@ const { tm } = useModuleI18n('features/command');
const { tm: tmTool } = useModuleI18n('features/tooluse');
const viewMode = ref<'commands' | 'tools'>('commands');
const toolSearch = ref('');
// 数据管理
const {
@@ -83,14 +81,15 @@ const {
openDetailsDialog
} = useCommandActions(toast, () => fetchCommands(tm('messages.loadFailed')));
const filteredTools = computed(() => {
const query = normalizeTextInput(toolSearch.value).trim().toLowerCase();
if (!query) return tools.value;
return tools.value.filter(tool =>
tool.name?.toLowerCase().includes(query) ||
tool.description?.toLowerCase().includes(query)
);
});
// 工具操作方法
const {
toolSearch,
showBuiltinTools,
filteredTools,
toolSummary,
toggleTool,
updateToolPermission,
} = useToolActions(tools, toast);
// 处理切换指令状态
const handleToggleCommand = async (cmd: CommandItem) => {
@@ -102,27 +101,11 @@ const handleUpdatePermission = async (cmd: CommandItem, permission: 'admin' | 'm
};
const handleToggleTool = async (tool: ToolItem) => {
if (tool.readonly) {
toast(tmTool('messages.toggleToolReadonly'), 'info');
return;
}
const previous = tool.active;
tool.active = !tool.active;
try {
const res = await axios.post('/api/tools/toggle-tool', {
name: tool.name,
activate: tool.active
});
if (res.data.status === 'ok') {
toast(res.data.message || tmTool('messages.toggleToolSuccess'));
} else {
tool.active = previous;
toast(res.data.message || tmTool('messages.toggleToolError', { error: '' }), 'error');
}
} catch (error: any) {
tool.active = previous;
toast(error?.response?.data?.message || error?.message || tmTool('messages.toggleToolError', { error: '' }), 'error');
}
await toggleTool(tool, tmTool('messages.toggleToolReadonly'), tmTool('messages.toggleToolSuccess'), tmTool('messages.toggleToolError', { error: '' }));
};
const handleUpdateToolPermission = async (tool: ToolItem, permission: 'admin' | 'member') => {
await updateToolPermission(tool, permission, tmTool('messages.updateToolPermissionSuccess', { name: tool.name }), tmTool('messages.updateToolPermissionBuiltin'), tmTool('messages.updateToolPermissionFailed'));
};
// 处理确认重命名
@@ -255,11 +238,10 @@ watch(viewMode, async (mode) => {
</div>
<div v-else>
<div class="d-flex flex-wrap align-center ga-3 mb-4">
<div class="d-flex flex-wrap align-center ga-4 mb-4">
<div style="min-width: 240px; max-width: 380px; flex: 1;">
<v-text-field
:model-value="toolSearch"
@update:model-value="toolSearch = normalizeTextInput($event)"
v-model="toolSearch"
prepend-inner-icon="mdi-magnify"
:label="tmTool('functionTools.search')"
variant="outlined"
@@ -268,12 +250,42 @@ watch(viewMode, async (mode) => {
clearable
/>
</div>
<div class="d-flex align-center ga-4">
<div class="d-flex align-center">
<v-icon size="18" color="primary" class="mr-1">mdi-function-variant</v-icon>
<span class="text-body-2 text-medium-emphasis mr-1">{{ tmTool('functionTools.summary.total') }}:</span>
<span class="text-body-1 font-weight-bold text-primary">{{ toolSummary.total }}</span>
</div>
<v-divider vertical class="mx-1" style="height: 20px;" />
<div class="d-flex align-center">
<v-icon size="18" color="success" class="mr-1">mdi-check-circle-outline</v-icon>
<span class="text-body-2 text-medium-emphasis mr-1">{{ tmTool('functionTools.summary.active') }}:</span>
<span class="text-body-1 font-weight-bold text-success">{{ toolSummary.active }}</span>
</div>
<v-divider vertical class="mx-1" style="height: 20px;" />
<div class="d-flex align-center">
<v-icon size="18" color="error" class="mr-1">mdi-close-circle-outline</v-icon>
<span class="text-body-2 text-medium-emphasis mr-1">{{ tmTool('functionTools.summary.inactive') }}:</span>
<span class="text-body-1 font-weight-bold text-error">{{ toolSummary.inactive }}</span>
</div>
<v-divider vertical class="mx-1" style="height: 20px;" />
<v-checkbox
v-model="showBuiltinTools"
:label="tmTool('functionTools.filter.showBuiltin')"
density="compact"
hide-details
class="builtin-tools-checkbox"
/>
</div>
</div>
<ToolTable
:items="filteredTools"
:loading="toolsLoading"
@toggle-tool="handleToggleTool"
@update-permission="handleUpdateToolPermission"
/>
</div>
</v-card-text>
@@ -306,3 +318,13 @@ watch(viewMode, async (mode) => {
{{ snackbar.message }}
</v-snackbar>
</template>
<style scoped>
.builtin-tools-checkbox {
flex: none;
}
.builtin-tools-checkbox :deep(.v-selection-control) {
min-height: auto;
}
</style>

View File

@@ -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;
}

View File

@@ -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}"
}
}

View File

@@ -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": {

View File

@@ -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}"
}
}

View File

@@ -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):
"""

View File

@@ -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"