diff --git a/src/agent_loop.py b/src/agent_loop.py index c452c6c22..257cc441d 100644 --- a/src/agent_loop.py +++ b/src/agent_loop.py @@ -15204,7 +15204,7 @@ _ADMIN_TOOLS = { "manage_documents", "manage_settings", "create_session", "list_sessions", "send_to_session", "search_chats", "pipeline", "ask_teacher", "list_models", "list_cached_models", "list_downloads", "list_cookbook_servers", - "list_serve_presets", "list_served_models", + "list_serve_presets", "list_served_models", "tail_serve_output", "search_hf_models", } def _build_base_prompt( diff --git a/src/tool_execution.py b/src/tool_execution.py index bc489b878..fd6dcfbbd 100644 --- a/src/tool_execution.py +++ b/src/tool_execution.py @@ -1288,6 +1288,16 @@ _ADMIN_TOOLS = { "serve_preset", "stop_served_model", "cancel_download", + # Read-only Cookbook tools reach require_admin routes over the internal + # loopback without an owner check (see NON_ADMIN_BLOCKED_TOOLS in + # src/tool_security.py), so they are admin-only at dispatch as well. + "list_served_models", + "tail_serve_output", + "list_downloads", + "search_hf_models", + "list_cached_models", + "list_serve_presets", + "list_cookbook_servers", } diff --git a/src/tool_security.py b/src/tool_security.py index f754a6356..418b1a54b 100644 --- a/src/tool_security.py +++ b/src/tool_security.py @@ -79,6 +79,22 @@ NON_ADMIN_BLOCKED_TOOLS = BUILTIN_EMAIL_TOOLS | { "stop_served_model", "cancel_download", "adopt_served_model", + # Read-only Cookbook inspection tools. They look harmless, but every one + # of them calls a require_admin route over the internal loopback with the + # process-wide INTERNAL_TOOL_TOKEN and never checks the session owner, so + # a non-admin agent would inherit admin reach through them: + # list_cached_models forwards a caller-chosen `host` into + # /api/model/cached, which SSHes to that host with the app user's keys and + # walks caller-chosen `model_dir`s; list_served_models / tail_serve_output + # expose serve/download logs and running process command lines. Model + # serving is admin-only in THREAT_MODEL.md, so block the read side too. + "list_served_models", + "tail_serve_output", + "list_downloads", + "search_hf_models", + "list_cached_models", + "list_serve_presets", + "list_cookbook_servers", } diff --git a/tests/test_cookbook_readonly_tools_admin_gate.py b/tests/test_cookbook_readonly_tools_admin_gate.py new file mode 100644 index 000000000..2e683aa6f --- /dev/null +++ b/tests/test_cookbook_readonly_tools_admin_gate.py @@ -0,0 +1,136 @@ +"""Read-only Cookbook tools must be admin-only for the agent. + +Regression: `list_cached_models`, `list_served_models`, `tail_serve_output`, +`list_downloads`, `search_hf_models`, `list_serve_presets` and +`list_cookbook_servers` sat in the plan-mode read-only allowlist but not in +NON_ADMIN_BLOCKED_TOOLS or the dispatcher's _ADMIN_TOOLS. Their +implementations call require_admin routes over the internal loopback with the +process-wide INTERNAL_TOOL_TOKEN and never look at the owner, so a non-admin +user's agent could reach admin capability through them. The worst case is +`list_cached_models`: a caller-chosen `host` is forwarded to +/api/model/cached, which SSHes to that host with the app user's keys and +walks caller-chosen `model_dir`s. THREAT_MODEL.md lists model serving as +admin-only, so the read side has to be gated the same way as serve/download. +""" + +import sys +import types +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest + +from src.tool_security import ( + NON_ADMIN_BLOCKED_TOOLS, + blocked_tools_for_owner, + delegated_credential_blocked_tools, +) +from src.tool_execution import _ADMIN_TOOLS +from src.agent_runtime.authority import OperationGrant, RequestAuthority + +READONLY_COOKBOOK_TOOLS = ( + "list_cached_models", + "list_served_models", + "tail_serve_output", + "list_downloads", + "search_hf_models", + "list_serve_presets", + "list_cookbook_servers", +) + + +def _install_core_auth_stub(monkeypatch, *, is_admin: bool): + """Narrow auth surface: AUTH enabled, configured, owner admin or not.""" + core_mod = types.ModuleType("core") + core_mod.__path__ = [] + auth_mod = types.ModuleType("core.auth") + + class FakeAuth: + is_configured = True + + def is_admin(self, username): + return is_admin + + auth_mod.AuthManager = lambda: FakeAuth() + core_mod.auth = auth_mod + monkeypatch.setitem(sys.modules, "core", core_mod) + monkeypatch.setitem(sys.modules, "core.auth", auth_mod) + monkeypatch.setattr("src.auth_helpers._auth_disabled", lambda: False) + return auth_mod + + +def test_readonly_cookbook_tools_are_in_every_admin_gate(): + from src.agent_loop import _ADMIN_TOOLS as prompt_admin_tools + + # Spelled out (not imported from a shared tuple in src/) so dropping one + # from either set fails here instead of silently reopening the hole. + for tool in READONLY_COOKBOOK_TOOLS: + assert tool in NON_ADMIN_BLOCKED_TOOLS, tool + assert tool in _ADMIN_TOOLS, tool + assert tool in prompt_admin_tools, tool + + +def test_readonly_cookbook_tools_hidden_from_non_admin_and_bearer_runs(monkeypatch): + _install_core_auth_stub(monkeypatch, is_admin=False) + hidden = blocked_tools_for_owner("regular-user") + for tool in READONLY_COOKBOOK_TOOLS: + assert tool in hidden, tool + assert tool in delegated_credential_blocked_tools(), tool + + +def test_admin_owner_keeps_readonly_cookbook_tools(monkeypatch): + _install_core_auth_stub(monkeypatch, is_admin=True) + assert blocked_tools_for_owner("admin-user") == set() + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tool_name", READONLY_COOKBOOK_TOOLS) +async def test_non_admin_agent_cannot_execute_readonly_cookbook_tools(monkeypatch, tool_name): + _install_core_auth_stub(monkeypatch, is_admin=False) + from src.tool_execution import NO_TOOL_SECURITY_CONTEXT, execute_tool_block + + # The loopback must never be touched: if the gate fails, the tool would + # hit a require_admin route with the internal token. + import httpx + + def _boom(*args, **kwargs): + raise AssertionError(f"{tool_name} reached the internal loopback for a non-admin owner") + + monkeypatch.setattr(httpx, "AsyncClient", _boom) + monkeypatch.setattr(httpx, "get", _boom) + monkeypatch.setattr(httpx, "post", _boom) + + payload = '{"host": "root@10.0.0.5", "ssh_port": "22", "model_dir": "/root,/etc"}' + desc, result = await execute_tool_block( + SimpleNamespace(tool_type=tool_name, content=payload), + owner="regular-user", + security_context=NO_TOOL_SECURITY_CONTEXT, + request_authority=RequestAuthority( + "cookbook-test", "regular-user", "", "", (OperationGrant(tool_name),) + ), + ) + assert desc == f"{tool_name}: BLOCKED" + assert result["exit_code"] == 1 + assert "admin" in result["error"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tool_name", READONLY_COOKBOOK_TOOLS) +async def test_admin_agent_can_dispatch_readonly_cookbook_tools(monkeypatch, tool_name): + from src import tool_implementations + from src.tool_execution import NO_TOOL_SECURITY_CONTEXT, execute_tool_block + + _install_core_auth_stub(monkeypatch, is_admin=True) + handler = AsyncMock(return_value={"output": "Cookbook result", "exit_code": 0}) + monkeypatch.setattr(tool_implementations, f"do_{tool_name}", handler) + desc, result = await execute_tool_block( + SimpleNamespace(tool_type=tool_name, content="{}"), + owner="admin-user", + security_context=NO_TOOL_SECURITY_CONTEXT, + request_authority=RequestAuthority( + "cookbook-test", "admin-user", "", "", (OperationGrant(tool_name),) + ), + ) + assert desc == tool_name + assert result["exit_code"] == 0 + handler.assert_awaited_once_with("{}", owner="admin-user")