fix(mcp): allow delegate PKCE bypass for internal MCP servers

Remove available_on_public_internet gating from delegate-auth-to-upstream
paths so oauth2 + delegate_auth_to_upstream interactive servers behave
the same when marked internal. Keeps M2M exclusion. Updates tests.
This commit is contained in:
Sameer Kankute
2026-05-15 08:32:09 +05:30
parent 27de6547dc
commit 5aabfccf57
5 changed files with 17 additions and 24 deletions
@@ -332,8 +332,6 @@ class MCPRequestHandler:
# non-bool must not silently enable the bypass.
if getattr(server, "delegate_auth_to_upstream", False) is not True:
return False
if not getattr(server, "available_on_public_internet", True):
return False
# Never delegate for M2M (client_credentials) servers: LiteLLM
# fetches the upstream token automatically using stored credentials,
# so allowing anonymous bypass would let any external caller invoke
@@ -995,9 +995,6 @@ class MCPServerManager:
# unauthenticated caller would get LiteLLM to proxy tool
# calls using its stored client_credentials.
and not server.has_client_credentials
# Internal-only servers must not be reachable from public
# internet callers who happen to carry an upstream token.
and getattr(server, "available_on_public_internet", True)
]
combined_servers.update(delegate_server_ids)
@@ -1578,7 +1578,6 @@ if MCP_AVAILABLE:
_s
and getattr(_s, "auth_type", None) == MCPAuth.oauth2
and getattr(_s, "delegate_auth_to_upstream", False) is True
and getattr(_s, "available_on_public_internet", True)
# M2M servers fetch tokens with stored credentials; never
# expose their /authorize or /token endpoints anonymously.
and not _s.has_client_credentials
@@ -1462,13 +1462,11 @@ class TestMCPDelegateAuthToUpstream:
assert exc_info.value.status_code == 401
mock_auth.assert_called_once()
async def test_delegate_ignored_for_non_public_server(self):
async def test_delegate_bypass_for_internal_server(self):
"""
Internal-only delegate servers must not bypass LiteLLM auth for
anonymous public callers.
Delegate + oauth2 interactive servers bypass LiteLLM auth even when
``available_on_public_internet`` is False (internal MCPs).
"""
from fastapi import HTTPException
from litellm.types.mcp import MCPAuth
from litellm.types.mcp_server.mcp_server_manager import MCPServer
@@ -1489,6 +1487,8 @@ class TestMCPDelegateAuthToUpstream:
)
async def mock_auth_raises(*_args, **_kwargs):
from fastapi import HTTPException
raise HTTPException(status_code=401, detail="No key provided")
with (
@@ -1501,10 +1501,9 @@ class TestMCPDelegateAuthToUpstream:
) as mock_mgr,
):
mock_mgr.get_mcp_server_by_name.return_value = internal_server
with pytest.raises(HTTPException) as exc_info:
await MCPRequestHandler.process_mcp_request(scope)
assert exc_info.value.status_code == 401
mock_auth.assert_called_once()
auth, *_rest = await MCPRequestHandler.process_mcp_request(scope)
mock_auth.assert_not_called()
assert auth.api_key is None
async def test_get_allowed_servers_excludes_client_credentials_delegate(self):
"""
@@ -1551,10 +1550,10 @@ class TestMCPDelegateAuthToUpstream:
assert "pkce-server" in result
assert "m2m-server" not in result
async def test_get_allowed_servers_excludes_non_public_delegate(self):
async def test_get_allowed_servers_includes_internal_delegate(self):
"""
Internal-only (available_on_public_internet=False) delegate servers
must not appear in the anonymous allow-list.
appear in the anonymous allow-list like public delegate servers.
"""
from litellm.proxy._experimental.mcp_server.mcp_server_manager import (
MCPServerManager,
@@ -1593,7 +1592,7 @@ class TestMCPDelegateAuthToUpstream:
result = await manager.get_allowed_mcp_servers(None)
assert "public-server" in result
assert "internal-server" not in result
assert "internal-server" in result
def test_extract_target_server_names_matches_routing_parser(self):
"""
@@ -1696,10 +1696,10 @@ class TestTemporaryMCPSessionEndpoints:
assert call_kwargs["api_key"] == ""
@pytest.mark.asyncio
async def test_mcp_oauth_user_api_key_auth_requires_public_server_for_delegate_bypass(
async def test_mcp_oauth_user_api_key_auth_internal_delegate_bypasses(
self,
):
"""Internal-only delegate servers must still require LiteLLM auth."""
"""Internal-only delegate servers still get anonymous PKCE /authorize bypass."""
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
_mcp_oauth_user_api_key_auth,
)
@@ -1711,6 +1711,8 @@ class TestTemporaryMCPSessionEndpoints:
mock_request.headers = {}
mock_request.cookies = {}
mock_request.path_params = {"server_id": "server-1"}
# Real path so ``endswith("/token")`` is not fooled by MagicMock truthiness.
mock_request.url = types.SimpleNamespace(path="/server-1/authorize")
internal_server = MagicMock()
internal_server.auth_type = MCPAuth.oauth2
internal_server.delegate_auth_to_upstream = True
@@ -1742,10 +1744,8 @@ class TestTemporaryMCPSessionEndpoints:
):
result = await _mcp_oauth_user_api_key_auth(mock_request)
assert result is expected_auth
auth_builder_mock.assert_awaited_once()
_, call_kwargs = auth_builder_mock.call_args
assert call_kwargs["api_key"] == ""
assert isinstance(result, UserAPIKeyAuth)
auth_builder_mock.assert_not_called()
@pytest.mark.asyncio
async def test_mcp_authorize_proxies_to_discoverable_endpoint(self):