scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-08-28 05:25:59 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 36

Merge pull request #38398 from daniel-meismer-zocdoc/litellm_mcp_bearer_scheme_refresh

Browse source 
fix(mcp): canonicalize bearer scheme on bridge egress
This commit is contained in:
Mateo Wang 2026-08-27 09:41:33 -07:00 • committed by GitHub
commit 98d231c09b
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 28 additions and 1 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -239,5 +239,6 @@ def resolve_bridge_envelope(
if opened.identity.server_id != expected_server_id:
return BridgeEnvelopeInvalid()
grant: Final = opened.grant
upstream_authorization: Final = f"{grant.token_type} {grant.access_token.get_secret_value()}"
authorization_scheme: Final = "Bearer" if grant.token_type.lower() == "bearer" else grant.token_type
upstream_authorization: Final = f"{authorization_scheme} {grant.access_token.get_secret_value()}"
return BridgeEnvelopeAdmitted(identity=opened.identity, upstream_authorization=SecretStr(upstream_authorization))

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

@ -10,6 +10,7 @@ through the consumer; and no path leaks the upstream token in a repr.
from datetime import datetime, timedelta, timezone
import pytest
from pydantic import SecretStr
from litellm.proxy._experimental.mcp_server.outbound_credentials.bridge_credentials import (
@ -188,6 +189,31 @@ def test_resolve_strips_optional_bearer_scheme_before_detection():
assert prefixed.upstream_authorization.get_secret_value() == bare.upstream_authorization.get_secret_value()
@pytest.mark.parametrize("token_type", ("bearer", "BEARER", "beArEr"))
def test_resolve_canonicalizes_case_insensitive_bearer_token_type(token_type: str):
keys = envelope_keys_from_master_key(_MASTER_KEY)
grant = UpstreamTokenGrant(access_token=SecretStr(_ACCESS_TOKEN), token_type=token_type, expires_in=600)
sealed = mint_envelope(_IDENTITY, grant, keys, _NOW)
assert isinstance(sealed, SealedEnvelope)
result = resolve_bridge_envelope(sealed.token.get_secret_value(), keys, _NOW, _SERVER_ID)
assert isinstance(result, BridgeEnvelopeAdmitted)
assert result.upstream_authorization.get_secret_value() == f"Bearer {_ACCESS_TOKEN}"
def test_resolve_preserves_non_bearer_token_type():
keys = envelope_keys_from_master_key(_MASTER_KEY)
grant = UpstreamTokenGrant(access_token=SecretStr(_ACCESS_TOKEN), token_type="DPoP", expires_in=600)
sealed = mint_envelope(_IDENTITY, grant, keys, _NOW)
assert isinstance(sealed, SealedEnvelope)
result = resolve_bridge_envelope(sealed.token.get_secret_value(), keys, _NOW, _SERVER_ID)
assert isinstance(result, BridgeEnvelopeAdmitted)
assert result.upstream_authorization.get_secret_value() == f"DPoP {_ACCESS_TOKEN}"
def test_resolve_expired_envelope_is_invalid_not_admitted():
keys = envelope_keys_from_master_key(_MASTER_KEY)
token = _sealed_token(keys, now=_NOW)