scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-14 23:21:35 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 102

fix(mcp): prevent delegated admission key forwarding

Browse source
This commit is contained in:
Joshua Valluru 2026-09-12 17:00:21 -07:00
parent e2d888bb91
commit 97b964dd9a
3 changed files with 106 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

9
litellm/proxy/_experimental/mcp_server/mcp_server_manager.py
View file

@ -1055,7 +1055,7 @@ def _should_strip_caller_authorization(
pass-through cold-start case (RFC 9728) the bearer in
``Authorization`` is the upstream OAuth token and must be
forwarded, so we keep it.
- **oauth_delegate servers**: admission always runs and there is no
- **Delegated OAuth servers**: admission always runs and there is no
anonymous path, so the caller's separate ``Authorization`` is
forwarded only when a distinct ``x-litellm-api-key`` carried
admission. Without that header the ``Authorization`` *was* the
@ -1075,11 +1075,14 @@ def _should_strip_caller_authorization(
# upstream — it would override another user's stored credential. Delegate and
# pass-through return None from to_server_spec and keep forwarding the bearer.
return True
if not (mcp_server.is_oauth_passthrough or mcp_server.is_oauth_delegate):
is_delegated_oauth: Final = mcp_server.is_oauth_delegate or (
mcp_server.auth_type == MCPAuth.oauth2 and mcp_server.delegate_auth_to_upstream
)
if not (mcp_server.is_oauth_passthrough or is_delegated_oauth):
return False
has_explicit_litellm_admission_header: Final = _has_explicit_litellm_admission_header(raw_headers)
if mcp_server.is_oauth_delegate:
if is_delegated_oauth:
return not has_explicit_litellm_admission_header
return _authorization_is_litellm_admission_credential(raw_headers, user_api_key_auth) or (
user_api_key_auth is None and not has_explicit_litellm_admission_header

57
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server.py
View file

@ -521,6 +521,63 @@ def test_prepare_mcp_server_headers_oauth2_interactive_drops_caller_authorizatio
assert extra_headers is None
def test_prepare_mcp_server_headers_legacy_delegate_strips_admission_authorization():
from litellm.proxy._experimental.mcp_server.server import (
_prepare_mcp_server_headers,
)
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="legacy-delegate-admission",
name="legacy-delegate",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
delegate_auth_to_upstream=True,
)
server_auth_header, extra_headers = _prepare_mcp_server_headers(
server=server,
mcp_server_auth_headers=None,
mcp_auth_header=None,
oauth2_headers={"Authorization": "Bearer sk-litellm-key"},
raw_headers={"authorization": "Bearer sk-litellm-key"},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
assert server_auth_header is None
assert extra_headers is None
def test_prepare_mcp_server_headers_legacy_delegate_preserves_separate_upstream_authorization():
from litellm.proxy._experimental.mcp_server.server import (
_prepare_mcp_server_headers,
)
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="legacy-delegate-dual-credential",
name="legacy-delegate",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
delegate_auth_to_upstream=True,
)
server_auth_header, extra_headers = _prepare_mcp_server_headers(
server=server,
mcp_server_auth_headers=None,
mcp_auth_header=None,
oauth2_headers={"Authorization": "Bearer upstream-token"},
raw_headers={
"x-litellm-api-key": "Bearer sk-litellm-key",
"authorization": "Bearer upstream-token",
},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
assert server_auth_header is None
assert extra_headers == {"Authorization": "Bearer upstream-token"}
def test_prepare_mcp_server_headers_m2m_skips_authorization_from_raw_extra_headers():
"""M2M must not merge caller Authorization from raw_headers when extra_headers lists it."""
try:

43
tests/test_litellm/proxy/_experimental/mcp_server/test_mcp_server_manager.py
View file

@ -3390,6 +3390,49 @@ class TestMCPServerManager:
)
assert not extra_headers or "authorization" not in {k.lower() for k in extra_headers}
@pytest.mark.asyncio
async def test_call_regular_mcp_tool_legacy_delegate_never_forwards_admission_key(self):
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="server-legacy-delegate-leak",
name="legacy-delegate",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
delegate_auth_to_upstream=True,
)
extra_headers = await self._capture_call_extra_headers(
server,
oauth2_headers={"Authorization": "Bearer sk-litellm-key"},
raw_headers={"authorization": "Bearer sk-litellm-key"},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
assert not extra_headers or "authorization" not in {k.lower() for k in extra_headers}
@pytest.mark.asyncio
async def test_call_regular_mcp_tool_legacy_delegate_forwards_separate_authorization(self):
from litellm.proxy._types import UserAPIKeyAuth
server = MCPServer(
server_id="server-legacy-delegate-dual-credential",
name="legacy-delegate",
url="https://example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
delegate_auth_to_upstream=True,
)
extra_headers = await self._capture_call_extra_headers(
server,
oauth2_headers={"Authorization": "Bearer upstream-token"},
raw_headers={
"x-litellm-api-key": "Bearer sk-litellm-key",
"authorization": "Bearer upstream-token",
},
user_api_key_auth=UserAPIKeyAuth(api_key="sk-litellm-key"),
)
assert extra_headers == {"Authorization": "Bearer upstream-token"}
def test_should_strip_caller_authorization_new_modes(self):
from litellm.proxy._types import UserAPIKeyAuth