fix(mcp): fail closed when a tool's only owners are outside the caller's scope

resolve_tool_route fell through to a scope-blind lookup when the tool had owners but
none intersected the caller's allowed set, so a caller scoped to one server could
resolve a tool served only by another. The downstream name-based permission check would
then misroute the call to a reachable server sharing that name. The scoped branch is now
authoritative: known owners with none in scope returns not_found rather than resolving a
server the caller cannot reach.

Also seed the logging tests from the specific server under test instead of every
manager server id, so adding a second fixture server cannot make a tool look ambiguous.
This commit is contained in:
Tin Chi Lo 2026-07-17 10:18:06 -07:00
parent 85eaa03679
commit 2f7adb7b11
3 changed files with 30 additions and 3 deletions

View file

@ -4883,6 +4883,10 @@ class MCPServerManager:
scoped_server = self.get_mcp_server_by_id(next(iter(in_scope)))
if scoped_server is not None:
return MCPToolRouteResolved(server=scoped_server)
# The tool exists but no owner is reachable by this caller. Fail closed
# rather than fall through to a scope-blind lookup that could hand back a
# server outside the caller's scope.
return MCPToolRouteNotFound(tool_name=tool_name)
server = self._get_mcp_server_from_tool_name(tool_name)
if server is None:
return MCPToolRouteNotFound(tool_name=tool_name)

View file

@ -126,7 +126,9 @@ async def test_mcp_cost_tracking():
)
# Manually add the tool mapping to ensure it's available (since mocking might not capture it properly)
zapier_server_ids = frozenset(local_mcp_server_manager.get_all_mcp_server_ids())
zapier_server = local_mcp_server_manager.get_mcp_server_by_name("zapier_gmail_server")
assert zapier_server is not None
zapier_server_ids = frozenset({zapier_server.server_id})
local_mcp_server_manager.tool_name_to_mcp_server_ids_mapping[
"add_tools"
] = zapier_server_ids
@ -241,7 +243,9 @@ async def test_mcp_cost_tracking_per_tool():
await local_mcp_server_manager._initialize_tool_name_to_mcp_server_ids_mapping()
# Manually add the tool mapping to ensure it's available (since mocking might not capture it properly)
test_server_ids = frozenset(local_mcp_server_manager.get_all_mcp_server_ids())
test_server = local_mcp_server_manager.get_mcp_server_by_name("test_server")
assert test_server is not None
test_server_ids = frozenset({test_server.server_id})
local_mcp_server_manager.tool_name_to_mcp_server_ids_mapping[
"expensive_tool"
] = test_server_ids
@ -406,7 +410,9 @@ async def test_mcp_tool_call_hook():
await local_mcp_server_manager._initialize_tool_name_to_mcp_server_ids_mapping()
# Manually add the tool mapping to ensure it's available (since mocking might not capture it properly)
zapier_server_ids = frozenset(local_mcp_server_manager.get_all_mcp_server_ids())
zapier_server = local_mcp_server_manager.get_mcp_server_by_name("zapier_gmail_server")
assert zapier_server is not None
zapier_server_ids = frozenset({zapier_server.server_id})
local_mcp_server_manager.tool_name_to_mcp_server_ids_mapping["add_tools"] = (
zapier_server_ids
)

View file

@ -3937,6 +3937,23 @@ class TestMCPServerManager:
assert manager._get_mcp_server_from_tool_name("shared-echo") is None
def test_resolve_tool_route_is_not_found_when_no_owner_is_in_scope(self):
"""A tool whose only owner is outside the caller's scope must fail closed.
Falling through to the scope-blind lookup would hand back a server the caller
cannot reach, and the downstream name-based permission check would misroute it
to a reachable same-named server.
"""
manager = MCPServerManager()
alpha = MCPServer(server_id="id-alpha", name="echo_alpha", transport=MCPTransport.http)
zulu = MCPServer(server_id="id-zulu", name="echo_zulu", transport=MCPTransport.http)
manager.registry = {"id-alpha": alpha, "id-zulu": zulu}
manager._register_tool_route("secret_tool", "id-zulu")
route = manager.resolve_tool_route("secret_tool", allowed_server_ids=frozenset({"id-alpha"}))
assert route.kind == "not_found"
def test_resolve_tool_route_names_every_ambiguous_owner(self):
"""The ambiguous route carries all owners so callers can report the real candidates."""
manager = MCPServerManager()