scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-07 08:26:10 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

fix(mcp): keep bridge server-id match total on non-ascii and strip bearer with any whitespace

Browse source
This commit is contained in:
Tin Chi Lo 2026-07-10 15:30:03 -07:00
parent 9d87527671
commit 0053527ccf
2 changed files with 27 additions and 7 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

14
litellm/proxy/_experimental/mcp_server/outbound_credentials/bridge_credentials.py
View file

@ -10,7 +10,6 @@ token-endpoint and admission wiring live in their respective call sites.
"""
import hashlib
import hmac
from datetime import datetime
from functools import lru_cache
from typing import Literal, TypeAlias
@ -31,7 +30,6 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.envelope import
_SIGNING_KEY_DOMAIN = b"litellm-mcp-bridge:envelope-signing:"
_ENCRYPTION_KEY_DOMAIN = b"litellm-mcp-bridge:envelope-encryption:"
_BEARER_PREFIX = "bearer "
# scrypt work factors (RFC 7914). n=2**15 with r=8/p=1 costs ~50ms and ~32MB per derivation, which
# makes offline guessing of a candidate master key memory-hard rather than a bare hash comparison.
@ -123,8 +121,9 @@ BridgeEnvelopeResult: TypeAlias = NotBridgeEnvelope | BridgeEnvelopeAdmitted | B
def _strip_bearer(value: str) -> str:
if value[: len(_BEARER_PREFIX)].lower() == _BEARER_PREFIX:
return value[len(_BEARER_PREFIX) :]
parts = value.split(None, 1)
if len(parts) == 2 and parts[0].lower() == "bearer":
return parts[1]
return value
@ -153,8 +152,9 @@ def resolve_bridge_envelope(
opened envelope whose sealed ``server_id`` does not match is rejected as
``BridgeEnvelopeInvalid``. Binding here (rather than leaving it to the caller) prevents
replaying an envelope minted for one server against another, which would forward the
first server's upstream credential across a server boundary. The comparison is constant
time so a mismatch does not leak the expected id through timing.
first server's upstream credential across a server boundary. ``server_id`` is not a
secret (the caller targets that server), so a plain equality check is sufficient and,
unlike ``hmac.compare_digest`` on ``str``, does not raise on a non-ASCII server_id.
"""
candidate = _strip_bearer(authorization_value)
if not is_envelope(candidate):
@ -162,7 +162,7 @@ def resolve_bridge_envelope(
opened = open_envelope(candidate, keys, now)
if not isinstance(opened, OpenedEnvelope):
return BridgeEnvelopeInvalid()
if not hmac.compare_digest(opened.identity.server_id, expected_server_id):
if opened.identity.server_id != expected_server_id:
return BridgeEnvelopeInvalid()
grant = opened.grant
upstream_authorization = f"{grant.token_type} {grant.access_token.get_secret_value()}"

20
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_bridge_credentials.py
View file

@ -150,6 +150,26 @@ def test_resolve_matching_server_binding_is_admitted():
assert isinstance(result, BridgeEnvelopeAdmitted)
def test_resolve_non_ascii_server_id_stays_total_and_does_not_raise():
"""The server-binding check must not raise on a non-ASCII server_id (an admin can register a
unicode server_id); it stays total and returns a typed result. A matching non-ASCII id admits,
a mismatching one is BridgeEnvelopeInvalid, and neither raises."""
keys = envelope_keys_from_master_key(_MASTER_KEY)
unicode_identity = EnvelopeIdentity(user_id=_IDENTITY.user_id, server_id="srv-café")
token = _sealed_token(keys, identity=unicode_identity)
assert isinstance(resolve_bridge_envelope(token, keys, _NOW, "srv-café"), BridgeEnvelopeAdmitted)
assert isinstance(resolve_bridge_envelope(token, keys, _NOW, "srv-cafe"), BridgeEnvelopeInvalid)
def test_resolve_strips_bearer_with_extra_whitespace():
"""A Bearer scheme separated by extra spaces or a tab still yields the envelope, so a client
using non-minimal but legal whitespace is not misclassified as a non-envelope."""
keys = envelope_keys_from_master_key(_MASTER_KEY)
token = _sealed_token(keys)
for header in (f"Bearer {token}", f"Bearer\t{token}", f" Bearer {token}"):
assert isinstance(resolve_bridge_envelope(header, keys, _NOW, _SERVER_ID), BridgeEnvelopeAdmitted)
def test_admitted_result_repr_never_leaks_upstream_token():
keys = envelope_keys_from_master_key(_MASTER_KEY)
result = resolve_bridge_envelope(_sealed_token(keys), keys, _NOW, _SERVER_ID)