Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 4 additions & 1 deletion docs/guide.md
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
23 changes: 20 additions & 3 deletions main.py
Original file line number Diff line number Diff line change
Expand Up @@ -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],
Expand All @@ -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)

Expand Down
76 changes: 76 additions & 0 deletions tests/test_host_telemetry.py
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
Loading