diff --git a/routes/mcp/mcp_routes.py b/routes/mcp/mcp_routes.py index 5d14ba8c3..5a8ae9c92 100644 --- a/routes/mcp/mcp_routes.py +++ b/routes/mcp/mcp_routes.py @@ -5,6 +5,7 @@ import os import uuid import urllib.parse import html +import secrets from pathlib import Path from fastapi import APIRouter, Form, HTTPException, Request from fastapi.responses import RedirectResponse, HTMLResponse @@ -12,7 +13,7 @@ import logging import httpx from core.database import McpServer, SessionLocal -from core.middleware import require_admin +from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN, require_admin from src.constants import DATA_DIR, MCP_OAUTH_DIR from src.mcp_manager import McpManager @@ -424,6 +425,11 @@ def setup_mcp_routes(mcp_manager: McpManager): Expects JSON body: {"disabled": ["tool_name_1", "tool_name_2"]} """ require_admin(request) + internal_token = request.headers.get(INTERNAL_TOOL_HEADER) + if internal_token and secrets.compare_digest(internal_token, INTERNAL_TOOL_TOKEN): + # Same boundary as POST /api/tools: an agent turn must not be able + # to re-enable MCP tools an admin switched off. + raise HTTPException(403, "Change MCP tool policy in Settings > Integrations, not from chat") db = SessionLocal() try: srv = db.query(McpServer).filter(McpServer.id == server_id).first() diff --git a/routes/model_routes.py b/routes/model_routes.py index 9749e0bbd..4535bbafe 100644 --- a/routes/model_routes.py +++ b/routes/model_routes.py @@ -7,6 +7,7 @@ import json import hashlib import ipaddress import socket +import secrets import time as _time import logging import threading @@ -3091,6 +3092,12 @@ def setup_model_routes(model_discovery): def update_tools(body: ToolsUpdate, request: Request): """Update which tools are disabled.""" require_admin(request) + from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN + internal_token = request.headers.get(INTERNAL_TOOL_HEADER) + if internal_token and secrets.compare_digest(internal_token, INTERNAL_TOOL_TOKEN): + # Owner impersonation can carry an admin username, so check the + # authenticated transport rather than request.state.current_user. + raise HTTPException(403, "Change tool policy in Settings > Agent Tools, not from chat") settings = _load_settings() settings["disabled_tools"] = body.disabled _save_settings(settings) diff --git a/src/agent_loop.py b/src/agent_loop.py index 2556ec27c..3b45c77d5 100644 --- a/src/agent_loop.py +++ b/src/agent_loop.py @@ -2834,7 +2834,7 @@ def _parse_qwen_explicit_admin_request(text: str) -> Optional[tuple[str, str]]: if re.search(r"\b(?:disabled\s+tools?|tools?\s+(?:are\s+)?(?:currently\s+)?disabled)\b", q): return "manage_settings", json.dumps({"action": "list_tools"}) if re.search(r"\b(?:turn|enable|re-enable|reenable)\b.{0,40}\b(?:image\s+generation|images?|generate_image)\b", q): - return "manage_settings", json.dumps({"action": "enable_tool", "tool": "images"}) + return "manage_settings", json.dumps({"action": "list_tools"}) if re.search(r"\b(?:disable|turn\s+off)\b.{0,40}\b(?:image\s+generation|images?|generate_image)\b", q): return "manage_settings", json.dumps({"action": "disable_tool", "tool": "images"}) if listish and re.search(r"\bapi\s+tokens?\b|\btokens?\b", q): @@ -8057,7 +8057,7 @@ Generate an image. Line 1 = description, line 2 = model name, line 3 = WxH (e.g. "manage_tokens": "- ```manage_tokens``` — Generate or revoke API access tokens for external integrations. Args (JSON): {\"action\": \"list|create|delete\", ...}", "manage_documents": "- ```manage_documents``` — List, read/open, delete, or tidy documents in the editor panel. Args (JSON): {\"action\": \"list|read|delete|tidy\", ...}. `list` returns rows like `[Title](#document-) — lang, size, updated 5m ago` sorted MOST-RECENT FIRST; the user clicks the anchor to open. `read` (aliases: view/open/get) takes `document_id` and returns the content. When the user asks \"open/show/read my notes\" or \"what documents do I have\", use this — do NOT shell out, do NOT curl.", "manage_research": "- ```manage_research``` — List, read/open, or delete saved DEEP RESEARCH results from the Library. Args (JSON): {\"action\": \"list|read|delete\", \"id\": \"\", \"search\": \"...\"}. `list` returns rows like `[query](#research-) — N sources` MOST-RECENT FIRST; the user clicks to open. `read` (aliases: open/view/get) takes `id` and returns the report text + sources. Use when the user says \"open/read/find/delete my research\" or \"that report\". This IS how you read a finished report: when the user refers to a just-completed deep-research job (\"check it out\", \"read that report\", \"summarize the research\") WITHOUT giving an id, call `manage_research` with `action:list` to get the most-recent id, then `action:read` with that id, and answer from the returned text. Do NOT `web_fetch`/`app_api` the `/api/research/report/{id}` URL — that endpoint renders HTML for the browser, not clean text — and do NOT start a fresh `web_search`/`trigger_research` just to read an existing report. To START new research, use trigger_research instead.", - "manage_settings": "- ```manage_settings``` — View/change the REAL app settings (same ones the Settings panel writes) AND turn tools on/off. Change a setting: `{\"action\":\"set\",\"key\":\"...\",\"value\":\"...\"}` — keys accept friendly aliases, e.g. voice→tts_voice, \"search engine\"→search_provider, \"default model\"→default_model, \"teacher model\"→teacher_model, \"task/background model\"→task_model, \"image quality\"→image_quality, \"reminder channel\"→reminder_channel (browser|email|ntfy), \"agent timeout\"/\"max tool calls\"/\"token budget\". Read: `{\"action\":\"get\",\"key\":\"...\"}`; see all: `{\"action\":\"list\"}`; reset one: `{\"action\":\"reset\",\"key\":\"...\"}`. Use this when the user asks to change ANY preference instead of making them open Settings. Secrets/API keys are read-only (tell them to set those in the panel). Tool toggles: `{\"action\":\"disable_tool|enable_tool\",\"tool\":\"shell\"}` (aliases: shell/search/browser/documents/memory/skills/images/tasks/notes/calendar/email), list disabled: `{\"action\":\"list_tools\"}`.", + "manage_settings": "- ```manage_settings``` — View/change the REAL app settings (same ones the Settings panel writes) AND turn tools on/off. Change a setting: `{\"action\":\"set\",\"key\":\"...\",\"value\":\"...\"}` — keys accept friendly aliases, e.g. voice→tts_voice, \"search engine\"→search_provider, \"default model\"→default_model, \"teacher model\"→teacher_model, \"task/background model\"→task_model, \"image quality\"→image_quality, \"reminder channel\"→reminder_channel (browser|email|ntfy), \"agent timeout\"/\"max tool calls\"/\"token budget\". Read: `{\"action\":\"get\",\"key\":\"...\"}`; see all: `{\"action\":\"list\"}`; reset one: `{\"action\":\"reset\",\"key\":\"...\"}`. Use this when the user asks to change ANY preference instead of making them open Settings. Secrets/API keys are read-only (tell them to set those in the panel). Tool toggles: `{\"action\":\"disable_tool\",\"tool\":\"shell\"}` (aliases: shell/search/browser/documents/memory/skills/images/tasks/notes/calendar/email), list disabled: `{\"action\":\"list_tools\"}`. You cannot re-enable a disabled tool: that is done by the user in Settings > Agent Tools.", "manage_notes": """\ ```manage_notes {"action": "add", "title": "", "due_date": ""} diff --git a/src/agent_tools/admin_tools.py b/src/agent_tools/admin_tools.py index f0a7662c2..419b7f742 100644 --- a/src/agent_tools/admin_tools.py +++ b/src/agent_tools/admin_tools.py @@ -585,7 +585,9 @@ async def do_manage_settings(content: str, owner: Optional[str] = None) -> Dict: return _ALIASES_SET.get(k2, (k or "").strip()) def _is_managed_key(key): - return key in DEFAULT_SETTINGS and key not in RETIRED_SETTING_KEYS + # Tool policy is not a general setting, even if a future default + # is added: resetting it must never re-enable disabled tools. + return key != "disabled_tools" and key in DEFAULT_SETTINGS and key not in RETIRED_SETTING_KEYS _ENUMS = { "image_quality": ["low", "medium", "high"], @@ -713,13 +715,35 @@ async def do_manage_settings(content: str, owner: Optional[str] = None) -> Dict: save_settings(s) return {"response": f"Reset {key} to default ({DEFAULT_SETTINGS[key]}).", "exit_code": 0} - elif action in ("disable_tool", "enable_tool", "list_tools"): + elif action == "enable_tool": + # The global disabled_tools denylist is the one hard permission + # boundary the Settings -> Agent Tools panel offers admins. It + # must only ever be *narrowed* from inside an agent turn: a + # model that can call enable_tool can undo the boundary itself, + # and any web page, email, or document the agent reads can ask + # it to (prompt injection re-enabling bash/email/etc.). Widening + # the tool set stays a human action in the Settings panel. + tool_name = (args.get("tool") or args.get("name") or "").strip().lower() + return { + "error": ( + f"Re-enabling tools is not available from chat" + f"{f' (requested: {tool_name})' if tool_name else ''}. " + "Disabled tools are an admin policy: ask the user to turn the tool " + "back on in Settings > Agent Tools. You can still disable tools " + "(disable_tool) or list the current state (list_tools)." + ), + "exit_code": 1, + } + + elif action in ("disable_tool", "list_tools"): # Tool-toggle actions. These edit settings.json:disabled_tools # (the global list read on every chat request) rather than # prefs.json. Friendly aliases accepted: "shell" -> "bash", # "search" -> "web_search", "browser" -> browser tools, # "documents" -> the document tool set, "memory" -> - # manage_memory, etc. + # manage_memory, etc. Only disable_tool mutates: see the + # enable_tool branch above for why the agent can never widen + # the denylist. from src.settings import get_setting, save_settings, load_settings _ALIASES = { "shell": ["bash"], @@ -753,7 +777,8 @@ async def do_manage_settings(content: str, owner: Optional[str] = None) -> Dict: "response": ( f"Currently disabled: {', '.join(current) if current else '(none)'}.\n" "Common toggles: shell (bash), search (web_search), browser, documents, " - "memory, skills, images, tasks, notes, calendar, email." + "memory, skills, images, tasks, notes, calendar, email. " + "To re-enable a tool, open Settings > Agent Tools." ), "disabled": list(current), "exit_code": 0, @@ -767,22 +792,19 @@ async def do_manage_settings(content: str, owner: Optional[str] = None) -> Dict: settings = load_settings() current = list(settings.get("disabled_tools") or []) before = set(current) - if action == "disable_tool": - for t in targets: - if t not in current: - current.append(t) - else: # enable_tool - current = [t for t in current if t not in targets] + for t in targets: + if t not in current: + current.append(t) after = set(current) settings["disabled_tools"] = current save_settings(settings) - verb = "Disabled" if action == "disable_tool" else "Enabled" changed = sorted(after.symmetric_difference(before)) return { "response": ( - f"{verb} {tool_name} ({', '.join(targets)}). " - f"Now disabled: {', '.join(current) if current else '(none)'}." + f"Disabled {tool_name} ({', '.join(targets)}). " + f"Now disabled: {', '.join(current) if current else '(none)'}. " + "Re-enabling is done by the user in Settings > Agent Tools." ), "changed": changed, "disabled": list(current), diff --git a/src/tool_capabilities.py b/src/tool_capabilities.py index 966937655..c7a17da80 100644 --- a/src/tool_capabilities.py +++ b/src/tool_capabilities.py @@ -410,7 +410,7 @@ _PRIVATE_ACTION_WRITES: Mapping[str, frozenset[str]] = MappingProxyType( "manage_endpoints": frozenset({"add", "delete", "enable", "disable"}), "manage_mcp": frozenset({"add", "delete", "enable", "disable", "reconnect"}), "manage_settings": frozenset( - {"set", "delete", "reset", "disable_tool", "enable_tool"} + {"set", "delete", "reset", "disable_tool"} ), "manage_tokens": frozenset({"create", "delete"}), "manage_webhooks": frozenset({"add", "delete", "enable", "disable"}), diff --git a/src/tool_index.py b/src/tool_index.py index 483cddbab..e0e4c8fb6 100644 --- a/src/tool_index.py +++ b/src/tool_index.py @@ -145,7 +145,7 @@ BUILTIN_TOOL_DESCRIPTIONS: Dict[str, str] = { "manage_tokens": "API token management: list, create, or delete API access tokens.", "manage_documents": "List, read, delete, or tidy documents in the editor panel. action='list' returns clickable rows (most-recent first) so the user can open any doc by clicking. action='read' (aka view/open/get) with document_id returns the content; supports offset= + limit= to page through large docs (response includes next_offset when more remains, so you can keep calling with offset=next_offset). action='delete' with document_id removes a doc (only way to delete). Use this for ANY 'show/read/list/open my documents/docs/files/notes' request — never shell or curl.", "manage_research": "List, read/open, or delete saved DEEP RESEARCH results from the Library. action='list' returns clickable [query](#research-) rows (most-recent first). action='read' (aka open/view/get) with id returns the report + sources. action='delete' with id removes it. Use this for ANY 'open/read/find/delete my research / that report / the research on X' request. NOTE: this is for EXISTING research; to START new research use trigger_research.", - "manage_settings": "Change ANY real app setting (the ones the Settings panel writes) so the user never has to open it: TTS voice/provider/speed, STT, search engine + result count, default/teacher/task/utility/vision/image/research models, image quality, reminder channel (browser/email/ntfy), agent timeout/tool-call budget, and more. action=set with key (friendly aliases ok: voice, 'search engine', 'default model', 'teacher model', 'image quality', 'reminder channel'...) + value; get/list/reset too. Also toggles tools on/off (disable_tool/enable_tool/list_tools). Secrets/API keys are read-only. Use for any 'change my…/set my…/use X for…/turn on…' preference request.", + "manage_settings": "Change ANY real app setting (the ones the Settings panel writes) so the user never has to open it: TTS voice/provider/speed, STT, search engine + result count, default/teacher/task/utility/vision/image/research models, image quality, reminder channel (browser/email/ntfy), agent timeout/tool-call budget, and more. action=set with key (friendly aliases ok: voice, 'search engine', 'default model', 'teacher model', 'image quality', 'reminder channel'...) + value; get/list/reset too. Also turns tools off globally (disable_tool) and lists disabled ones (list_tools); re-enabling is only done by the user in Settings > Agent Tools. Secrets/API keys are read-only. Use for any 'change my…/set my…/use X for…/turn on…' preference request.", "create_session": "Create a new chat with a name and model.", "list_sessions": "List all chats with their metadata (the UI calls these 'chats'). Use for 'list my chats', 'rename all my chats' (list first, then manage_session to rename each).", "send_to_session": "Send a message to another chat. Cross-chat communication.", diff --git a/src/tool_schemas.py b/src/tool_schemas.py index 1d583a33b..6468f4f45 100644 --- a/src/tool_schemas.py +++ b/src/tool_schemas.py @@ -1292,14 +1292,14 @@ FUNCTION_TOOL_SCHEMAS = [ "type": "function", "function": { "name": "manage_settings", - "description": "Manage user preferences and settings. Use `disable_tool`/`enable_tool`/`list_tools` to turn individual tools on or off globally (e.g. shell, search, browser, documents, memory, skills, images, tasks, notes, calendar, email). Use list/get/set/delete for free-form preferences.", + "description": "Manage user preferences and settings. Use `disable_tool`/`list_tools` to turn individual tools off globally or list what is disabled (e.g. shell, search, browser, documents, memory, skills, images, tasks, notes, calendar, email). Disabled tools can only be re-enabled by the user in Settings > Agent Tools, never from chat. Use list/get/set/delete for free-form preferences.", "parameters": { "type": "object", "properties": { - "action": {"type": "string", "enum": ["list", "get", "set", "delete", "disable_tool", "enable_tool", "list_tools"]}, + "action": {"type": "string", "enum": ["list", "get", "set", "delete", "disable_tool", "list_tools"]}, "key": {"type": "string", "description": "Setting key (for get/set/delete)"}, "value": {"description": "Setting value (for set) — can be string, number, boolean, or object"}, - "tool": {"type": "string", "description": "Tool name to disable/enable (for disable_tool/enable_tool). Accepts aliases: shell, search, browser, documents, memory, skills, images, tasks, notes, calendar, email — or a raw tool name like 'bash' or 'web_search'."} + "tool": {"type": "string", "description": "Tool name to disable (for disable_tool). Accepts aliases: shell, search, browser, documents, memory, skills, images, tasks, notes, calendar, email — or a raw tool name like 'bash' or 'web_search'."} }, "required": ["action"] } diff --git a/src/tools/system.py b/src/tools/system.py index 9b0e6d1bb..09ca1fd41 100644 --- a/src/tools/system.py +++ b/src/tools/system.py @@ -769,6 +769,8 @@ _APP_API_BLOCKLIST_PREFIXES = ( # {"tasks": []} to /api/cookbook/state, which overwrote the whole file. # Use dedicated tools or UI flows instead. _APP_API_BLOCKLIST_METHOD_PATH = ( + ("POST", "/api/tools"), # global tool policy is changed by a human in Settings + ("PATCH", "/api/mcp/servers"), # per-server MCP tool policy and server toggles, same reason ("GET", "/api/email/accounts"), # owner-filtered in tool context; use list_email_accounts MCP tool ("POST", "/api/cookbook/state"), # whole-file overwrite — agent must use serve_preset/serve_model instead ("DELETE", "/api/cookbook/state"), @@ -893,6 +895,10 @@ async def do_app_api(content: str, owner: Optional[str] = None) -> Dict: if method not in ("GET", "POST", "PUT", "PATCH", "DELETE"): return {"error": f"Unsupported method: {method}", "exit_code": 1} if any(method == m and path.startswith(p) for m, p in _APP_API_BLOCKLIST_METHOD_PATH): + if path.startswith("/api/tools"): + return {"error": "Tool policy cannot be replaced from chat. Re-enable tools in Settings > Agent Tools; use manage_settings to list or disable them.", "exit_code": 1} + if path.startswith("/api/mcp/servers"): + return {"error": "MCP server and tool policy cannot be changed from chat. Ask the user to change it in Settings > Integrations.", "exit_code": 1} if "/api/email/accounts" in path: return {"error": "Don't use /api/email/accounts via app_api — it is owner-filtered in tool context and may return empty. Use the `list_email_accounts` email tool, then pass `account` to list_emails/read_email.", "exit_code": 1} if "/api/cookbook/packages/install" in path: diff --git a/tests/test_manage_settings_denylist_hard.py b/tests/test_manage_settings_denylist_hard.py new file mode 100644 index 000000000..790710e14 --- /dev/null +++ b/tests/test_manage_settings_denylist_hard.py @@ -0,0 +1,227 @@ +"""The global disabled_tools denylist must be a hard boundary for the agent. + +``manage_settings`` used to expose ``enable_tool``, which let any agent turn +(admin-run, but reading arbitrary web pages, emails, and documents) remove +entries from ``settings.json:disabled_tools`` — the only global tool +permission mechanism the Settings > Agent Tools panel offers. A prompt +injection could therefore re-enable bash, email, or any other tool an admin +had switched off. Regression for issue #5523: the agent may still narrow the +denylist (``disable_tool``) and inspect it (``list_tools``), but widening it +is refused and left to a human in the Settings panel. +""" + +import asyncio +import json + +import pytest + +import src.settings as settings_mod +from src.agent_tools.admin_tools import do_manage_settings + + +@pytest.fixture +def store(monkeypatch): + state = {"disabled_tools": ["bash", "manage_memory", "web_search", "web_fetch"]} + monkeypatch.setattr(settings_mod, "load_settings", lambda: dict(state)) + + def _save(s): + state.clear() + state.update(s) + + monkeypatch.setattr(settings_mod, "save_settings", _save) + monkeypatch.setattr( + settings_mod, "get_setting", + lambda key, default=None: state.get(key, default), + ) + return state + + +def _call(payload, owner="admin"): + return asyncio.run(do_manage_settings(json.dumps(payload), owner=owner)) + + +@pytest.mark.parametrize("tool", ["shell", "bash", "memory", "manage_memory", "search"]) +def test_enable_tool_is_refused_and_denylist_unchanged(store, tool): + before = list(store["disabled_tools"]) + result = _call({"action": "enable_tool", "tool": tool}) + assert result["exit_code"] == 1, result + assert "Settings" in result["error"] + assert store["disabled_tools"] == before + + +def test_enable_tool_refused_without_tool_name(store): + before = list(store["disabled_tools"]) + result = _call({"action": "enable_tool"}) + assert result["exit_code"] == 1, result + assert store["disabled_tools"] == before + + +def test_enable_tool_refused_via_name_alias_argument(store): + # The toggle branch also accepted `name` as the tool argument. + before = list(store["disabled_tools"]) + result = _call({"action": "enable_tool", "name": "shell"}) + assert result["exit_code"] == 1, result + assert store["disabled_tools"] == before + + +def test_disable_tool_still_narrows_the_denylist(store): + result = _call({"action": "disable_tool", "tool": "images"}) + assert result["exit_code"] == 0, result + assert "generate_image" in store["disabled_tools"] + # Nothing that was disabled before is re-enabled as a side effect. + for name in ("bash", "manage_memory", "web_search", "web_fetch"): + assert name in store["disabled_tools"] + + +def test_list_tools_still_reports_the_denylist(store): + result = _call({"action": "list_tools"}) + assert result["exit_code"] == 0, result + assert set(result["disabled"]) == {"bash", "manage_memory", "web_search", "web_fetch"} + assert "Settings > Agent Tools" in result["response"] + + +@pytest.mark.parametrize("action", ["set", "delete", "reset"]) +def test_disabled_tools_key_is_not_reachable_through_settings_actions(store, action): + # Belt and braces: the free-form settings actions must not offer a second + # path to the denylist (a `reset` would restore the default empty list). + before = list(store["disabled_tools"]) + payload = {"action": action, "key": "disabled_tools"} + if action == "set": + payload["value"] = [] + result = _call(payload) + assert result["exit_code"] == 1, result + assert store["disabled_tools"] == before + + +@pytest.mark.parametrize("action", ["set", "delete", "reset"]) +def test_disabled_tools_stays_protected_if_added_to_defaults(store, monkeypatch, action): + # The old rejection depended on disabled_tools not being in DEFAULT_SETTINGS. + monkeypatch.setitem(settings_mod.DEFAULT_SETTINGS, "disabled_tools", []) + before = list(store["disabled_tools"]) + result = _call({"action": action, "key": "disabled_tools", "value": []}) + assert result["exit_code"] == 1, result + assert store["disabled_tools"] == before + + +@pytest.mark.parametrize("path", ["/api/tools", "api/tools", "/api/tools/", "/api/tools?source=chat"]) +def test_app_api_cannot_replace_global_tool_policy(monkeypatch, path): + import httpx + from src.tools.system import do_app_api + + def no_loopback(*args, **kwargs): + raise AssertionError("Tool policy write reached the HTTP client") + + monkeypatch.setattr(httpx, "AsyncClient", no_loopback) + result = asyncio.run(do_app_api(json.dumps({ + "action": "call", "method": "post", "path": path, "body": {"disabled": []}, + }), owner="admin")) + assert result["exit_code"] == 1 + assert "Settings > Agent Tools" in result["error"] + + +@pytest.mark.parametrize("internal", [True, False]) +def test_tools_api_requires_human_admin_for_policy_changes(store, monkeypatch, internal): + from types import SimpleNamespace + from starlette.requests import Request + from fastapi import HTTPException + from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN + import routes.model_routes as routes + + monkeypatch.setenv("AUTH_ENABLED", "true") + monkeypatch.setattr(routes, "_load_settings", lambda: dict(store)) + monkeypatch.setattr(routes, "_save_settings", lambda value: store.update(value)) + router = routes.setup_model_routes(SimpleNamespace()) + endpoint = next(r.endpoint for r in router.routes if r.path == "/api/tools" and "POST" in r.methods) + body = endpoint.__annotations__["body"](disabled=[]) + headers = [(INTERNAL_TOOL_HEADER.lower().encode(), INTERNAL_TOOL_TOKEN.encode())] if internal else [] + request = Request({ + "type": "http", "method": "POST", "path": "/api/tools", "headers": headers, + "state": {"current_user": "admin"}, + "app": SimpleNamespace(state=SimpleNamespace(auth_manager=SimpleNamespace( + is_configured=True, is_admin=lambda user: user == "admin", + ))), + }) + if internal: + before = list(store["disabled_tools"]) + with pytest.raises(HTTPException) as error: + endpoint(body, request) + assert error.value.status_code == 403 + assert store["disabled_tools"] == before + else: + assert endpoint(body, request) == {"ok": True, "disabled": []} + assert store["disabled_tools"] == [] + + +def test_schema_no_longer_advertises_enable_tool(): + import src.agent_tools # noqa: F401 (resolves the schema import cycle) + from src.tool_schemas import FUNCTION_TOOL_SCHEMAS + + schema = next( + t["function"] for t in FUNCTION_TOOL_SCHEMAS + if (t.get("function") or {}).get("name") == "manage_settings" + ) + actions = schema["parameters"]["properties"]["action"]["enum"] + assert "enable_tool" not in actions + assert "disable_tool" in actions and "list_tools" in actions + assert "enable_tool" not in schema["description"] + + +@pytest.mark.parametrize("path", [ + "/api/mcp/servers/srv1/tools", "api/mcp/servers/srv1/tools", "/api/mcp/servers/srv1", +]) +def test_app_api_cannot_change_mcp_tool_policy(monkeypatch, path): + import httpx + from src.tools.system import do_app_api + + def no_loopback(*args, **kwargs): + raise AssertionError("MCP tool policy write reached the HTTP client") + + monkeypatch.setattr(httpx, "AsyncClient", no_loopback) + result = asyncio.run(do_app_api(json.dumps({ + "action": "call", "method": "patch", "path": path, "body": {"disabled": []}, + }), owner="admin")) + assert result["exit_code"] == 1 + assert "Settings > Integrations" in result["error"] + + +@pytest.mark.parametrize("internal", [True, False]) +def test_mcp_tools_api_requires_human_admin_for_policy_changes(monkeypatch, internal): + from types import SimpleNamespace + from unittest.mock import MagicMock + from starlette.requests import Request + from fastapi import HTTPException + from core.middleware import INTERNAL_TOOL_HEADER, INTERNAL_TOOL_TOKEN + from routes.mcp import mcp_routes + + server = SimpleNamespace(id="srv1", disabled_tools=json.dumps(["send_email"])) + session = MagicMock() + session.query.return_value.filter.return_value.first.return_value = server + monkeypatch.setenv("AUTH_ENABLED", "true") + monkeypatch.setattr(mcp_routes, "SessionLocal", lambda: session) + router = mcp_routes.setup_mcp_routes(MagicMock()) + # setup_mcp_routes appends to the shared module router; take the latest. + endpoint = [r.endpoint for r in router.routes if getattr(r, "name", None) == "update_disabled_tools"][-1] + headers = [(b"content-type", b"application/json")] + if internal: + headers.append((INTERNAL_TOOL_HEADER.lower().encode(), INTERNAL_TOOL_TOKEN.encode())) + + async def receive(): + return {"type": "http.request", "body": b'{"disabled": []}', "more_body": False} + + request = Request({ + "type": "http", "method": "PATCH", "path": "/api/mcp/servers/srv1/tools", "headers": headers, + "state": {"current_user": "admin"}, + "app": SimpleNamespace(state=SimpleNamespace(auth_manager=SimpleNamespace( + is_configured=True, is_admin=lambda user: user == "admin", + ))), + }, receive) + if internal: + with pytest.raises(HTTPException) as error: + asyncio.run(endpoint("srv1", request)) + assert error.value.status_code == 403 + assert server.disabled_tools == json.dumps(["send_email"]) + session.commit.assert_not_called() + else: + assert asyncio.run(endpoint("srv1", request)) == {"id": "srv1", "disabled_count": 0} + assert server.disabled_tools is None + session.commit.assert_called_once() diff --git a/tests/test_review_regressions.py b/tests/test_review_regressions.py index f8844f2cd..8cc7c5a79 100644 --- a/tests/test_review_regressions.py +++ b/tests/test_review_regressions.py @@ -1050,12 +1050,14 @@ async def test_disable_tool_email_covers_full_builtin_set(monkeypatch): assert tool_name in disabled, tool_name assert f"mcp__email__{tool_name}" in disabled, tool_name - # enable_tool email must remove the full set again. + # enable_tool is refused from chat: the denylist is admin policy and can + # only be widened in Settings > Agent Tools (issue #5523). The full set + # must stay disabled. result = await do_manage_settings( '{"action": "enable_tool", "tool": "email"}', owner="admin" ) - assert result["exit_code"] == 0 - assert store["disabled_tools"] == [] + assert result["exit_code"] == 1 + assert set(store["disabled_tools"]) == disabled def _install_admin_auth_stub(monkeypatch): diff --git a/tests/test_tool_policy.py b/tests/test_tool_policy.py index d9e9c75f8..37af8ed77 100644 --- a/tests/test_tool_policy.py +++ b/tests/test_tool_policy.py @@ -3087,7 +3087,7 @@ def test_qwen_explicit_admin_requests_build_safe_args(): ) assert al._parse_qwen_explicit_admin_request("turn image generation back on now") == ( "manage_settings", - json.dumps({"action": "enable_tool", "tool": "images"}), + json.dumps({"action": "list_tools"}), ) assert al._parse_qwen_explicit_admin_request("list API tokens by name and prefix only") == ( "manage_tokens",