diff --git a/litellm/proxy/_experimental/mcp_server/outbound_credentials/bridge_credentials.py b/litellm/proxy/_experimental/mcp_server/outbound_credentials/bridge_credentials.py index c352f3a683e..08115ad3c56 100644 --- a/litellm/proxy/_experimental/mcp_server/outbound_credentials/bridge_credentials.py +++ b/litellm/proxy/_experimental/mcp_server/outbound_credentials/bridge_credentials.py @@ -167,7 +167,8 @@ class NotBridgeEnvelope(BaseModel): class BridgeEnvelopeAdmitted(BaseModel): """A valid envelope: the identity to admit under and the full upstream ``Authorization`` - value (``token_type access_token``) to forward to the upstream MCP server.""" + value (``token_type access_token``, with a case-insensitive bearer type canonicalized to + ``Bearer``) to forward to the upstream MCP server.""" model_config = ConfigDict(frozen=True) tag: Literal["admitted"] = "admitted" @@ -193,6 +194,16 @@ def _strip_bearer(value: str) -> str: return value +def _authorization_scheme(token_type: str) -> str: + """Canonicalize the upstream OAuth token type into the scheme to put on the wire. + + RFC 6749 defines ``token_type`` as case insensitive, so an upstream may report ``bearer``, + but RFC 6750 spells the HTTP auth scheme ``Bearer`` and strict resource servers reject any + other casing with 401 ``invalid_token``. Any non-bearer type is forwarded verbatim. + """ + return "Bearer" if token_type.lower() == "bearer" else token_type + + def is_bridge_envelope_shaped(authorization_value: str) -> bool: """Cheap, keyless test that an ``Authorization`` value carries an envelope of either kind (optional ``Bearer`` scheme stripped). The admission edge engages the bridge arm for an access envelope (to @@ -239,5 +250,5 @@ def resolve_bridge_envelope( 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()}" + upstream_authorization = f"{_authorization_scheme(grant.token_type)} {grant.access_token.get_secret_value()}" return BridgeEnvelopeAdmitted(identity=opened.identity, upstream_authorization=SecretStr(upstream_authorization)) diff --git a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_bridge_credentials.py b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_bridge_credentials.py index 753a3d6a942..a63ff044d70 100644 --- a/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_bridge_credentials.py +++ b/tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_bridge_credentials.py @@ -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,29 @@ 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("upstream_token_type", ["bearer", "BEARER", "Bearer", "beArEr"]) +def test_resolve_canonicalizes_case_insensitive_bearer_token_type_on_egress(upstream_token_type: str): + keys = envelope_keys_from_master_key(_MASTER_KEY) + grant = UpstreamTokenGrant( + access_token=SecretStr(_ACCESS_TOKEN), token_type=upstream_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_forwards_non_bearer_token_type_verbatim(): + 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)