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

Merge pull request #38724 from BerriAI/litellm_mcp_oauth_identity_binding

Browse source 
fix(mcp): bind per-user OAuth credentials to the authenticated LiteLLM caller
This commit is contained in:
joshua-berri 2026-09-10 14:44:11 -07:00 • committed by GitHub
commit 4d067b56af
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
20 changed files with 1948 additions and 39 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
AGENTS.md
View file

@ -1 +1,3 @@
Read @CLAUDE.md for coding guidelines
Before requesting maintainer review, verify the current PR tip passes required CI and code coverage, meets Greptile confidence of at least 4/5, and has acceptable Veria and Bugbot reviews. Inspect warnings and findings, fix actionable issues, and rerun the affected checks and reviewers after changes. Record evidence for any false positive or unavailable review; never treat a pending or missing bot result as a pass. Do not lower coverage thresholds or lint budgets to satisfy a check

50
litellm/proxy/_experimental/mcp_server/db.py
View file

@ -6,10 +6,18 @@ from collections.abc import Awaitable, Callable, Iterable, Mapping, Sequence
from datetime import datetime, timedelta, timezone
from typing import TYPE_CHECKING, Any, Final, Literal, Protocol, TypedDict, cast
from fastapi import HTTPException
from typing_extensions import ReadOnly
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.constants import MCP_PER_USER_TOKEN_EXPIRY_BUFFER_SECONDS
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import (
RefreshTokenPresented,
credential_binding_matches,
enforce_oauth_identity_binding,
)
from litellm.proxy._experimental.mcp_server.oauth_utils import build_upstream_oauth2_token_request
from litellm.proxy._types import (
LiteLLM_MCPServerTable,
@ -117,6 +125,7 @@ class _OAuthCredentialAccessToken(TypedDict):
class OAuthCredentialPayload(_OAuthCredentialAccessToken, total=False):
identity_binding_proof: ReadOnly[str]
type: str
refresh_token: str
expires_at: str
@ -1393,6 +1402,7 @@ async def store_user_oauth_credential(
expires_in: int | None = None,
scopes: list[str] | None = None,
skip_byok_guard: bool = False,
identity_binding_proof: str | None = None,
) -> None:
"""Persist an OAuth2 access token for a user+server pair.
@ -1409,6 +1419,7 @@ async def store_user_oauth_credential(
"type": "oauth2",
"access_token": access_token,
"connected_at": datetime.now(timezone.utc).isoformat(),
**({"identity_binding_proof": identity_binding_proof} if identity_binding_proof else {}),
}
if refresh_token:
payload["refresh_token"] = refresh_token
@ -1628,6 +1639,11 @@ async def refresh_user_oauth_token(
warning and returns ``None`` — the caller is responsible for clearing the
stale credential and triggering re-authentication.
"""
binding: Final = server.oauth_identity_binding
if binding is not None and binding.mode == "enforce":
if not await credential_binding_matches(binding, user_id, server.server_id, cred):
return None
refresh_token: Final[str | None] = cred.get("refresh_token")
token_url: Final[str | None] = getattr(server, "effective_token_url", None) or getattr(server, "token_url", None)
server_id: Final[str] = getattr(server, "server_id", "")
@ -1677,6 +1693,19 @@ async def refresh_user_oauth_token(
)
return None
try:
binding_proof: Final = await enforce_oauth_identity_binding(
server=server,
token_response=body,
litellm_user_id=user_id,
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented(refresh_token),
)
except HTTPException as exc:
if exc.status_code != 403:
raise
return None
access_token: Final[str | None] = body.get("access_token")
if not access_token:
verbose_proxy_logger.warning(
@ -1709,6 +1738,7 @@ async def refresh_user_oauth_token(
refresh_token=new_refresh_token,
expires_in=expires_in,
scopes=scopes,
identity_binding_proof=binding_proof,
skip_byok_guard=True, # Row is already OAuth2; skip the extra find_unique check
)
@ -1742,6 +1772,10 @@ async def resolve_valid_user_oauth_token(
grant: Final = oauth_grant_state(cred)
if cred is None or grant == "absent":
return None
binding: Final = server.oauth_identity_binding
if binding is not None and binding.mode == "enforce":
if not await credential_binding_matches(binding, user_id, server.server_id, cred):
return None
if grant == "valid":
return cred
if prisma_client is None:
@ -1782,7 +1816,17 @@ async def resolve_user_oauth_access_token(
mcp_per_user_token_cache,
)
if prefetched_creds is None:
binding: Final = server.oauth_identity_binding
enforce_binding: Final = binding is not None and binding.mode == "enforce"
if prefetched_creds is None and enforce_binding and binding is not None:
bound_token: Final = await mcp_per_user_token_cache.get_token(user_id, server_id)
if bound_token is not None:
if await credential_binding_matches(
binding, user_id, server_id, {"identity_binding_proof": bound_token.identity_binding_proof}
):
return bound_token.access_token
await mcp_per_user_token_cache.delete(user_id, server_id)
if prefetched_creds is None and not enforce_binding:
cached_token: Final = await mcp_per_user_token_cache.get(user_id, server_id)
if cached_token is not None:
return cached_token
@ -1816,7 +1860,9 @@ async def resolve_user_oauth_access_token(
access_token: Final[str] = cred["access_token"]
if prefetched_creds is None:
ttl: Final = _compute_per_user_token_ttl(server, _remaining_token_seconds(cred.get("expires_at")))
await mcp_per_user_token_cache.set(user_id, server_id, access_token, ttl)
await mcp_per_user_token_cache.set(
user_id, server_id, access_token, ttl, identity_binding_proof=cred.get("identity_binding_proof")
)
return access_token
except Exception as e:
verbose_proxy_logger.warning(

112
litellm/proxy/_experimental/mcp_server/discoverable_endpoints.py
View file

@ -57,6 +57,11 @@ from litellm.proxy._experimental.mcp_server.gateway_dcr_flow import (
relative_request_url,
revoke_refresh_token,
)
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import (
RefreshOwnershipProven,
RefreshTokenPresented,
enforce_oauth_identity_binding,
)
from litellm.proxy._experimental.mcp_server.oauth_utils import (
TOKEN_NO_CACHE_HEADERS,
build_upstream_oauth2_token_request,
@ -139,6 +144,7 @@ def encode_state_with_base_url(
dcr_client_id: str | None = None,
dcr_client_secret: str | None = None,
dcr_token_endpoint_auth_method: MCPTokenEndpointAuthMethod | None = None,
oauth_nonce: str | None = None,
) -> str:
"""
Encode the base_url, original state, and PKCE parameters using encryption.
@ -149,9 +155,8 @@ def encode_state_with_base_url(
code_challenge: PKCE code challenge from client
code_challenge_method: PKCE code challenge method from client
client_redirect_uri: Original redirect_uri from client
litellm_user_id: The SSO-authenticated litellm user captured at the bridge authorize
(interactive dcr_bridge oauth_delegate only); the callback seals it into the gateway
authorization code so the token mint can bind the envelope to this user
litellm_user_id: The authenticated user captured for bridge or identity-bound per-user OAuth;
the callback seals this credential owner into the authorization code
mcp_server_id: The server the flow targets, sealed alongside litellm_user_id (bridge) or
dcr_client_id (ephemeral mint) so the gateway code cannot be replayed against another
server
@ -169,6 +174,7 @@ def encode_state_with_base_url(
An encrypted string that encodes all values
"""
state_data: Final = {
"oauth_nonce": oauth_nonce,
"base_url": base_url,
"original_state": original_state,
"code_challenge": code_challenge,
@ -210,10 +216,10 @@ _BRIDGE_AUTH_CODE_PREFIX: Final = "llm_bcode_"
class _BridgeAuthorizationCode(BaseModel):
"""The identity and upstream code the gateway seals into the authorization code it hands a DCR
client for an interactive dcr_bridge oauth_delegate sign-in, recovered at the token endpoint."""
"""Authenticated caller and upstream code sealed for bridge or identity-bound per-user OAuth."""
model_config = ConfigDict(frozen=True)
oauth_nonce: str | None = None
upstream_code: str = Field(min_length=1)
litellm_user_id: str = Field(min_length=1)
mcp_server_id: str = Field(min_length=1)
@ -225,7 +231,12 @@ def is_bridge_authorization_code(code: str) -> bool:
return code.startswith(_BRIDGE_AUTH_CODE_PREFIX)
def seal_bridge_authorization_code(upstream_code: str, litellm_user_id: str, mcp_server_id: str) -> str:
def seal_bridge_authorization_code(
upstream_code: str,
litellm_user_id: str,
mcp_server_id: str,
oauth_nonce: str | None = None,
) -> str:
"""Seal the upstream authorization code and the SSO-captured litellm user into a gateway
authorization code. The DCR client only echoes this opaque value back at the token endpoint; the
gateway decrypts it there to recover the user (to bind the envelope) and the upstream code (to
@ -234,7 +245,12 @@ def seal_bridge_authorization_code(upstream_code: str, litellm_user_id: str, mcp
authenticated symmetric helper (the same family the OAuth state uses), so the client can neither
read nor forge it."""
payload: Final = json.dumps(
{"upstream_code": upstream_code, "litellm_user_id": litellm_user_id, "mcp_server_id": mcp_server_id},
{
"upstream_code": upstream_code,
"litellm_user_id": litellm_user_id,
"mcp_server_id": mcp_server_id,
"oauth_nonce": oauth_nonce,
},
sort_keys=True,
)
return _BRIDGE_AUTH_CODE_PREFIX + encrypt_value_helper(payload)
@ -547,6 +563,7 @@ async def _store_per_user_token_server_side(
server: MCPServer,
user_id: str,
token_response: dict[str, Any],
identity_binding_proof: str | None = None,
) -> None:
"""Persist the OAuth token server-side and warm the Redis cache.
@ -588,6 +605,7 @@ async def _store_per_user_token_server_side(
refresh_token=refresh_token,
expires_in=expires_in,
scopes=scopes,
identity_binding_proof=identity_binding_proof,
)
verbose_logger.info(
"_store_per_user_token_server_side: stored token for user=%s server=%s",
@ -616,6 +634,7 @@ async def _store_per_user_token_server_side(
server_id=server.server_id,
access_token=access_token,
ttl=ttl,
identity_binding_proof=identity_binding_proof,
)
@ -854,6 +873,11 @@ async def authorize_with_server(
),
)
binding: Final = resolved_server.oauth_identity_binding
enforce_binding: Final = binding is not None and binding.mode == "enforce"
if enforce_binding:
_require_s256_pkce(code_challenge, code_challenge_method)
if resolved_server.is_dcr_bridge:
# Enforce S256 PKCE on both bridge arms. The relay arm forwards the validated,
# now-non-optional pair to the upstream authorize; the short-circuit arm keeps
@ -884,19 +908,16 @@ async def authorize_with_server(
base_url: Final = urlunparse(parsed._replace(query=""))
request_base_url: Final = get_request_base_url(request)
# Interactive dcr_bridge oauth_delegate sign-in: this arm runs the gateway /callback and /token in
# the loop, so the gateway can capture the litellm user here (from the browser's UI session) and
# carry it to the back-channel token mint. Seal the SSO user and the target server into the state;
# the callback reads them back to mint the gateway authorization code. A DCR client cannot present a
# litellm key, so the browser session is the only identity source; without one there is nothing to
# bind, so send the user through login first. Every other oauth2 server keeps the identity-less state.
# Seal the authenticated caller into state so the token exchange cannot select another credential owner.
litellm_user_id: str | None = None
if resolved_server.is_dcr_bridge and resolved_server.is_oauth_delegate:
if enforce_binding or (resolved_server.is_dcr_bridge and resolved_server.is_oauth_delegate):
from litellm.proxy._experimental.mcp_server.byok_oauth_endpoints import ( # noqa: PLC0415 # inline import avoids a module-load circular import
_user_id_from_session_cookie,
)
litellm_user_id = _user_id_from_session_cookie(request)
litellm_user_id = (
await _extract_user_id_from_request(request) if enforce_binding else None
) or _user_id_from_session_cookie(request)
if litellm_user_id is None:
return _redirect_to_litellm_login(request)
denial: Final = await _bridge_authorize_access_denial(
@ -908,9 +929,11 @@ async def authorize_with_server(
if denial is not None:
return denial
oauth_nonce: Final = secrets.token_urlsafe(32) if enforce_binding else None
encoded_state: Final = encode_state_with_base_url(
base_url=base_url,
original_state=state,
oauth_nonce=oauth_nonce,
code_challenge=code_challenge,
code_challenge_method=code_challenge_method,
client_redirect_uri=redirect_uri,
@ -930,11 +953,16 @@ async def authorize_with_server(
"state": relay_state,
"response_type": response_type or "code",
}
if oauth_nonce:
params["nonce"] = oauth_nonce
if scope:
params["scope"] = scope
elif resolved_server.scopes:
params["scope"] = " ".join(resolved_server.scopes)
if enforce_binding and "openid" not in params.get("scope", "").split():
params["scope"] = f"openid {params.get('scope', '')}".strip()
if code_challenge:
params["code_challenge"] = code_challenge
if code_challenge_method:
@ -1015,6 +1043,12 @@ async def exchange_token_with_server(
except TokenEndpointAuthConfigError as exc:
raise HTTPException(status_code=400, detail=str(exc)) from exc
request_user_id: Final = (
await _extract_user_id_from_request(request)
if resolved_server.needs_user_oauth_token or resolved_server.oauth_identity_binding is not None
else None
)
bridge_identity: _BridgeAuthorizationCode | None = None
bridge_mint_ready: _BridgeMintReady | None = None
bridge_upstream_refresh: SecretStr | None = None
@ -1051,7 +1085,13 @@ async def exchange_token_with_server(
refresh_request_scope = scope or bridge_upstream_scope
if refresh_request_scope:
token_data["scope"] = refresh_request_scope
refresh_ownership = ( # rebind-ok: grant-specific branches assign one ownership value
RefreshOwnershipProven()
if bridge_upstream_refresh is not None
else RefreshTokenPresented(upstream_refresh_token)
)
else:
refresh_ownership = None # rebind-ok: grant-specific branches assign one ownership value
if not code:
raise HTTPException(
status_code=400,
@ -1070,6 +1110,14 @@ async def exchange_token_with_server(
detail="Authorization code was issued for a different MCP server",
)
code = bridge_identity.upstream_code
binding: Final = resolved_server.oauth_identity_binding
if binding is not None and binding.mode == "enforce":
if bridge_identity is None or not bridge_identity.oauth_nonce:
raise HTTPException(status_code=403, detail={"error": "oauth_identity_binding_failed"})
if request_user_id is not None and request_user_id != bridge_identity.litellm_user_id:
raise HTTPException(status_code=403, detail={"error": "oauth_principal_mismatch"})
if not code_verifier:
raise HTTPException(status_code=403, detail={"error": "oauth_identity_binding_failed"})
bridge_token_relay: Final = _dcr_bridge_relays_client_registration(resolved_server)
if bridge_token_relay and not redirect_uri:
raise HTTPException(
@ -1097,6 +1145,16 @@ async def exchange_token_with_server(
return _bridge_mint_error_response(prepared)
bridge_mint_ready = prepared
refresh_binding: Final = resolved_server.oauth_identity_binding
if grant_type == "refresh_token" and refresh_binding is not None and refresh_binding.mode == "enforce":
await enforce_oauth_identity_binding(
server=resolved_server,
token_response={},
litellm_user_id=request_user_id,
grant_type=grant_type,
refresh_ownership=refresh_ownership,
)
async_client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
try:
response: Final = await async_client.post(
@ -1137,17 +1195,34 @@ async def exchange_token_with_server(
server_id=resolved_server.server_id,
)
# Bind the exchanged token to the LiteLLM caller BEFORE it is returned, stored, or cached, so a
# token minted for a different upstream principal never becomes usable under the caller's user_id.
resolved_user_id: Final = bridge_identity.litellm_user_id if bridge_identity else request_user_id
binding_proof: Final = (
await enforce_oauth_identity_binding(
server=resolved_server,
token_response=token_response,
litellm_user_id=resolved_user_id,
grant_type=grant_type,
refresh_ownership=refresh_ownership,
expected_nonce=bridge_identity.oauth_nonce if bridge_identity else None,
)
if isinstance(token_response, dict)
else None
)
# Store server-side when the server is configured for per-user OAuth and
# the calling client has provided a valid LiteLLM identity.
# Errors are non-fatal: the token is still returned to the client.
if resolved_server.needs_user_oauth_token:
user_id: Final = await _extract_user_id_from_request(request)
user_id: Final = resolved_user_id
if user_id:
try:
await _store_per_user_token_server_side(
server=resolved_server,
user_id=user_id,
token_response=token_response,
identity_binding_proof=binding_proof,
)
except Exception as exc:
verbose_logger.warning(
@ -2134,7 +2209,10 @@ async def callback(
forwarded_code = code
if isinstance(litellm_user_id, str) and litellm_user_id and isinstance(mcp_server_id, str) and mcp_server_id:
forwarded_code = seal_bridge_authorization_code(
upstream_code=code, litellm_user_id=litellm_user_id, mcp_server_id=mcp_server_id
upstream_code=code,
litellm_user_id=litellm_user_id,
mcp_server_id=mcp_server_id,
oauth_nonce=state_data.get("oauth_nonce"),
)
elif isinstance(dcr_client_id, str) and dcr_client_id and isinstance(mcp_server_id, str) and mcp_server_id:
forwarded_code = seal_passthrough_authorization_code(

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

@ -2444,6 +2444,8 @@ class MCPServerManager:
allow_elicitation=bool(server_config.get("allow_elicitation", False)),
timeout=server_config.get("timeout", None),
max_concurrent_requests=server_config.get("max_concurrent_requests", None),
token_validation=server_config.get("token_validation", None),
oauth_identity_binding=server_config.get("oauth_identity_binding", None),
)
self._assign_unique_short_prefix(new_server)
_warn_internal_delegate_pkce_if_applicable(new_server, source="config")

25
litellm/proxy/_experimental/mcp_server/oauth2_token_cache.py
View file

@ -28,6 +28,8 @@ from litellm.proxy._experimental.mcp_server.oauth_utils import (
build_upstream_oauth2_token_request,
resolve_upstream_resource,
)
from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import OAuthToken
from litellm.proxy._experimental.mcp_server.outbound_credentials.token_cache_codec import OAuthTokenCacheCodec
from litellm.proxy.common_utils.encrypt_decrypt_utils import (
decrypt_value_helper,
encrypt_value_helper,
@ -233,8 +235,17 @@ class MCPPerUserTokenCache:
def _cache_key(self, user_id: str, server_id: str) -> str:
return f"{MCP_PER_USER_TOKEN_REDIS_KEY_PREFIX}:{user_id}:{server_id}"
def _codec(self) -> OAuthTokenCacheCodec:
return OAuthTokenCacheCodec(
encrypt_value_helper,
lambda blob: decrypt_value_helper(blob, key="mcp_per_user_token", exception_type="debug"),
)
async def get(self, user_id: str, server_id: str) -> str | None:
"""Return the plaintext access_token, or None on miss/error."""
token: Final = await self.get_token(user_id, server_id)
return token.access_token if token is not None else None
async def get_token(self, user_id: str, server_id: str) -> OAuthToken | None:
try:
from litellm.proxy.proxy_server import user_api_key_cache # noqa: PLC0415
@ -242,12 +253,7 @@ class MCPPerUserTokenCache:
encrypted: Final = await user_api_key_cache.async_get_cache(key)
if encrypted is None:
return None
plaintext: Final = decrypt_value_helper(
encrypted,
key="mcp_per_user_token",
exception_type="debug",
)
return plaintext or None
return self._codec().decode(encrypted)
except Exception as exc:
verbose_logger.debug(
"MCPPerUserTokenCache.get failed for user=%s server=%s: %s",
@ -263,13 +269,16 @@ class MCPPerUserTokenCache:
server_id: str,
access_token: str,
ttl: int,
identity_binding_proof: str | None = None,
) -> None:
"""Store NaCl-encrypted access_token in Redis with the given TTL."""
try:
from litellm.proxy.proxy_server import user_api_key_cache # noqa: PLC0415
key: Final = self._cache_key(user_id, server_id)
encrypted: Final = encrypt_value_helper(access_token)
encrypted: Final = self._codec().encode(
OAuthToken(access_token=access_token, identity_binding_proof=identity_binding_proof)
)
await user_api_key_cache.async_set_cache(key, encrypted, ttl=ttl)
verbose_logger.debug(
"MCPPerUserTokenCache.set: cached token for user=%s server=%s ttl=%ds",

423
litellm/proxy/_experimental/mcp_server/oauth_identity_binding.py Normal file
View file

@ -0,0 +1,423 @@
"""Per-user OAuth identity binding: verify the upstream OIDC principal matches the LiteLLM caller.
Closes the confused-deputy gap where a browser authenticated upstream as one principal produces a
token that the relay stores under a different, LiteLLM-authenticated principal: before the token
endpoint returns, stores, or caches an exchanged token for an identity-bound server, the id_token
is validated (signature via the pinned issuer's JWKS, issuer, audience, expiry) and its principal
claim is compared to the caller's trusted LiteLLM identity. Mismatches fail closed in enforce mode
and are logged in audit mode.
"""
import hashlib
import hmac
import json
from collections.abc import Awaitable, Callable, Mapping, Sequence
from dataclasses import dataclass
from typing import Final, Literal, Protocol, TypeAlias
import jwt
from fastapi import HTTPException
from jwt.types import Options
from typing_extensions import assert_never
from litellm._logging import verbose_logger
from litellm.caching.in_memory_cache import InMemoryCache
from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
from litellm.types.llms.custom_http import httpxSpecialProvider
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
_ALLOWED_ID_TOKEN_ALGORITHMS: Final = (
"RS256",
"RS384",
"RS512",
"ES256",
"ES384",
"ES512",
"PS256",
"PS384",
"PS512",
)
_JWKS_CACHE_TTL_SECONDS: Final = 3600
_jwks_cache: Final = InMemoryCache(default_ttl=_JWKS_CACHE_TTL_SECONDS)
JwksFetcher: TypeAlias = Callable[
[MCPOAuthIdentityBinding], # mutable-ok: Callable parameter syntax requires a list
Awaitable[Sequence[Mapping[str, object]]],
]
CallerPrincipalLoader: TypeAlias = Callable[
[str, MCPOAuthIdentityBinding], # mutable-ok: Callable parameter syntax requires a list
Awaitable[str | None],
]
@dataclass(frozen=True, slots=True)
class VerifiedRefreshToken:
refresh_token: str
binding_proof: str
StoredRefreshTokenLoader: TypeAlias = Callable[
[str, str, MCPOAuthIdentityBinding], # mutable-ok: Callable parameter syntax requires a list
Awaitable[VerifiedRefreshToken | None],
]
_RejectionCode: TypeAlias = Literal["oauth_principal_mismatch", "oauth_identity_binding_failed"]
@dataclass(frozen=True, slots=True)
class _BindingRejection:
code: _RejectionCode
description: str
@dataclass(frozen=True, slots=True)
class RefreshOwnershipProven:
"""The gateway itself unwrapped the upstream refresh token from a sealed per-user envelope."""
@dataclass(frozen=True, slots=True)
class RefreshTokenPresented:
refresh_token: str
RefreshOwnership: TypeAlias = RefreshOwnershipProven | RefreshTokenPresented | None
class BindingValidator(Protocol):
async def __call__(
self,
*,
server: MCPServer,
token_response: Mapping[str, object],
litellm_user_id: str | None,
grant_type: str,
refresh_ownership: RefreshOwnership,
) -> str | None: ...
async def _fetch_issuer_jwks(binding: MCPOAuthIdentityBinding) -> Sequence[Mapping[str, object]]:
jwks_url: Final[str] = binding.jwks_url or await _discover_jwks_url(binding.issuer)
cached: Final = await _jwks_cache.async_get_cache(jwks_url)
if isinstance(cached, list):
return cached
client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
response: Final = await client.get(jwks_url)
response.raise_for_status()
document: Final = response.json()
keys: Final = document.get("keys") if isinstance(document, dict) else None
if not isinstance(keys, list):
raise TypeError(f"JWKS document at {jwks_url} has no 'keys' array")
await _jwks_cache.async_set_cache(jwks_url, keys, ttl=_JWKS_CACHE_TTL_SECONDS)
return keys
async def _discover_jwks_url(issuer: str) -> str:
discovery_url: Final = f"{issuer.rstrip('/')}/.well-known/openid-configuration"
client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
response: Final = await client.get(discovery_url)
response.raise_for_status()
metadata: Final = response.json()
jwks_uri: Final = metadata.get("jwks_uri") if isinstance(metadata, dict) else None
if not isinstance(jwks_uri, str) or not jwks_uri:
raise ValueError(f"OIDC discovery at {discovery_url} returned no jwks_uri")
return jwks_uri
def _select_signing_key(id_token: str, keys: Sequence[Mapping[str, object]]) -> "jwt.PyJWK | _BindingRejection":
header: Final = jwt.get_unverified_header(id_token)
kid: Final = header.get("kid")
for key in keys:
if kid is None or key.get("kid") == kid:
return jwt.PyJWK(dict(key)) # mutable-ok: PyJWT requires a concrete JWK dictionary
return _BindingRejection(
code="oauth_identity_binding_failed",
description=f"id_token signing key (kid={kid!r}) not found in the issuer's JWKS",
)
def _decode_id_token(
id_token: str,
binding: MCPOAuthIdentityBinding,
signing_key: "jwt.PyJWK",
) -> "Mapping[str, object] | _BindingRejection":
try:
decode_options: Final[Options] = {"require": ("iss", "exp", "aud", "sub", "iat")}
return jwt.decode(
id_token,
signing_key.key,
algorithms=_ALLOWED_ID_TOKEN_ALGORITHMS,
issuer=binding.issuer,
audience=binding.audiences,
options=decode_options,
)
except jwt.InvalidTokenError as exc:
return _BindingRejection(
code="oauth_identity_binding_failed",
description=f"id_token validation failed: {exc}",
)
def _upstream_principal(
claims: Mapping[str, object],
binding: MCPOAuthIdentityBinding,
) -> "str | _BindingRejection":
principal: Final = claims.get(binding.principal_claim)
if not isinstance(principal, str) or not principal:
return _BindingRejection(
code="oauth_identity_binding_failed",
description=f"id_token has no usable '{binding.principal_claim}' claim",
)
if (
binding.principal_claim == "email"
and binding.require_email_verified
and claims.get("email_verified") is not True
):
return _BindingRejection(
code="oauth_identity_binding_failed",
description="id_token email is not verified (email_verified is not true)",
)
return principal
async def _load_caller_principal(litellm_user_id: str, binding: MCPOAuthIdentityBinding) -> str | None:
if binding.caller_field == "user_id":
return litellm_user_id
from litellm.proxy._experimental.mcp_server.bridge_token_flow import ( # noqa: PLC0415 # inline import avoids a module-load circular import
load_active_user_by_id,
)
loaded: Final = await load_active_user_by_id(litellm_user_id)
if isinstance(loaded, str):
return None
return loaded.user_email
async def _load_stored_refresh_token(
litellm_user_id: str, server_id: str, binding: MCPOAuthIdentityBinding
) -> VerifiedRefreshToken | None:
try:
from litellm.proxy._experimental.mcp_server.db import ( # noqa: PLC0415 # keep database imports lazy
get_user_oauth_credential,
)
from litellm.proxy.utils import get_prisma_client_or_throw # noqa: PLC0415 # keep database imports lazy
prisma_client: Final = get_prisma_client_or_throw(
"Database not connected. Cannot verify OAuth refresh token ownership."
)
cred: Final = await get_user_oauth_credential(
prisma_client=prisma_client,
user_id=litellm_user_id,
server_id=server_id,
)
if not cred or not await credential_binding_matches(binding, litellm_user_id, server_id, cred):
return None
refresh_token: Final = cred.get("refresh_token")
proof: Final = cred.get("identity_binding_proof")
return VerifiedRefreshToken(refresh_token, proof) if refresh_token and proof else None
except Exception: # noqa: BLE001 # a credential lookup failure must fail closed
return None
async def current_binding_proof(
binding: MCPOAuthIdentityBinding,
user_id: str,
server_id: str,
caller_principal_loader: CallerPrincipalLoader = _load_caller_principal,
) -> str | None:
principal: Final = await caller_principal_loader(user_id, binding)
if not principal:
return None
return _binding_proof(binding, user_id, server_id, principal)
def _binding_proof(binding: MCPOAuthIdentityBinding, user_id: str, server_id: str, principal: str) -> str:
payload: Final = json.dumps(
("oidc-nonce-v1", server_id, user_id, principal, binding.model_dump(mode="json")),
sort_keys=True,
)
return hashlib.sha256(payload.encode()).hexdigest()
async def credential_binding_matches(
binding: MCPOAuthIdentityBinding,
user_id: str,
server_id: str,
credential: Mapping[str, object],
caller_principal_loader: CallerPrincipalLoader = _load_caller_principal,
) -> bool:
stored: Final = credential.get("identity_binding_proof")
if not isinstance(stored, str) or not stored:
return False
expected: Final = await current_binding_proof(binding, user_id, server_id, caller_principal_loader)
return expected is not None and hmac.compare_digest(stored, expected)
def _principals_match(upstream: str, caller: str, binding: MCPOAuthIdentityBinding) -> bool:
if binding.principal_claim == "email" or binding.caller_field == "user_email":
return upstream.strip().casefold() == caller.strip().casefold()
return upstream == caller
async def _evaluate_refresh_ownership(
binding: MCPOAuthIdentityBinding,
litellm_user_id: str | None,
server_id: str,
refresh_ownership: RefreshOwnership,
stored_refresh_token_loader: StoredRefreshTokenLoader,
) -> _BindingRejection | str:
match refresh_ownership:
case RefreshOwnershipProven():
return _BindingRejection(
code="oauth_identity_binding_failed",
description="an identity envelope alone does not prove upstream principal binding",
)
case None:
return _BindingRejection(
code="oauth_identity_binding_failed",
description="refresh_token grant without an id_token carries no refresh token to prove ownership of",
)
case RefreshTokenPresented(refresh_token):
if not litellm_user_id:
return _BindingRejection(
code="oauth_identity_binding_failed",
description="the request carries no resolvable LiteLLM user identity to bind the credential to",
)
stored: Final = await stored_refresh_token_loader(litellm_user_id, server_id, binding)
if stored is None or not hmac.compare_digest(stored.refresh_token, refresh_token):
return _BindingRejection(
code="oauth_identity_binding_failed",
description="the presented refresh_token is not the caller's stored credential for this server",
)
return stored.binding_proof
assert_never(refresh_ownership) # pragma: no cover
async def _evaluate_binding(
binding: MCPOAuthIdentityBinding,
token_response: Mapping[str, object],
litellm_user_id: str | None,
grant_type: str,
server_id: str,
refresh_ownership: RefreshOwnership,
jwks_fetcher: JwksFetcher,
caller_principal_loader: CallerPrincipalLoader,
stored_refresh_token_loader: StoredRefreshTokenLoader,
expected_nonce: str | None,
) -> _BindingRejection | str:
id_token: Final = token_response.get("id_token")
if not isinstance(id_token, str) or not id_token:
if grant_type != "refresh_token":
return _BindingRejection(
code="oauth_identity_binding_failed",
description="the upstream token response carries no id_token to bind the credential to a principal",
)
return await _evaluate_refresh_ownership(
binding,
litellm_user_id,
server_id,
refresh_ownership,
stored_refresh_token_loader,
)
if not litellm_user_id:
return _BindingRejection(
code="oauth_identity_binding_failed",
description="the request carries no resolvable LiteLLM user identity to bind the credential to",
)
try:
keys: Final = await jwks_fetcher(binding)
except Exception as exc: # noqa: BLE001 # a JWKS fetch failure must fail closed, not surface as a 500
return _BindingRejection(
code="oauth_identity_binding_failed",
description=f"could not fetch the issuer's JWKS: {exc}",
)
try:
signing_key: Final = _select_signing_key(id_token, keys)
except (jwt.PyJWTError, ValueError, TypeError):
return _BindingRejection(
code="oauth_identity_binding_failed",
description="invalid id_token header or issuer signing key",
)
if isinstance(signing_key, _BindingRejection):
return signing_key
claims: Final = _decode_id_token(id_token, binding, signing_key)
if isinstance(claims, _BindingRejection):
return claims
if grant_type == "authorization_code" and (binding.mode == "enforce" or expected_nonce is not None):
nonce: Final = claims.get("nonce")
if not expected_nonce or not isinstance(nonce, str) or not hmac.compare_digest(nonce, expected_nonce):
return _BindingRejection(
code="oauth_identity_binding_failed",
description="id_token nonce does not match the authenticated authorization transaction",
)
upstream: Final = _upstream_principal(claims, binding)
if isinstance(upstream, _BindingRejection):
return upstream
caller: Final = await caller_principal_loader(litellm_user_id, binding)
if not caller:
return _BindingRejection(
code="oauth_identity_binding_failed",
description=f"the LiteLLM user has no '{binding.caller_field}' to compare the upstream principal against",
)
if not _principals_match(upstream, caller, binding):
return _BindingRejection(
code="oauth_principal_mismatch",
description="The browser account does not match the selected credential owner.",
)
return _binding_proof(binding, litellm_user_id, server_id, caller)
async def enforce_oauth_identity_binding(
server: MCPServer,
token_response: Mapping[str, object],
litellm_user_id: str | None,
grant_type: str,
refresh_ownership: RefreshOwnership,
jwks_fetcher: JwksFetcher = _fetch_issuer_jwks,
caller_principal_loader: CallerPrincipalLoader = _load_caller_principal,
stored_refresh_token_loader: StoredRefreshTokenLoader = _load_stored_refresh_token,
expected_nonce: str | None = None,
) -> str | None:
"""Validate the exchanged token's upstream principal against the LiteLLM caller.
No-op when the server has no binding or it is disabled. In enforce mode a failure raises 403
before the caller returns, stores, or caches the token; in audit mode failures are logged only.
A refresh_token grant without an id_token is allowed only when the presented refresh token matches
the caller's stored credential and that credential was previously identity-validated.
"""
binding: Final = server.oauth_identity_binding
if binding is None or binding.mode not in ("audit", "enforce"):
return
rejection: Final = await _evaluate_binding(
binding=binding,
token_response=token_response,
litellm_user_id=litellm_user_id,
grant_type=grant_type,
server_id=server.server_id,
refresh_ownership=refresh_ownership,
jwks_fetcher=jwks_fetcher,
caller_principal_loader=caller_principal_loader,
stored_refresh_token_loader=stored_refresh_token_loader,
expected_nonce=expected_nonce,
)
if isinstance(rejection, str):
return rejection if binding.mode == "enforce" else None
if binding.mode == "audit":
verbose_logger.warning(
"oauth_identity_binding audit: server=%s user=%s grant=%s rejected=%s (%s)",
server.server_id,
litellm_user_id,
grant_type,
rejection.code,
rejection.description,
)
return
raise HTTPException(
status_code=403,
detail={
"error": rejection.code,
"error_description": rejection.description,
"server_id": server.server_id,
"credential_owner": "caller",
"credential_stored": False,
},
)

48
litellm/proxy/_experimental/mcp_server/outbound_credentials/authz_code_refresher.py
View file

@ -14,10 +14,17 @@ import time
from collections.abc import Awaitable, Callable
from typing import TYPE_CHECKING, Final, Protocol
from fastapi import HTTPException
from litellm._logging import verbose_logger
from litellm.proxy._experimental.mcp_server.auth.token_endpoint_auth import (
TokenEndpointAuthConfigError,
)
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import (
BindingValidator,
RefreshTokenPresented,
enforce_oauth_identity_binding,
)
from litellm.proxy._experimental.mcp_server.oauth_utils import build_upstream_oauth2_token_request
from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
OAuthToken,
@ -39,6 +46,7 @@ class CredentialPersist(Protocol):
refresh_token: str | None,
expires_in: int | None,
scopes: tuple[str, ...] | None,
identity_binding_proof: str | None = None,
) -> None: ...
@ -78,13 +86,23 @@ class AuthorizationCodeRefresher:
persist: CredentialPersist,
*,
clock: Callable[[], float] = time.time,
identity_validator: BindingValidator = enforce_oauth_identity_binding,
) -> None:
self._server_lookup = server_lookup
self._token_endpoint = token_endpoint
self._persist = persist
self._clock = clock
self._identity_validator = identity_validator
async def refresh(self, user_id: str, server_id: str, token: OAuthToken) -> OAuthToken | None:
try:
return await self._refresh(user_id, server_id, token)
except HTTPException as exc:
if exc.status_code != 403:
raise
return None
async def _refresh(self, user_id: str, server_id: str, token: OAuthToken) -> OAuthToken | None:
if token.refresh_token is None:
return None
server: Final = self._server_lookup(server_id)
@ -104,6 +122,15 @@ class AuthorizationCodeRefresher:
except TokenEndpointAuthConfigError as exc:
verbose_logger.warning("MCP OAuth refresh misconfigured for server %s: %s", server_id, exc)
return None
binding: Final = server.oauth_identity_binding
if binding is not None and binding.mode == "enforce":
await self._identity_validator(
server=server,
token_response={},
litellm_user_id=user_id,
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented(token.refresh_token),
)
form: Final = {
"grant_type": "refresh_token",
"refresh_token": token.refresh_token,
@ -116,15 +143,34 @@ class AuthorizationCodeRefresher:
if not isinstance(access_token, str) or not access_token:
return None
binding_proof: Final = await self._identity_validator(
server=server,
token_response=body,
litellm_user_id=user_id,
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented(token.refresh_token),
)
rotated: Final = body.get("refresh_token")
new_refresh: Final = rotated if isinstance(rotated, str) and rotated else token.refresh_token
expires_in: Final = _parse_expires_in(body.get("expires_in"))
scopes: Final = _parse_scopes(body.get("scope")) or token.scopes
await self._persist(user_id, server_id, access_token, new_refresh, expires_in, scopes or None)
if binding_proof is not None:
await self._persist(
user_id,
server_id,
access_token,
new_refresh,
expires_in,
scopes or None,
identity_binding_proof=binding_proof,
)
else:
await self._persist(user_id, server_id, access_token, new_refresh, expires_in, scopes or None)
return OAuthToken(
access_token=access_token,
expires_at=self._clock() + expires_in if expires_in is not None else None,
refresh_token=new_refresh,
scopes=scopes,
identity_binding_proof=binding_proof,
)

1
litellm/proxy/_experimental/mcp_server/outbound_credentials/oauth_token_store.py
View file

@ -41,6 +41,7 @@ class OAuthToken:
expires_at: float | None = None
refresh_token: str | None = None
scopes: tuple[str, ...] = ()
identity_binding_proof: str | None = None
def __repr__(self) -> str:
has_refresh: Final = self.refresh_token is not None

32
litellm/proxy/_experimental/mcp_server/outbound_credentials/per_user_oauth_store.py
View file

@ -12,9 +12,11 @@ from __future__ import annotations
import asyncio
from collections.abc import Callable, Mapping
from functools import partial
from typing import TYPE_CHECKING, Final
from litellm._logging import verbose_logger
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import credential_binding_matches
from litellm.proxy._experimental.mcp_server.outbound_credentials.authz_code_refresher import (
AuthorizationCodeRefresher,
)
@ -69,6 +71,7 @@ async def _persist_credential(
refresh_token: str | None,
expires_in: int | None,
scopes: tuple[str, ...] | None,
identity_binding_proof: str | None = None,
) -> None:
from litellm.proxy._experimental.mcp_server.db import ( # noqa: PLC0415
store_user_oauth_credential,
@ -86,6 +89,7 @@ async def _persist_credential(
expires_in=expires_in,
scopes=list(scopes) if scopes else None,
skip_byok_guard=True,
identity_binding_proof=identity_binding_proof,
)
@ -142,12 +146,26 @@ def _runtime_backend_and_coordinator() -> tuple[TokenCacheBackend | None, Refres
return backend, coordinator, True
async def _read_bound_credential(
server_lookup: ServerLookup, user_id: str, server_id: str
) -> Mapping[str, object] | None:
credential: Final = await _read_credential(user_id, server_id)
server: Final = server_lookup(server_id)
binding: Final = server.oauth_identity_binding if server else None
if credential is not None and binding is not None and binding.mode == "enforce":
if not await credential_binding_matches(binding, user_id, server_id, credential):
return None
return credential
def _build_per_user_oauth_token_store(
server_lookup: ServerLookup,
) -> tuple[CachedOAuthTokenStore, bool]:
backend, coordinator, uses_redis = _runtime_backend_and_coordinator()
refresher: Final = AuthorizationCodeRefresher(server_lookup, _post_token_endpoint, _persist_credential)
refreshing: Final = RefreshingTokenStore(V2PerUserTokenStore(_read_credential), refresher, coordinator=coordinator)
refreshing: Final = RefreshingTokenStore(
V2PerUserTokenStore(partial(_read_bound_credential, server_lookup)), refresher, coordinator=coordinator
)
return CachedOAuthTokenStore(refreshing, default_ttl_seconds=_DEFAULT_TTL_SECONDS, backend=backend), uses_redis
@ -182,6 +200,18 @@ class LazyPerUserOAuthTokenStore:
self._local_fetches = 0
async def fetch(self, user_id: str, server_id: str) -> OAuthToken | None:
token: Final = await self._fetch_token(user_id, server_id)
server: Final = self._server_lookup(server_id)
binding: Final = server.oauth_identity_binding if server else None
if token is not None and binding is not None and binding.mode == "enforce":
if not await credential_binding_matches(
binding, user_id, server_id, {"identity_binding_proof": token.identity_binding_proof}
):
await self.invalidate(user_id, server_id)
return None
return token
async def _fetch_token(self, user_id: str, server_id: str) -> OAuthToken | None:
if self._uses_redis:
store = self._store
if store is not None:

36
litellm/proxy/_experimental/mcp_server/outbound_credentials/token_cache_codec.py
View file

@ -1,24 +1,26 @@
"""Serialize + encrypt boundary for caching an OAuth token in a shared (Redis) cache.
A cross-replica cache must serialize the token, and a plaintext bearer in Redis is a leak, so this
encrypts the value (NaCl in production via the injected ``encrypt``, identity in tests). It caches
**only** the ``access_token``: the hot path needs just the bearer, expiry is carried by the cache
entry's TTL (set from the token's ``expires_at`` by the cache), and the long-lived refresh_token stays
in the DB - the refresh path is always a cache miss that re-reads it - so it never reaches Redis. A
decoded token therefore carries only the bearer (``expires_at`` and ``refresh_token`` both None); the
TTL, not the value, bounds its life. An empty/undecryptable blob (e.g. master-key rotation) is a miss.
Shared cache values contain an encrypted access token and optional identity-binding proof.
Refresh tokens remain in the database; cache TTL bounds the access token's lifetime.
Legacy bearer-only entries decode without proof and cannot satisfy identity enforcement.
"""
from __future__ import annotations
import json
from collections.abc import Callable
from dataclasses import dataclass
from typing import Final
from pydantic import TypeAdapter, ValidationError
from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import (
OAuthToken,
)
_BOUND_PREFIX: Final = "litellm-bound-oauth-v1:"
_BOUND_PAYLOAD: Final = TypeAdapter(dict[str, str])
@dataclass(frozen=True, slots=True)
class OAuthTokenCacheCodec:
@ -26,10 +28,30 @@ class OAuthTokenCacheCodec:
decrypt: Callable[[str], str | None]
def encode(self, token: OAuthToken) -> str:
if token.identity_binding_proof is not None:
return self.encrypt(
_BOUND_PREFIX
+ json.dumps(
{
"access_token": token.access_token,
"identity_binding_proof": token.identity_binding_proof,
}
)
)
return self.encrypt(token.access_token)
def decode(self, blob: str) -> OAuthToken | None:
access_token: Final = self.decrypt(blob)
if not access_token:
return None
if access_token.startswith(_BOUND_PREFIX):
try:
payload: Final = _BOUND_PAYLOAD.validate_json(access_token[len(_BOUND_PREFIX) :])
except ValidationError:
return None
bearer: Final = payload.get("access_token")
proof: Final = payload.get("identity_binding_proof")
if not bearer or not proof:
return None
return OAuthToken(access_token=bearer, identity_binding_proof=proof)
return OAuthToken(access_token=access_token, refresh_token=None)

2
litellm/proxy/_experimental/mcp_server/outbound_credentials/v2_token_store.py
View file

@ -46,11 +46,13 @@ def _to_oauth_token(payload: Mapping[str, object]) -> OAuthToken | None:
return None
refresh_token: Final = payload.get("refresh_token")
expires_at: Final = payload.get("expires_at")
binding_proof: Final = payload.get("identity_binding_proof")
return OAuthToken(
access_token=access_token,
expires_at=_iso_to_epoch(expires_at) if isinstance(expires_at, str) else None,
refresh_token=refresh_token if isinstance(refresh_token, str) else None,
scopes=_to_scopes(payload.get("scopes")),
identity_binding_proof=binding_proof if isinstance(binding_proof, str) else None,
)

22
litellm/proxy/management_endpoints/mcp_management_endpoints.py
View file

@ -2278,6 +2278,28 @@ if MCP_AVAILABLE:
"""Persist the OAuth2 access token obtained by the calling user."""
prisma_client: Final = get_prisma_client_or_throw("Database not connected. Connect a database to your proxy")
await _authorize_and_fetch_mcp_server(prisma_client, user_api_key_dict, server_id)
from litellm.proxy._experimental.mcp_server.mcp_server_manager import ( # noqa: PLC0415 # keep manager import lazy
global_mcp_server_manager as _manager,
)
# This endpoint accepts an opaque token with no upstream identity validation, so it must be
# closed for identity-bound servers or it becomes a bypass of the token-relay binding check.
registry_server: Final = _manager.get_mcp_server_by_id(server_id)
binding: Final = registry_server.oauth_identity_binding if registry_server else None
if binding is not None and binding.mode == "enforce":
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail={ # mutable-ok: FastAPI exception detail requires a JSON-serializable dictionary
"error": "oauth_identity_binding_enforced",
"error_description": (
"Direct credential storage is disabled for this server: its OAuth identity "
"binding is enforced and this endpoint cannot validate the token's principal. "
"Complete the OAuth flow through the gateway instead."
),
"server_id": server_id,
"credential_stored": False,
},
)
user_id: Final = user_api_key_dict.user_id or ""
if not user_id:
raise HTTPException(

32
litellm/types/mcp_server/mcp_server_manager.py
View file

@ -1,7 +1,8 @@
from datetime import datetime
from typing import Any, Final, Literal
from pydantic import BaseModel, ConfigDict, field_validator
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from typing_extensions import Self
from litellm.types.mcp import (
DEFAULT_SUBJECT_TOKEN_TYPE,
@ -38,6 +39,26 @@ class MCPOAuthMetadata(BaseModel):
usable in memory but must never be persisted as configuration."""
class MCPOAuthIdentityBinding(BaseModel):
"""Per-server policy binding stored per-user OAuth credentials to the authenticated LiteLLM caller.
When enabled for an interactive oauth2 server, the token relay validates the upstream OIDC
``id_token`` (signature via the pinned issuer's JWKS, issuer, audience, expiry, nonce) and compares its
principal claim to the LiteLLM caller's trusted identity before the token is returned, stored,
or cached. ``audit`` logs mismatches without changing behavior; ``enforce`` fails closed with
403 ``oauth_principal_mismatch`` and disables the direct ``oauth-user-credential`` POST, which
would otherwise bypass validation with an arbitrary opaque token.
"""
mode: Literal["disabled", "audit", "enforce"] = "disabled"
issuer: str
jwks_url: str | None = None
audiences: list[str] = Field(min_length=1) # mutable-ok: public Pydantic schema requires list values
principal_claim: str = "email"
caller_field: Literal["user_email", "user_id"] = "user_email"
require_email_verified: bool = True
class MCPServer(BaseModel):
server_id: str
name: str
@ -173,6 +194,7 @@ class MCPServer(BaseModel):
# response (supports dot-notation for nested fields, e.g. "team.enterprise_id").
# Tokens that fail validation are rejected before storage.
token_validation: dict[str, Any] | None = None
oauth_identity_binding: MCPOAuthIdentityBinding | None = None
# Optional TTL override (seconds) for the Redis per-user token cache, capped
# at the token's expires_in minus the expiry buffer so a cached entry never
# outlives the token. Defaults to the token's expires_in minus the expiry
@ -225,6 +247,14 @@ class MCPServer(BaseModel):
"""
return self.oauth2_flow == "client_credentials"
@model_validator(mode="after")
def validate_identity_binding_mode(self) -> Self:
binding: Final = self.oauth_identity_binding
if binding is not None and binding.mode != "disabled":
if not self.needs_user_oauth_token or self.delegate_auth_to_upstream:
raise ValueError("oauth_identity_binding requires gateway-managed per-user OAuth2 credentials")
return self
@property
def needs_user_oauth_token(self) -> bool:
"""True if this is an OAuth2 server that relies on per-user tokens (no client_credentials)."""

38
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_authz_code_refresher.py
View file

@ -22,6 +22,7 @@ class _Server:
server_id="srv",
configured_token_url=None,
):
self.oauth_identity_binding = None
self.token_url = token_url
self.configured_token_url = configured_token_url
self.client_id = client_id
@ -287,3 +288,40 @@ async def test_refresh_uses_admin_entered_token_url_when_issuer_yield_empties_re
assert token is not None
assert token.access_token == "new-at"
assert posted[0][0] == "https://idp.example.com/token"
@pytest.mark.asyncio
async def test_identity_rejection_never_persists_or_returns_refreshed_token():
from unittest.mock import AsyncMock
from fastapi import HTTPException
validator = AsyncMock(side_effect=HTTPException(status_code=403, detail="oauth_principal_mismatch"))
persist = AsyncMock()
refresher = AuthorizationCodeRefresher(
_lookup(_Server()),
_endpoint({"access_token": "foreign-token", "id_token": "foreign-identity"}),
persist,
identity_validator=validator,
)
assert await refresher.refresh("alice", "srv", OAuthToken(access_token="old", refresh_token="old-rt")) is None
persist.assert_not_awaited()
@pytest.mark.asyncio
async def test_verified_refresh_preserves_binding_proof_in_storage():
from unittest.mock import AsyncMock
validator = AsyncMock(return_value="verified-binding")
persist = AsyncMock()
refresher = AuthorizationCodeRefresher(
_lookup(_Server()),
_endpoint({"access_token": "new", "refresh_token": "rotated"}),
persist,
identity_validator=validator,
)
token = await refresher.refresh("alice", "srv", OAuthToken(access_token="old", refresh_token="old-rt"))
assert token.access_token == "new"
assert token.refresh_token == "rotated"
assert token.identity_binding_proof == "verified-binding"
assert persist.await_args.kwargs["identity_binding_proof"] == "verified-binding"

91
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_per_user_oauth_store.py
View file

@ -246,3 +246,94 @@ async def test_lazy_store_invalidate_works_after_redis_chain_is_built() -> None:
assert build_calls == 1
assert redis_store.invalidations == [("u", "s")]
@pytest.mark.asyncio
async def test_enforcement_invalidates_cached_legacy_credentials_before_use():
from litellm.types.mcp import MCPAuth, MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
server = MCPServer(
server_id="srv",
name="srv",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
oauth_identity_binding=MCPOAuthIdentityBinding(
mode="enforce",
issuer="https://idp.example.com",
audiences=["client"],
),
)
cached = _RecordingStore("belongs-to-bob")
store = LazyPerUserOAuthTokenStore(
lambda server_id: server,
store_builder=lambda lookup: (cached, False),
redis_available=lambda: False,
)
assert await store.fetch("alice", "srv") is None
assert cached.calls == [("alice", "srv")]
assert cached.invalidations == [("alice", "srv")]
@pytest.mark.asyncio
async def test_enforced_cache_hit_avoids_credential_read_and_rejects_changed_policy(monkeypatch):
from unittest.mock import AsyncMock
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import current_binding_proof
from litellm.proxy._experimental.mcp_server.outbound_credentials import per_user_oauth_store as module
from litellm.types.mcp_server.mcp_server_manager import MCPServer
server = MCPServer(
server_id="srv",
name="srv",
transport="http",
auth_type="oauth2",
oauth_identity_binding={
"mode": "enforce",
"issuer": "https://idp.example",
"audiences": ["client"],
"caller_field": "user_id",
"principal_claim": "sub",
},
)
proof = await current_binding_proof(server.oauth_identity_binding, "alice", "srv")
read = AsyncMock(return_value={"access_token": "alice-token", "identity_binding_proof": proof})
monkeypatch.setattr(module, "_read_credential", read)
monkeypatch.setattr(module, "_runtime_backend_and_coordinator", lambda: (None, None, False))
store = LazyPerUserOAuthTokenStore(lambda _: server, redis_available=lambda: False)
assert (await store.fetch("alice", "srv")).access_token == "alice-token"
assert (await store.fetch("alice", "srv")).access_token == "alice-token"
read.assert_awaited_once_with("alice", "srv")
server.oauth_identity_binding = server.oauth_identity_binding.model_copy(update={"audiences": ["changed"]})
assert await store.fetch("alice", "srv") is None
read.assert_awaited_once()
assert await store.fetch("alice", "srv") is None
assert read.await_count == 2
@pytest.mark.asyncio
async def test_expired_unverified_credential_never_reaches_refresh(monkeypatch):
from unittest.mock import AsyncMock
from litellm.proxy._experimental.mcp_server.outbound_credentials import per_user_oauth_store as module
from litellm.types.mcp_server.mcp_server_manager import MCPServer
server = MCPServer(
server_id="srv",
name="srv",
transport="http",
auth_type="oauth2",
token_url="https://idp.example/token",
oauth_identity_binding={"mode": "enforce", "issuer": "https://idp.example", "audiences": ["client"]},
)
read = AsyncMock(
return_value={"access_token": "bob", "refresh_token": "bob-refresh", "expires_at": "2000-01-01T00:00:00Z"}
)
post = AsyncMock()
monkeypatch.setattr(module, "_read_credential", read)
monkeypatch.setattr(module, "_post_token_endpoint", post)
monkeypatch.setattr(module, "_runtime_backend_and_coordinator", lambda: (None, None, False))
store = LazyPerUserOAuthTokenStore(lambda _: server, redis_available=lambda: False)
assert await store.fetch("alice", "srv") is None
post.assert_not_awaited()

14
tests/test_litellm/proxy/_experimental/mcp_server/outbound_credentials/test_token_cache_codec.py
View file

@ -52,3 +52,17 @@ def test_undecryptable_blob_is_a_miss():
def test_empty_plaintext_is_a_miss():
codec = OAuthTokenCacheCodec(encrypt=lambda s: s, decrypt=lambda b: b)
assert codec.decode("") is None
def test_bound_token_round_trip_preserves_proof_without_refresh_secret():
codec = _wrapping_codec()
blob = codec.encode(OAuthToken("alice-token", refresh_token="private-refresh", identity_binding_proof="proof"))
assert "private-refresh" not in blob
decoded = codec.decode(blob)
assert decoded == OAuthToken("alice-token", identity_binding_proof="proof")
def test_malformed_bound_entries_fail_closed():
codec = _wrapping_codec()
for payload in ("not-json", "{}", '{"access_token":"at"}', '{"access_token":1,"identity_binding_proof":"p"}'):
assert codec.decode("enc:litellm-bound-oauth-v1:" + payload) is None

99
tests/test_litellm/proxy/_experimental/mcp_server/test_db_credentials.py
View file

@ -628,6 +628,7 @@ async def test_oauth_round_trip_returns_payload():
access_token,
refresh_token="rfr-xyz",
scopes=["a", "b"],
identity_binding_proof="verified-proof",
)
stored = _stored_value(prisma)
@ -642,6 +643,7 @@ async def test_oauth_round_trip_returns_payload():
assert result["access_token"] == access_token
assert result["refresh_token"] == "rfr-xyz"
assert result["scopes"] == ["a", "b"]
assert result["identity_binding_proof"] == "verified-proof"
@pytest.mark.asyncio
@ -1184,6 +1186,7 @@ class _RefreshResponse:
def _refresh_server(**overrides):
base = dict(
oauth_identity_binding=None,
token_url="https://idp.example.com/token",
server_id="srv-1",
client_id="cid",
@ -1309,7 +1312,7 @@ async def test_refresh_user_oauth_token_uses_client_secret_basic(monkeypatch):
sends HTTP Basic and keeps the secret out of the body."""
import litellm.proxy._experimental.mcp_server.db as db_mod
server = MagicMock()
server = MagicMock(oauth_identity_binding=None)
server.token_url = "https://idp.example.com/oauth2/token"
server.server_id = "srv"
server.client_id = "cid"
@ -1348,7 +1351,7 @@ async def test_refresh_user_oauth_token_defaults_to_client_secret_post(monkeypat
the body (client_secret_post) and sends no Authorization header."""
import litellm.proxy._experimental.mcp_server.db as db_mod
server = MagicMock()
server = MagicMock(oauth_identity_binding=None)
server.token_url = "https://idp.example.com/oauth2/token"
server.server_id = "srv"
server.client_id = "cid"
@ -1609,3 +1612,95 @@ def test_partial_update_defers_omitted_eligibility_fields_to_the_stored_row():
request = UpdateMCPServerRequest(server_id="relay-update", per_server_oauth_discovery=True)
assert request.per_server_oauth_discovery is True
@pytest.mark.asyncio
async def test_enforcement_rejects_preexisting_unverified_credential():
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
server = MCPServer(
server_id="srv-1", name="srv-1", url="https://mcp.example.com", transport=MCPTransport.http, auth_type=MCPAuth.oauth2,
oauth_identity_binding=MCPOAuthIdentityBinding(
mode="enforce", issuer="https://idp.example.com", audiences=["client"],
),
)
result = await resolve_valid_user_oauth_token(
user_id="alice", server=server,
cred={"access_token": "belongs-to-bob", "refresh_token": "bobs-refresh-token"},
)
assert result is None
@pytest.mark.asyncio
async def test_refresh_identity_rejection_returns_reauthentication_without_persisting(monkeypatch):
from fastapi import HTTPException
from litellm.proxy._experimental.mcp_server import db as module
validator = AsyncMock(side_effect=HTTPException(status_code=403, detail="oauth_principal_mismatch"))
monkeypatch.setattr(module, "enforce_oauth_identity_binding", validator)
result, captured = await _run_refresh(
monkeypatch, _refresh_server(), {"access_token": "bob", "refresh_token": "rotated"}
)
assert result is None
assert captured["data"]["grant_type"] == "refresh_token"
module.store_user_oauth_credential.assert_not_awaited()
@pytest.mark.asyncio
async def test_verified_legacy_cache_reads_avoid_database_and_reject_policy_changes(monkeypatch):
from litellm.caching.caching import DualCache
from litellm.proxy import proxy_server
from litellm.proxy._experimental.mcp_server import db as module
from litellm.proxy._experimental.mcp_server.oauth2_token_cache import mcp_per_user_token_cache
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import current_binding_proof
from litellm.types.mcp_server.mcp_server_manager import MCPServer
server = MCPServer(
server_id="srv",
name="srv",
transport="http",
auth_type="oauth2",
oauth_identity_binding={
"mode": "enforce",
"issuer": "https://idp.example",
"audiences": ["client"],
"caller_field": "user_id",
"principal_claim": "sub",
},
)
proof = await current_binding_proof(server.oauth_identity_binding, "alice", "srv")
monkeypatch.setattr(proxy_server, "user_api_key_cache", DualCache())
read = AsyncMock(return_value=None)
monkeypatch.setattr(module, "get_user_oauth_credential", read)
monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
await mcp_per_user_token_cache.set("alice", "srv", "alice-token", 60, identity_binding_proof=proof)
assert await module.resolve_user_oauth_access_token("alice", server) == "alice-token"
assert await module.resolve_user_oauth_access_token("alice", server) == "alice-token"
read.assert_not_awaited()
server.oauth_identity_binding = server.oauth_identity_binding.model_copy(update={"audiences": ["changed"]})
assert await module.resolve_user_oauth_access_token("alice", server) is None
read.assert_awaited_once()
assert await mcp_per_user_token_cache.get_token("alice", "srv") is None
@pytest.mark.asyncio
async def test_unverified_legacy_cache_cannot_bypass_enforcement(monkeypatch):
from litellm.caching.caching import DualCache
from litellm.proxy import proxy_server
from litellm.proxy._experimental.mcp_server import db as module
from litellm.proxy._experimental.mcp_server.oauth2_token_cache import mcp_per_user_token_cache
from litellm.types.mcp_server.mcp_server_manager import MCPServer
server = MCPServer(
server_id="srv",
name="srv",
transport="http",
auth_type="oauth2",
oauth_identity_binding={"mode": "enforce", "issuer": "https://idp.example", "audiences": ["client"]},
)
monkeypatch.setattr(proxy_server, "user_api_key_cache", DualCache())
monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
monkeypatch.setattr(module, "get_user_oauth_credential", AsyncMock(return_value={"access_token": "bob"}))
await mcp_per_user_token_cache.set("alice", "srv", "bob", 60)
assert await module.resolve_user_oauth_access_token("alice", server) is None
assert await mcp_per_user_token_cache.get("alice", "srv") is None

225
tests/test_litellm/proxy/_experimental/mcp_server/test_discoverable_endpoints.py
View file

@ -8123,6 +8123,137 @@ async def test_token_exchange_pairs_client_secret_with_server_client_id():
assert "client_secret" not in sent
@pytest.mark.asyncio
async def test_token_exchange_refresh_passes_presented_refresh_ownership():
from fastapi import Request
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
exchange_token_with_server,
)
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import RefreshTokenPresented
from litellm.proxy._types import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
server = MCPServer(
server_id="srv-1",
name="srv-1",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
client_id="cid",
token_url="https://provider.example/token",
oauth_identity_binding=MCPOAuthIdentityBinding(
mode="enforce",
issuer="https://provider.example",
audiences=["cid"],
),
)
request = MagicMock(spec=Request)
request.base_url = "https://litellm.example.com/"
request.headers = {}
response = MagicMock()
response.json.return_value = {"access_token": "at"}
response.raise_for_status = MagicMock()
client = MagicMock()
client.post = AsyncMock(return_value=response)
enforce = AsyncMock()
with (
patch( # test-quality-ok: no injection seam exists for the exchange's HTTP and identity collaborators
"litellm.proxy._experimental.mcp_server.discoverable_endpoints.get_async_httpx_client",
return_value=client,
),
patch( # test-quality-ok: no injection seam exists for request identity extraction
"litellm.proxy._experimental.mcp_server.discoverable_endpoints._extract_user_id_from_request",
new=AsyncMock(return_value="user-a"),
),
patch( # test-quality-ok: captures the ownership value at the exchange boundary
"litellm.proxy._experimental.mcp_server.discoverable_endpoints.enforce_oauth_identity_binding",
new=enforce,
),
):
await exchange_token_with_server(
request=request,
mcp_server=server,
grant_type="refresh_token",
code=None,
redirect_uri=None,
client_id="cid",
client_secret=None,
code_verifier=None,
refresh_token="rt-1",
)
ownership = enforce.await_args.kwargs["refresh_ownership"]
assert isinstance(ownership, RefreshTokenPresented)
assert ownership.refresh_token == "rt-1"
@pytest.mark.asyncio
async def test_token_exchange_authorization_code_passes_no_refresh_ownership(monkeypatch):
from fastapi import Request
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
exchange_token_with_server,
seal_bridge_authorization_code,
)
from litellm.proxy._types import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
monkeypatch.setenv("LITELLM_SALT_KEY", "identity-binding-test-salt")
server = MCPServer(
server_id="srv-1",
name="srv-1",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
client_id="cid",
token_url="https://provider.example/token",
oauth_identity_binding=MCPOAuthIdentityBinding(
mode="enforce",
issuer="https://provider.example",
audiences=["cid"],
),
)
request = MagicMock(spec=Request)
request.base_url = "https://litellm.example.com/"
request.headers = {}
response = MagicMock()
response.json.return_value = {"access_token": "at"}
response.raise_for_status = MagicMock()
client = MagicMock()
client.post = AsyncMock(return_value=response)
enforce = AsyncMock()
with (
patch( # test-quality-ok: no injection seam exists for the exchange's HTTP and identity collaborators
"litellm.proxy._experimental.mcp_server.discoverable_endpoints.get_async_httpx_client",
return_value=client,
),
patch( # test-quality-ok: no injection seam exists for request identity extraction
"litellm.proxy._experimental.mcp_server.discoverable_endpoints._extract_user_id_from_request",
new=AsyncMock(return_value="user-a"),
),
patch( # test-quality-ok: captures the ownership value at the exchange boundary
"litellm.proxy._experimental.mcp_server.discoverable_endpoints.enforce_oauth_identity_binding",
new=enforce,
),
):
result = await exchange_token_with_server(
request=request,
mcp_server=server,
grant_type="authorization_code",
code=seal_bridge_authorization_code("auth-code", "user-a", "srv-1", "login-nonce"),
redirect_uri="https://litellm.example.com/callback",
client_id="cid",
client_secret=None,
code_verifier="test-verifier",
)
assert result.status_code == 200
assert json.loads(result.body)["access_token"] == "at"
assert enforce.await_args.kwargs["refresh_ownership"] is None
assert enforce.await_args.kwargs["expected_nonce"] == "login-nonce"
def _upstream_token_response(status_code: int, *, json_body: object = None, text_body: str = "") -> "httpx.Response":
import httpx
@ -10916,3 +11047,97 @@ def test_introspect_route_answers_for_authenticated_caller(monkeypatch):
assert active.status_code == 200
assert active.json()["active"] is True
assert active.json()["sub"] == "u1"
@pytest.mark.asyncio
async def test_identity_bound_authorization_carries_nonce_and_caller_through_callback(monkeypatch):
from http.cookies import SimpleCookie
from urllib.parse import parse_qs, urlparse
from fastapi import Request
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import (
_oauth_state_cookie_name, authorize_with_server, callback, open_bridge_authorization_code,
)
from litellm.types.mcp import MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
monkeypatch.setenv("LITELLM_SALT_KEY", "identity-binding-test-salt")
server = MCPServer(
server_id="srv", name="srv", transport=MCPTransport.http, auth_type=MCPAuth.oauth2,
client_id="client", authorization_url="https://idp.example.com/authorize",
token_url="https://idp.example.com/token",
oauth_identity_binding=MCPOAuthIdentityBinding(
mode="enforce", issuer="https://idp.example.com", audiences=["client"],
),
)
request = Request({"type": "http", "scheme": "https", "server": ("proxy.example.com", 443),
"path": "/authorize", "query_string": b"", "headers": []})
with (
patch( # test-quality-ok: isolate authenticated request resolution from the real encrypted OAuth round trip
"litellm.proxy._experimental.mcp_server.discoverable_endpoints._extract_user_id_from_request",
new=AsyncMock(return_value="alice")),
patch( # test-quality-ok: isolate user access lookup while testing nonce and caller preservation
"litellm.proxy._experimental.mcp_server.discoverable_endpoints._bridge_authorize_access_denial",
new=AsyncMock(return_value=None)),
):
authorized = await authorize_with_server(
request, server, "client", "http://127.0.0.1:6274/callback", state="client-state",
code_challenge="pkce-challenge", code_challenge_method="S256",
)
query = parse_qs(urlparse(authorized.headers["location"]).query)
assert len(query["nonce"][0]) >= 32
cookies = SimpleCookie()
cookies.load(authorized.headers["set-cookie"])
name = _oauth_state_cookie_name(query["state"][0])
callback_request = Request({**request.scope, "path": "/callback",
"headers": [(b"cookie", f"{name}={cookies[name].value}".encode())]})
completed = await callback(callback_request, code="upstream-code", state=query["state"][0])
returned = parse_qs(urlparse(completed.headers["location"]).query)
sealed = open_bridge_authorization_code(returned["code"][0])
assert sealed.litellm_user_id == "alice"
assert sealed.mcp_server_id == "srv"
assert sealed.upstream_code == "upstream-code"
assert sealed.oauth_nonce == query["nonce"][0]
assert returned["state"] == ["client-state"]
@pytest.mark.asyncio
async def test_enforced_login_warms_verified_token_readable_without_database_lookup(monkeypatch):
from types import SimpleNamespace
from litellm.caching.caching import DualCache
from litellm.proxy import proxy_server
from litellm.proxy._experimental.mcp_server import db, mcp_server_manager
from litellm.proxy._experimental.mcp_server.discoverable_endpoints import _store_per_user_token_server_side
from litellm.proxy._experimental.mcp_server.oauth2_token_cache import mcp_per_user_token_cache
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import current_binding_proof
from litellm.proxy._experimental.mcp_server.outbound_credentials.dual_cache_token_backend import DualCacheTokenCacheBackend
from litellm.proxy._experimental.mcp_server.outbound_credentials.oauth_token_store import CachedOAuthTokenStore
from litellm.proxy._experimental.mcp_server.outbound_credentials.per_user_oauth_store import LazyPerUserOAuthTokenStore
from litellm.proxy._experimental.mcp_server.outbound_credentials.v2_token_store import V2PerUserTokenStore
from litellm.types.mcp_server.mcp_server_manager import MCPServer
server = MCPServer(
server_id="srv", name="srv", transport="http", auth_type="oauth2",
oauth_identity_binding={"mode": "enforce", "issuer": "https://idp.example", "audiences": ["client"],
"caller_field": "user_id", "principal_claim": "sub"},
)
proof = await current_binding_proof(server.oauth_identity_binding, "alice", "srv")
cache = DualCache()
monkeypatch.setenv("LITELLM_SALT_KEY", "test-cache-warm-encryption-salt")
monkeypatch.setattr(proxy_server, "user_api_key_cache", cache)
monkeypatch.setattr(proxy_server, "prisma_client", MagicMock())
monkeypatch.setattr(mcp_server_manager, "global_mcp_server_manager", SimpleNamespace(invalidate_user_oauth_token_cache=AsyncMock()))
monkeypatch.setattr(db, "store_user_oauth_credential", AsyncMock())
await _store_per_user_token_server_side(
server=server, user_id="alice", token_response={"access_token": "alice-token", "refresh_token": "private", "expires_in": 3600},
identity_binding_proof=proof,
)
read = AsyncMock(return_value=None)
cached = CachedOAuthTokenStore(
V2PerUserTokenStore(read), default_ttl_seconds=300, backend=DualCacheTokenCacheBackend(cache, mcp_per_user_token_cache._codec()),
)
store = LazyPerUserOAuthTokenStore(lambda _: server, store_builder=lambda _: (cached, True), redis_available=lambda: True)
token = await store.fetch("alice", "srv")
assert token is not None and token.access_token == "alice-token"
assert token.identity_binding_proof == proof
assert token.refresh_token is None
read.assert_not_awaited()

667
tests/test_litellm/proxy/_experimental/mcp_server/test_oauth_identity_binding.py Normal file
View file

@ -0,0 +1,667 @@
import time
from collections.abc import Mapping
from types import SimpleNamespace
from typing import Final
from unittest.mock import AsyncMock, MagicMock, patch
import jwt
import pytest
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from fastapi import HTTPException
from pydantic import ValidationError
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import (
RefreshOwnershipProven,
RefreshTokenPresented,
VerifiedRefreshToken,
_discover_jwks_url,
_fetch_issuer_jwks,
_load_caller_principal,
_load_stored_refresh_token,
_select_signing_key,
current_binding_proof,
enforce_oauth_identity_binding,
)
from litellm.types.mcp import MCPAuth, MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
ISSUER: Final = "https://idp.example.com"
AUDIENCE: Final = "litellm-client"
KID: Final = "test-key"
_PRIVATE_KEY: Final = rsa.generate_private_key(public_exponent=65537, key_size=2048)
_PRIVATE_PEM: Final = _PRIVATE_KEY.private_bytes(
encoding=serialization.Encoding.PEM,
format=serialization.PrivateFormat.PKCS8,
encryption_algorithm=serialization.NoEncryption(),
)
_PUBLIC_JWK: Final = {
**jwt.algorithms.RSAAlgorithm.to_jwk(_PRIVATE_KEY.public_key(), as_dict=True),
"kid": KID,
"alg": "RS256",
"use": "sig",
}
def _sign_id_token(claims: Mapping[str, object]) -> str:
payload: Final = {
"iss": ISSUER,
"aud": AUDIENCE,
"exp": int(time.time()) + 300,
"iat": int(time.time()),
"nonce": "test-nonce",
"sub": "upstream-user",
**claims,
}
return jwt.encode(payload, _PRIVATE_PEM, algorithm="RS256", headers={"kid": KID})
async def _jwks_fetcher(_binding: MCPOAuthIdentityBinding) -> list[Mapping[str, object]]:
return [_PUBLIC_JWK]
def _caller_loader(email: str | None):
async def load(_user_id: str, _binding: MCPOAuthIdentityBinding) -> str | None:
return email
return load
def _stored_refresh_token_loader(refresh_token: str | None):
async def load(_user_id: str, _server_id: str, _binding: MCPOAuthIdentityBinding) -> VerifiedRefreshToken | None:
return VerifiedRefreshToken(refresh_token, "verified-binding") if refresh_token else None
return load
def _server(mode: str = "enforce", **binding_overrides: object) -> MCPServer:
return MCPServer(
server_id="srv-1",
name="srv-1",
url="https://mcp.example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
oauth_identity_binding=MCPOAuthIdentityBinding(
mode=mode,
issuer=ISSUER,
audiences=[AUDIENCE],
**binding_overrides,
),
)
@pytest.mark.asyncio
async def test_fetch_issuer_jwks_fetches_and_caches_keys():
binding: Final = _server(jwks_url="https://idp.example.com/jwks-cache").oauth_identity_binding
assert binding is not None
response: Final = MagicMock()
response.json.return_value = {"keys": [_PUBLIC_JWK]}
client: Final = MagicMock()
client.get = AsyncMock(return_value=response)
with patch( # test-quality-ok: no HTTP dependency injection seam exists for JWKS fetching
"litellm.proxy._experimental.mcp_server.oauth_identity_binding.get_async_httpx_client",
return_value=client,
):
first: Final = await _fetch_issuer_jwks(binding)
second: Final = await _fetch_issuer_jwks(binding)
assert first == second == [_PUBLIC_JWK]
client.get.assert_awaited_once_with("https://idp.example.com/jwks-cache")
@pytest.mark.asyncio
async def test_fetch_issuer_jwks_rejects_malformed_document():
binding: Final = _server(jwks_url="https://idp.example.com/jwks-invalid").oauth_identity_binding
assert binding is not None
response: Final = MagicMock()
response.json.return_value = {}
client: Final = MagicMock()
client.get = AsyncMock(return_value=response)
with patch( # test-quality-ok: no HTTP dependency injection seam exists for JWKS fetching
"litellm.proxy._experimental.mcp_server.oauth_identity_binding.get_async_httpx_client",
return_value=client,
):
with pytest.raises(TypeError, match="has no 'keys' array"):
await _fetch_issuer_jwks(binding)
@pytest.mark.asyncio
async def test_discover_jwks_url_returns_provider_uri():
response: Final = MagicMock()
response.json.return_value = {"jwks_uri": "https://idp.example.com/jwks"}
client: Final = MagicMock()
client.get = AsyncMock(return_value=response)
with patch( # test-quality-ok: no HTTP dependency injection seam exists for OIDC discovery
"litellm.proxy._experimental.mcp_server.oauth_identity_binding.get_async_httpx_client",
return_value=client,
):
result: Final = await _discover_jwks_url("https://idp.example.com/")
assert result == "https://idp.example.com/jwks"
client.get.assert_awaited_once_with("https://idp.example.com/.well-known/openid-configuration")
@pytest.mark.asyncio
async def test_discover_jwks_url_rejects_missing_provider_uri():
response: Final = MagicMock()
response.json.return_value = {}
client: Final = MagicMock()
client.get = AsyncMock(return_value=response)
with patch( # test-quality-ok: no HTTP dependency injection seam exists for OIDC discovery
"litellm.proxy._experimental.mcp_server.oauth_identity_binding.get_async_httpx_client",
return_value=client,
):
with pytest.raises(ValueError, match="returned no jwks_uri"):
await _discover_jwks_url("https://idp.example.com")
def test_select_signing_key_returns_matching_key_or_rejection():
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": True})
selected: Final = _select_signing_key(token, [_PUBLIC_JWK])
rejected: Final = _select_signing_key(token, [])
assert selected.__class__.__name__ == "PyJWK"
assert rejected.code == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_load_caller_principal_supports_user_id_and_database_email():
user_id_binding: Final = _server(caller_field="user_id").oauth_identity_binding
assert user_id_binding is not None
assert await _load_caller_principal("user-a", user_id_binding) == "user-a"
with patch( # test-quality-ok: caller loading is a lazy database boundary without injection
"litellm.proxy._experimental.mcp_server.bridge_token_flow.load_active_user_by_id",
new=AsyncMock(side_effect=["no_active_key", SimpleNamespace(user_email="alice@example.com")]),
):
assert await _load_caller_principal("user-a", _server().oauth_identity_binding) is None
assert await _load_caller_principal("user-a", _server().oauth_identity_binding) == "alice@example.com"
@pytest.mark.asyncio
async def test_load_stored_refresh_token_returns_credential_and_fails_closed():
binding: Final = _server(caller_field="user_id", principal_claim="sub").oauth_identity_binding
proof: Final = await current_binding_proof(binding, "user-a", "srv-1")
get_credential: Final = AsyncMock(return_value={"refresh_token": "rt-1", "identity_binding_proof": proof})
with (
patch( # test-quality-ok: stored-token loading is a lazy database boundary without injection
"litellm.proxy._experimental.mcp_server.db.get_user_oauth_credential",
new=get_credential,
),
patch( # test-quality-ok: stored-token loading is a lazy database boundary without injection
"litellm.proxy.utils.get_prisma_client_or_throw",
return_value="prisma",
),
):
assert await _load_stored_refresh_token("user-a", "srv-1", binding) == VerifiedRefreshToken("rt-1", proof)
get_credential.return_value = {"refresh_token": "rt-1"}
assert await _load_stored_refresh_token("user-a", "srv-1", binding) is None
with patch( # test-quality-ok: stored-token loading is a lazy database boundary without injection
"litellm.proxy.utils.get_prisma_client_or_throw",
side_effect=RuntimeError("database unavailable"),
):
assert await _load_stored_refresh_token("user-a", "srv-1", binding) is None
@pytest.mark.asyncio
async def test_matching_principal_passes():
token: Final = _sign_id_token({"email": "Alice@Example.com", "email_verified": True})
result: Final = await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert result is not None
@pytest.mark.asyncio
async def test_mismatched_principal_rejected():
token: Final = _sign_id_token({"email": "mallory@example.com", "email_verified": True})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_principal_mismatch"
assert exc_info.value.detail["credential_stored"] is False
@pytest.mark.asyncio
async def test_missing_upstream_principal_rejected():
token: Final = _sign_id_token({"email_verified": True})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
assert "no usable" in exc_info.value.detail["error_description"]
@pytest.mark.asyncio
async def test_jwks_fetch_failure_is_rejected():
async def fail(_binding: MCPOAuthIdentityBinding) -> list[Mapping[str, object]]:
raise RuntimeError("jwks unavailable")
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": True})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=fail,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
assert "jwks unavailable" in exc_info.value.detail["error_description"]
@pytest.mark.asyncio
async def test_missing_signing_key_is_rejected():
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": True})
async def no_keys(_binding: MCPOAuthIdentityBinding) -> tuple[Mapping[str, object], ...]:
return ()
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=no_keys,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
assert "signing key" in exc_info.value.detail["error_description"]
@pytest.mark.asyncio
async def test_missing_caller_principal_is_rejected():
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": True})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader(None),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
assert "has no" in exc_info.value.detail["error_description"]
@pytest.mark.asyncio
async def test_refresh_without_id_token_requires_litellm_identity_for_presented_token():
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id=None,
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented("rt-1"),
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
stored_refresh_token_loader=_stored_refresh_token_loader("rt-1"),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
assert "no resolvable LiteLLM user identity" in exc_info.value.detail["error_description"]
@pytest.mark.asyncio
async def test_user_id_principal_matching_uses_exact_comparison():
token: Final = _sign_id_token({"sub": "user-a"})
result: Final = await enforce_oauth_identity_binding(
server=_server(principal_claim="sub", caller_field="user_id"),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("user-a"),
)
assert result is not None
@pytest.mark.asyncio
async def test_missing_id_token_rejected_on_authorization_code():
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_refresh_without_id_token_allowed_when_presented_token_matches_stored_credential():
result: Final = await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id="user-a",
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented("rt-1"),
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
stored_refresh_token_loader=_stored_refresh_token_loader("rt-1"),
)
assert result is not None
@pytest.mark.asyncio
async def test_refresh_without_id_token_rejects_different_presented_token():
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id="user-a",
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented("rt-stolen"),
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
stored_refresh_token_loader=_stored_refresh_token_loader("rt-1"),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_refresh_without_id_token_rejects_missing_stored_token():
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id="user-a",
grant_type="refresh_token",
refresh_ownership=RefreshTokenPresented("rt-1"),
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
stored_refresh_token_loader=_stored_refresh_token_loader(None),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_identity_envelope_does_not_prove_upstream_binding():
with pytest.raises(HTTPException) as error:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id="user-a",
grant_type="refresh_token",
refresh_ownership=RefreshOwnershipProven(),
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert error.value.status_code == 403
@pytest.mark.asyncio
async def test_refresh_without_id_token_rejects_without_ownership_proof():
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at"},
litellm_user_id="user-a",
grant_type="refresh_token",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
stored_refresh_token_loader=_stored_refresh_token_loader("rt-1"),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_refresh_with_mismatched_id_token_rejected():
token: Final = _sign_id_token({"email": "mallory@example.com", "email_verified": True})
with pytest.raises(HTTPException):
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="refresh_token",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
@pytest.mark.asyncio
async def test_audit_mode_logs_but_does_not_reject(caplog):
token: Final = _sign_id_token({"email": "mallory@example.com", "email_verified": True})
result: Final = await enforce_oauth_identity_binding(
server=_server(mode="audit"),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert result is None
assert "oauth_principal_mismatch" in caplog.text
assert "nonce" not in caplog.text
@pytest.mark.asyncio
async def test_unverified_email_rejected():
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": False})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_wrong_issuer_rejected():
payload: Final = {
"iss": "https://evil.example.com",
"aud": AUDIENCE,
"exp": int(time.time()) + 300,
"email": "alice@example.com",
"email_verified": True,
}
token: Final = jwt.encode(payload, _PRIVATE_PEM, algorithm="RS256", headers={"kid": KID})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_no_litellm_identity_rejected():
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": True})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id=None,
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.status_code == 403
@pytest.mark.asyncio
async def test_disabled_binding_is_noop():
result: Final = await enforce_oauth_identity_binding(
server=_server(mode="disabled"),
token_response={"access_token": "at"},
litellm_user_id=None,
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader(None),
)
assert result is None
@pytest.mark.asyncio
async def test_no_binding_is_noop():
server: Final = MCPServer(
server_id="srv-2",
name="srv-2",
url="https://mcp.example.com",
transport=MCPTransport.http,
)
result: Final = await enforce_oauth_identity_binding(
server=server,
token_response={"access_token": "at"},
litellm_user_id=None,
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader(None),
)
assert result is None
def test_identity_binding_requires_non_empty_audiences():
with pytest.raises(ValidationError):
MCPOAuthIdentityBinding(mode="enforce", issuer=ISSUER, audiences=[])
with pytest.raises(ValidationError):
MCPOAuthIdentityBinding(mode="enforce", issuer=ISSUER)
@pytest.mark.asyncio
async def test_wrong_audience_rejected():
token: Final = _sign_id_token({"aud": "other-client", "email": "alice@example.com", "email_verified": True})
with pytest.raises(HTTPException) as exc_info:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
expected_nonce="test-nonce",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
@pytest.mark.parametrize("nonce", [None, "another-login"])
async def test_authorization_code_rejects_missing_or_foreign_nonce(nonce):
token = _sign_id_token({"email": "alice@example.com", "email_verified": True, "nonce": nonce})
with pytest.raises(HTTPException) as error:
await enforce_oauth_identity_binding(
server=_server(),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
refresh_ownership=None,
expected_nonce="this-login",
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert error.value.status_code == 403
assert error.value.detail["error"] == "oauth_identity_binding_failed"
@pytest.mark.asyncio
async def test_binding_proof_rejects_changed_user_or_policy():
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import credential_binding_matches
binding = _server(caller_field="user_id", principal_claim="sub").oauth_identity_binding
proof = await current_binding_proof(binding, "alice", "srv-1")
credential = {"identity_binding_proof": proof}
assert await credential_binding_matches(binding, "alice", "srv-1", credential)
assert not await credential_binding_matches(binding, "bob", "srv-1", credential)
assert not await credential_binding_matches(binding, "alice", "other-server", credential)
changed = binding.model_copy(update={"audiences": ["different-client"]})
assert not await credential_binding_matches(changed, "alice", "srv-1", credential)
@pytest.mark.parametrize("auth_type", [MCPAuth.oauth_delegate, MCPAuth.true_passthrough])
def test_identity_binding_rejects_modes_without_gateway_credential_custody(auth_type):
with pytest.raises(ValidationError, match="gateway-managed per-user"):
MCPServer(
server_id="srv",
name="srv",
transport=MCPTransport.http,
auth_type=auth_type,
oauth_identity_binding=_server().oauth_identity_binding,
)
@pytest.mark.asyncio
async def test_audit_matching_login_without_nonce_does_not_report_failure(caplog):
token: Final = _sign_id_token({"email": "alice@example.com", "email_verified": True})
result: Final = await enforce_oauth_identity_binding(
server=_server(mode="audit"),
token_response={"access_token": "at", "id_token": token},
litellm_user_id="user-a",
grant_type="authorization_code",
refresh_ownership=None,
jwks_fetcher=_jwks_fetcher,
caller_principal_loader=_caller_loader("alice@example.com"),
)
assert result is None
assert "oauth_identity_binding audit" not in caplog.text

66
tests/test_litellm/proxy/management_endpoints/test_mcp_management_endpoints.py
View file

@ -4802,6 +4802,72 @@ async def test_store_mcp_oauth_user_credential_returns_status():
assert result.expires_at == "2099-01-01T00:00:00+00:00"
@pytest.mark.asyncio
async def test_store_mcp_oauth_user_credential_blocked_when_identity_binding_enforced():
"""The direct opaque-token POST must be closed for enforce-mode identity-bound servers,
otherwise it bypasses the token-relay principal check."""
from litellm.proxy._types import MCPOAuthUserCredentialRequest
from litellm.types.mcp import MCPAuth, MCPTransport
from litellm.types.mcp_server.mcp_server_manager import MCPOAuthIdentityBinding, MCPServer
if not mgmt_endpoints.MCP_AVAILABLE:
pytest.skip("MCP module not installed")
from litellm.proxy._experimental.mcp_server import mcp_server_manager as manager_module
from litellm.proxy.management_endpoints.mcp_management_endpoints import (
store_mcp_oauth_user_credential,
)
server_id = "srv-binding-1"
bound_server = MCPServer(
server_id=server_id,
name=server_id,
url="https://mcp.example.com",
transport=MCPTransport.http,
auth_type=MCPAuth.oauth2,
oauth_identity_binding=MCPOAuthIdentityBinding(
mode="enforce",
issuer="https://idp.example.com",
audiences=["litellm-client"],
),
)
store_mock = AsyncMock(return_value=None)
with (
patch( # test-quality-ok: mirrors the existing store-credential tests in this file
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_prisma_client_or_throw",
return_value=_make_prisma_client(),
),
patch( # test-quality-ok: mirrors the existing store-credential tests in this file
"litellm.proxy.management_endpoints.mcp_management_endpoints.get_mcp_server",
new=AsyncMock(return_value=generate_mock_mcp_server_db_record(server_id=server_id)),
),
patch( # test-quality-ok: mirrors the existing store-credential tests in this file
"litellm.proxy.management_endpoints.mcp_management_endpoints._user_has_admin_view",
return_value=True,
),
patch.object( # test-quality-ok: registry is a module-level singleton; injecting it would change the endpoint signature
manager_module.global_mcp_server_manager,
"get_mcp_server_by_id",
return_value=bound_server,
),
patch( # test-quality-ok: asserting the DB write is never reached is the point of the test
"litellm.proxy.management_endpoints.mcp_management_endpoints.store_user_oauth_credential",
new=store_mock,
),
):
with pytest.raises(HTTPException) as exc_info:
await store_mcp_oauth_user_credential(
server_id=server_id,
payload=MCPOAuthUserCredentialRequest(access_token="opaque-tok", expires_in=3600),
user_api_key_dict=_make_user_auth("user-123"),
)
assert exc_info.value.status_code == 403
assert exc_info.value.detail["error"] == "oauth_identity_binding_enforced"
store_mock.assert_not_called()
@pytest.mark.asyncio
async def test_delete_mcp_oauth_user_credential_only_deletes_oauth():
"""delete_mcp_oauth_user_credential only deletes OAuth2 credentials, not BYOK."""