fix(tools): gate MCP resource and prompt registration behind enabledTools

The enabledTools allowlist was only enforced for MCP tools returned by
session.list_tools(). Resources and prompts from session.list_resources()
and session.list_prompts() were registered unconditionally, allowing a
deny-all or restrictive enabledTools config to leak resource and prompt
capabilities to the model.

Now resources and prompts are only registered when allow_all_tools is
true (default ["*"] wildcard). Any explicit tool restriction — including
enabledTools: [] (deny-all) or a list of specific tool names — also
blocks resource and prompt registration from that server.

Fixes #4435
This commit is contained in:
michaelxer
2026-06-25 16:10:37 +08:00
committed by Xubin Ren
parent f60b3c7920
commit 246ea8ef61
2 changed files with 114 additions and 21 deletions
+41 -21
View File
@@ -797,31 +797,51 @@ async def connect_mcp_servers(
", ".join(available_wrapped_names) or "(none)",
)
try:
resources_result = await session.list_resources()
for resource in resources_result.resources:
wrapper = MCPResourceWrapper(
session, name, resource, resource_timeout=cfg.tool_timeout
)
registry.register(wrapper)
registered_count += 1
# Only register resources and prompts when no tool restriction is
# active. enabledTools is a per-*tool* allowlist; resources and
# prompts have no equivalent name filter, so they must be skipped
# whenever the operator specified a tool subset. An empty list
# (deny-all) or a list of specific tool names both indicate that
# the operator intended to restrict capabilities — registering
# unrestricted resource/prompt wrappers would violate that intent.
# The default ["*"] (allow-all) means no restriction was intended.
register_extras = allow_all_tools
if register_extras:
try:
resources_result = await session.list_resources()
for resource in resources_result.resources:
wrapper = MCPResourceWrapper(
session, name, resource, resource_timeout=cfg.tool_timeout
)
registry.register(wrapper)
registered_count += 1
logger.debug(
"MCP: registered resource '{}' from server '{}'",
wrapper.name,
name,
)
except Exception as e:
logger.debug(
"MCP: registered resource '{}' from server '{}'", wrapper.name, name
"MCP server '{}': resources not supported or failed: {}", name, e
)
except Exception as e:
logger.debug("MCP server '{}': resources not supported or failed: {}", name, e)
try:
prompts_result = await session.list_prompts()
for prompt in prompts_result.prompts:
wrapper = MCPPromptWrapper(
session, name, prompt, prompt_timeout=cfg.tool_timeout
try:
prompts_result = await session.list_prompts()
for prompt in prompts_result.prompts:
wrapper = MCPPromptWrapper(
session, name, prompt, prompt_timeout=cfg.tool_timeout
)
registry.register(wrapper)
registered_count += 1
logger.debug(
"MCP: registered prompt '{}' from server '{}'",
wrapper.name,
name,
)
except Exception as e:
logger.debug(
"MCP server '{}': prompts not supported or failed: {}", name, e
)
registry.register(wrapper)
registered_count += 1
logger.debug("MCP: registered prompt '{}' from server '{}'", wrapper.name, name)
except Exception as e:
logger.debug("MCP server '{}': prompts not supported or failed: {}", name, e)
logger.info(
"MCP server '{}': connected, {} capabilities registered", name, registered_count
+73
View File
@@ -433,6 +433,79 @@ async def test_connect_mcp_servers_enabled_tools_empty_list_registers_none(
assert registry.tool_names == []
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_empty_list_blocks_resources_and_prompts(
fake_mcp_runtime: dict[str, object | None],
) -> None:
"""enabledTools: [] (deny-all) must also block resource and prompt registration."""
fake_mcp_runtime["session"] = _make_fake_session_with_capabilities(
tool_names=["demo"],
resource_names=["secret_data"],
prompt_names=["admin_prompt"],
)
registry = ToolRegistry()
stacks = await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=[])},
registry,
)
for stack in stacks.values():
await stack.aclose()
assert registry.tool_names == []
# Resources and prompts must also be blocked
assert not any("secret_data" in name for name in registry.tool_names)
assert not any("admin_prompt" in name for name in registry.tool_names)
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_specific_list_blocks_resources_and_prompts(
fake_mcp_runtime: dict[str, object | None],
) -> None:
"""enabledTools with specific tool names must not leak resources or prompts."""
fake_mcp_runtime["session"] = _make_fake_session_with_capabilities(
tool_names=["demo", "other"],
resource_names=["secret_data"],
prompt_names=["admin_prompt"],
)
registry = ToolRegistry()
stacks = await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=["demo"])},
registry,
)
for stack in stacks.values():
await stack.aclose()
# Only the allowed tool should be registered
assert "mcp_test_demo" in registry.tool_names
assert "mcp_test_other" not in registry.tool_names
# Resources and prompts must not leak
assert not any("secret_data" in name for name in registry.tool_names)
assert not any("admin_prompt" in name for name in registry.tool_names)
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_wildcard_allows_resources_and_prompts(
fake_mcp_runtime: dict[str, object | None],
) -> None:
"""enabledTools: ['*'] should allow all tools, resources, and prompts."""
fake_mcp_runtime["session"] = _make_fake_session_with_capabilities(
tool_names=["demo"],
resource_names=["public_data"],
prompt_names=["help_prompt"],
)
registry = ToolRegistry()
stacks = await connect_mcp_servers(
{"test": MCPServerConfig(command="fake", enabled_tools=["*"])},
registry,
)
for stack in stacks.values():
await stack.aclose()
assert "mcp_test_demo" in registry.tool_names
assert any("public_data" in name for name in registry.tool_names)
assert any("help_prompt" in name for name in registry.tool_names)
@pytest.mark.asyncio
async def test_connect_mcp_servers_enabled_tools_warns_on_unknown_entries(
fake_mcp_runtime: dict[str, object | None], monkeypatch: pytest.MonkeyPatch