fix(mcp): close three provider MCP admission gaps

- Deny an OpenAI allowed_tools filter with an empty tool_names list,
  which OpenAI reads as every tool on the server
- Fail closed when a key's tool grant cannot be read, instead of
  admitting every tool the registration allows
- Check MCP tools carried inside Responses input items such as
  tool_search_output, like top-level tools
This commit is contained in:
Mrinal Chanshetty 2026-11-03 21:50:51 -08:00
parent 1da68c1ada
commit 46fd07cc07
4 changed files with 72 additions and 6 deletions

View file

@ -53,6 +53,9 @@ def _openai_attachment(tool: Mapping[str, object]) -> ProviderMCPAttachment | MC
return MCPAdmissionError("Provider MCP connector IDs are not supported by URL registrations")
allowed: Final = tool.get("allowed_tools")
names: Final = _OBJECT.validate_python(allowed).get("tool_names") if isinstance(allowed, Mapping) else allowed
if isinstance(allowed, Mapping) and not names:
# OpenAI reads an empty tool_names filter as no filter at all, i.e. every tool on the server.
return MCPAdmissionError("Provider MCP tool filters require a nonempty tool_names list")
return _attachment(tool.get("server_url"), _names(names))
@ -144,7 +147,9 @@ async def authorize_provider_mcp(
async def _granted_tools(registration: LiteLLM_MCPServerTable, auth: UserAPIKeyAuth) -> frozenset[str]:
permitted: Final = await MCPRequestHandler.get_allowed_tools_for_server(registration.server_id, auth)
permitted: Final = await MCPRequestHandler.get_allowed_tools_for_server(
registration.server_id, auth, fail_closed=True
)
ceiling: Final = frozenset(registration.allowed_tools or ())
return ceiling if permitted is None else ceiling.intersection(permitted)

View file

@ -3246,6 +3246,7 @@ class MCPRequestHandler:
user_api_key_auth: UserAPIKeyAuth | None = None,
*,
keyless_source: bool = False,
fail_closed: bool = False,
) -> list[str] | None:
"""
Get list of allowed tool names for a specific server based on key/team permissions.
@ -3254,6 +3255,8 @@ class MCPRequestHandler:
Args:
server_id: Server ID to check permissions for
user_api_key_auth: User auth
fail_closed: Deny every tool ([]) when the lookup fails, instead of the legacy allow-all.
Provider-native admission sets it, because None there means every registered tool.
Returns:
List[str] if restrictions exist, None if no restrictions (allow all)
@ -3295,7 +3298,8 @@ class MCPRequestHandler:
# keyless_source AND the marker are needed: each source resolves through an UNMARKED auth, so
# without keyless_source a fault under a source returns None and wins the union as allow-all.
deny_all: Final = (
unreadable_entitlement
fail_closed
or unreadable_entitlement
or keyless_source
or is_mcp_admitted_user_subject(user_api_key_auth)
or personal_key_inherits_team_grants(user_api_key_auth)

View file

@ -11,6 +11,7 @@ still executes the tool, just with no credentials.
from collections.abc import Iterable, Mapping
from copy import deepcopy
from dataclasses import dataclass
from itertools import chain
from types import MappingProxyType
from typing import TYPE_CHECKING, Any, Final
@ -42,11 +43,11 @@ async def authorize_provider_mcp_request(
effective: Final = (
{**body, **TypeAdapter(dict[str, object]).validate_python(extra)} if isinstance(extra, Mapping) else body
)
tools: Final = effective.get("tools")
has_native_tools: Final = isinstance(tools, (list, tuple)) and any(
tools: Final = (*_tool_list(effective.get("tools")), *_input_item_tools(effective.get("input")))
has_native_tools: Final = any(
isinstance(tool, Mapping)
and TypeAdapter(dict[str, object]).validate_python(tool).get("type") in ("mcp", "mcp_toolset", "url")
for tool in TypeAdapter(tuple[object, ...]).validate_python(tools)
for tool in tools
)
if not has_native_tools and not effective.get("mcp_servers"):
return
@ -56,7 +57,23 @@ async def authorize_provider_mcp_request(
return
from litellm.proxy.mcp_server.auth.provider_mcp import authorize_provider_mcp
await authorize_provider_mcp(effective, auth, provider_bound=provider_bound)
await authorize_provider_mcp({**effective, "tools": list(tools)}, auth, provider_bound=provider_bound)
def _tool_list(value: object) -> tuple[object, ...]:
return TypeAdapter(tuple[object, ...]).validate_python(value) if isinstance(value, (list, tuple)) else ()
def _input_item_tools(value: object) -> tuple[object, ...]:
"""Tool definitions carried inside Responses input items, such as ``tool_search_output``; the provider
loads them like top-level tools, so they need the same admission."""
return tuple(
chain.from_iterable(
_tool_list(TypeAdapter(dict[str, object]).validate_python(item).get("tools"))
for item in _tool_list(value)
if isinstance(item, Mapping)
)
)
@dataclass(frozen=True, slots=True)

View file

@ -201,6 +201,46 @@ async def test_duplicate_url_registration_neither_shadows_nor_widens_the_granted
assert denied.value.code == "403"
async def test_empty_openai_tool_name_filter_is_denied_because_openai_reads_it_as_every_tool(
native_auth: UserAPIKeyAuth,
) -> None:
with pytest.raises(ProxyException) as denied:
await authorize_provider_mcp(_openai({"tool_names": []}), native_auth)
assert (denied.value.code, denied.value.message) == (
"403",
"Provider MCP tool filters require a nonempty tool_names list",
)
async def test_failed_tool_grant_lookup_denies_instead_of_allowing_every_registered_tool(
native_auth: UserAPIKeyAuth, monkeypatch: pytest.MonkeyPatch
) -> None:
from litellm.proxy.mcp_server.auth.user_api_key_auth_mcp import MCPRequestHandler
async def unavailable(*_: object, **__: object) -> list[str] | None:
raise RuntimeError("policy store unavailable")
monkeypatch.setattr(MCPRequestHandler, "_single_credential_tools", staticmethod(unavailable))
monkeypatch.setattr(MCPRequestHandler, "resolve_personal_key_team_tools", staticmethod(unavailable))
with pytest.raises(ProxyException) as denied:
await authorize_provider_mcp(_openai(["fetch"]), native_auth)
assert denied.value.code == "403"
@pytest.mark.parametrize("tools,admitted", [(["search"], True), (["delete"], False)])
async def test_mcp_tools_carried_in_responses_input_items_are_admitted_like_top_level_tools(
native_auth: UserAPIKeyAuth, tools: list[str], admitted: bool
) -> None:
body: Final = {"input": [{"type": "tool_search_output", "tools": _openai(tools)["tools"]}]}
context: Final = {"litellm_metadata": {"user_api_key_auth": native_auth}}
if admitted:
await authorize_provider_mcp_request(body, context)
return
with pytest.raises(ProxyException) as denied:
await authorize_provider_mcp_request(body, context)
assert denied.value.code == "403"
async def test_canonical_host_and_default_port_match(native_auth: UserAPIKeyAuth) -> None:
body: Final = _openai(["search"], "https://NATIVE.example:443/mcp")
await authorize_provider_mcp(body, native_auth)