scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-22 00:31:44 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 96

fix(mcp): validate rendered static credential payloads

Browse source
This commit is contained in:
Joshua Valluru 2026-09-15 20:32:25 -07:00
parent be506936bd
commit 258176de76
2 changed files with 70 additions and 3 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

8
litellm/proxy/_experimental/mcp_server/upstream.py
View file

@ -4,7 +4,7 @@ import base64
from collections.abc import Mapping
from typing import Final
from litellm.experimental_mcp_client.client import MCPClient
from litellm.experimental_mcp_client.client import MCPClient, strip_auth_scheme
from litellm.proxy._experimental.mcp_server.outbound_credentials.adapter import raise_public
from litellm.proxy._experimental.mcp_server.outbound_credentials.result import Error, Ok, Result
from litellm.proxy._experimental.mcp_server.outbound_credentials.types import CredError
@ -24,13 +24,17 @@ def _usable_credential_value(auth_type: MCPAuthType, name: str, value: str) -> b
return True
if value.lower() in ("bearer", "basic", "token", "apikey"):
return False
if auth_type in (MCPAuth.bearer_token, MCPAuth.token):
scheme: Final = "Bearer" if auth_type == MCPAuth.bearer_token else "token"
credential: Final = strip_auth_scheme(value, scheme).strip()
return bool(credential) and credential.lower() != scheme.lower()
if auth_type == MCPAuth.basic:
parts: Final = value.split(None, 1)
if len(parts) != 2 or parts[0].lower() != "basic":
return False
try:
decoded: Final = base64.b64decode(parts[1], validate=True).strip()
return bool(decoded) and decoded.lower() != b"basic"
return b":" in decoded
except ValueError:
return False
return True

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

@ -13639,7 +13639,7 @@ class TestProtectedCredentialPreparation:
assert exc.value.status_code == 500
@pytest.mark.asyncio
@pytest.mark.parametrize("header", ["Basic", "Basic @@@", "Other abc", "Basic QmFzaWM="])
@pytest.mark.parametrize("header", ["Basic", "Basic @@@", "Other abc", "Basic QmFzaWM=", "Basic bm8tY29sb24="])
async def test_basic_headers_without_usable_credentials_reject(self, header: str) -> None:
server = MCPServer(server_id="bad-basic", name="bad-basic", url="https://upstream.example/mcp",
transport=MCPTransport.http, auth_type=MCPAuth.basic)
@ -13724,3 +13724,66 @@ class TestProtectedCredentialPreparation:
with pytest.raises(HTTPException) as exc:
await MCPServerManager()._create_mcp_client(server, mcp_auth_header={"Authorization": value})
assert exc.value.status_code == 500
@pytest.mark.asyncio
@pytest.mark.parametrize("value", ["no-colon", "Basic bm8tY29sb24="])
@pytest.mark.parametrize("source", ["configured", "caller"])
async def test_basic_requires_a_username_password_separator(self, value: str, source: str) -> None:
server: Final = MCPServer(
server_id="basic-pair", name="basic-pair", url="https://upstream.example/mcp",
transport=MCPTransport.http, auth_type=MCPAuth.basic,
authentication_token=value if source == "configured" else None,
)
with pytest.raises(HTTPException) as exc:
await MCPServerManager()._create_mcp_client(server, mcp_auth_header=value if source == "caller" else None)
assert exc.value.status_code == 500
@pytest.mark.asyncio
@pytest.mark.parametrize("value", ["user:pass", "user:", ":pass", ":"])
async def test_basic_preserves_username_password_pairs(self, value: str) -> None:
import base64
server: Final = MCPServer(
server_id="basic-valid", name="basic-valid", url="https://upstream.example/mcp",
transport=MCPTransport.http, auth_type=MCPAuth.basic, authentication_token=value,
)
client: Final = await MCPServerManager()._create_mcp_client(server)
request: Final = await client.prepare_request_auth()
scheme, encoded = request.headers["Authorization"].split(" ", 1)
assert scheme == "Basic"
assert base64.b64decode(encoded) == value.encode()
@pytest.mark.asyncio
@pytest.mark.parametrize("auth_type,value", [
(MCPAuth.bearer_token, "Bearer"), (MCPAuth.bearer_token, "Bearer "), (MCPAuth.bearer_token, "bearer"),
(MCPAuth.token, "token"), (MCPAuth.token, "token "), (MCPAuth.token, "TOKEN"),
])
@pytest.mark.parametrize("source", ["configured", "caller"])
async def test_static_scheme_only_input_cannot_hide_behind_rendered_prefix(
self, auth_type: MCPAuthType, value: str, source: str
) -> None:
server: Final = MCPServer(
server_id="empty-scheme", name="empty-scheme", url="https://upstream.example/mcp",
transport=MCPTransport.http, auth_type=auth_type,
authentication_token=value if source == "configured" else None,
)
with pytest.raises(HTTPException) as exc:
await MCPServerManager()._create_mcp_client(server, mcp_auth_header=value if source == "caller" else None)
assert exc.value.status_code == 500
@pytest.mark.asyncio
@pytest.mark.parametrize("auth_type,value,expected", [
(MCPAuth.bearer_token, "token", "Bearer token"),
(MCPAuth.bearer_token, "Bearertoken", "Bearer Bearertoken"),
(MCPAuth.token, "tokenish", "token tokenish"),
])
async def test_static_credentials_that_resemble_schemes_remain_usable(
self, auth_type: MCPAuthType, value: str, expected: str
) -> None:
server: Final = MCPServer(
server_id="real-token", name="real-token", url="https://upstream.example/mcp",
transport=MCPTransport.http, auth_type=auth_type, authentication_token=value,
)
client: Final = await MCPServerManager()._create_mcp_client(server)
request: Final = await client.prepare_request_auth()
assert request.headers["Authorization"] == expected