diff --git a/docs/guide.md b/docs/guide.md index 856eb8e..cbb1afa 100644 --- a/docs/guide.md +++ b/docs/guide.md @@ -461,7 +461,10 @@ Both profiles use explicit tools and agent allowlists, deny out-of-catalog calls and execution permissions, and disable ambient instruction/configuration, file-hook, host-Git, memory, and cross-session discovery. Every MCP process, host policy/image loader, and delegated public agent inherits the host-selected -profile. Public access checks verify canonical repository identity and public +profile. MCP permission checks bind each allowlisted tool to its owning server, +accepting both bare and SDK server-qualified tool names. Read-only annotations +do not grant out-of-catalog access, and requests needing managed approval remain +denied. Public access checks verify canonical repository identity and public visibility before content reads and again before releasing results, without a visibility cache. A failed recheck discards the result; after a possible write, the outcome remains unknown rather than implying nothing was written. App diff --git a/main.py b/main.py index a1afc58..aae4d77 100644 --- a/main.py +++ b/main.py @@ -488,6 +488,10 @@ def _session_options( }) tools = [tool for tool in tools if tool.name in custom_names] delegated_agents = {agent["name"] for agent in agents} - {"issuelens"} + mcp_tools = { + server: {name for name in allowed if name.startswith(f"{server}-")} + for server in ("github", "wiki-writer") + } def pre_tool_use( value: PreToolUseHookInput, context: dict[str, str], @@ -513,12 +517,25 @@ def permission( ) -> PermissionRequestResult: kind = getattr(request, "kind", None) name = getattr(request, "tool_name", None) + server = getattr(request, "server_name", None) + server_tools = mcp_tools.get(server, ()) + # SDK MCP requests can already carry the server-qualified tool name. permitted = ( kind == "mcp" - and f"{getattr(request, 'server_name', '')}-{name}" in allowed + and (name in server_tools or f"{server}-{name}" in server_tools) ) or (kind == "custom-tool" and name in custom_names) - if not permitted or getattr(request, "managed_approval_required", False): - logger.info("Session permission denied by the confidentiality profile") + if not permitted: + logger.info( + "Session permission denied by the confidentiality profile: outside the tool catalog" + ) + return PermissionDecisionUserNotAvailable() + if ( + getattr(request, "managed_approval_required", False) + or invocation.get("managed_settings_enabled", False) + ): + logger.info( + "Session permission denied by the confidentiality profile: managed approval required" + ) return PermissionDecisionUserNotAvailable() return PermissionHandler.approve_all(request, invocation) diff --git a/tests/test_host_telemetry.py b/tests/test_host_telemetry.py index 2f8ed3b..7892d0b 100644 --- a/tests/test_host_telemetry.py +++ b/tests/test_host_telemetry.py @@ -695,6 +695,82 @@ def test_permission_handler_never_approves_shell_or_arbitrary_mcp_or_custom_tool ) self.assertEqual(permission(write, {"session_id": "test"}).kind, "user-not-available" if internal else "approve-once") + def test_permission_handler_accepts_bare_and_sdk_qualified_mcp_catalog_names(self): + catalogs = { + "github": [*self.host._GITHUB_READ_TOOLS, *self.host._GITHUB_WRITE_TOOLS], + "wiki-writer": ["wiki-writer-write_wiki_pages", "wiki-writer-patch_wiki_pages"], + } + for internal in (False, True): + options = self.host._session_options({}, [], internal_analysis=internal) + permission = options["on_permission_request"] + for server, tools in catalogs.items(): + for tool in tools: + expected = ( + "approve-once" + if not internal or tool in self.host._GITHUB_READ_TOOLS + else "user-not-available" + ) + for name in (tool.removeprefix(f"{server}-"), tool): + with self.subTest(internal=internal, server=server, tool=name): + # SDK 1.0.29 emits qualified names and readOnly=false + # for the bundled MCP server's unannotated read tools. + request = PermissionRequestMcp.from_dict({ + "kind": "mcp", "serverName": server, "toolName": name, + "toolTitle": "Fixture tool", "readOnly": False, + }) + self.assertEqual( + permission(request, {"session_id": "test"}).kind, expected, + ) + + def test_permission_handler_requires_the_matching_mcp_server_not_just_a_known_tool_name(self): + for internal in (False, True): + permission = self.host._session_options( + {}, [], internal_analysis=internal, + )["on_permission_request"] + for server, tool in ( + ("toolbox", "github-get_repository"), + ("wiki-writer", "github-get_wiki_snapshot"), + ("github", "wiki-writer-write_wiki_pages"), + ("wiki", "writer-write_wiki_pages"), + ("github", "github-github-get_repository"), + ("github", "future_private_read"), + ("github", "github-future_private_read"), + ): + with self.subTest(internal=internal, server=server, tool=tool): + request = PermissionRequestMcp.from_dict({ + "kind": "mcp", "serverName": server, "toolName": tool, + "toolTitle": "Unknown tool", "readOnly": True, + }) + self.assertEqual( + permission(request, {"session_id": "test"}).kind, "user-not-available", + ) + + def test_permission_handler_denies_managed_approvals_without_logging_request_content(self): + for internal in (False, True): + permission = self.host._session_options( + {}, [], internal_analysis=internal, + )["on_permission_request"] + for name in ("get_repository", "github-get_repository"): + for managed_request, managed_session in ((True, False), (False, True)): + with self.subTest( + internal=internal, tool=name, + managed_request=managed_request, managed_session=managed_session, + ): + request = PermissionRequestMcp.from_dict({ + "kind": "mcp", "serverName": "github", "toolName": name, + "toolTitle": PROMPT_SECRET, "readOnly": True, + "args": {"content": ANSWER_SECRET}, + "managedApprovalRequired": managed_request, + }) + with self.assertLogs(HOST_MODULE, level="INFO") as logs: + decision = permission(request, { + "session_id": "test", "managed_settings_enabled": managed_session, + }) + self.assertEqual(decision.kind, "user-not-available") + self.assertIn("managed approval", logs.output[0]) + for secret in (PROMPT_SECRET, ANSWER_SECRET): + self.assertNotIn(secret, "\n".join(logs.output)) + async def test_sdk_create_and_resume_wire_preserve_tool_catalog_and_history_isolation_flags(self): for internal in (False, True): with self.subTest(internal=internal):