scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-10-12 03:48:56 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 57

fix(auth): FIPS-aware JWT algorithm allowlists for proxy auth and MCP (LIT-8429)

Browse source 
Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>
This commit is contained in:
yucheng 2026-09-24 08:09:55 +00:00
parent f05002acb2
commit 6dff6ae5bc
14 changed files with 1241 additions and 819 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

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

@ -169,6 +169,7 @@ from litellm.proxy._types import (
is_per_server_oauth_discovery_eligible,
)
from litellm.proxy.auth.ip_address_utils import IPAddressUtils
from litellm.proxy.auth.jwt_algorithms import APPROVED_JWT_ALGORITHMS, ApprovedJwtAlgorithm
from litellm.proxy.common_utils.encrypt_decrypt_utils import decrypt_value_helper
from litellm.proxy.common_utils.user_api_key_cache import get_management_object_ttl
from litellm.proxy.management_endpoints.sso.id_jag_assertion_capture import (
@ -386,7 +387,7 @@ class MCPServerConfig(TypedDict, total=False):
id_jag_resource: str
client_private_key: str
client_private_key_id: str
client_assertion_signing_alg: str
client_assertion_signing_alg: ApprovedJwtAlgorithm
timeout: float
max_concurrent_requests: int
@ -1482,6 +1483,18 @@ def _warn_on_shared_identifier_prefixes(servers: Iterable[MCPServer]) -> None:
)
def _stored_client_assertion_signing_alg(value: object, server_name: str) -> ApprovedJwtAlgorithm:
if value in APPROVED_JWT_ALGORITHMS:
return cast(ApprovedJwtAlgorithm, value)
if value is not None:
verbose_logger.warning(
"MCP server %s: client_assertion_signing_alg %r is not an approved algorithm, using RS256",
server_name,
value,
)
return "RS256"
def _warn_legacy_delegate_auth_if_applicable(server: MCPServer, *, source: str) -> None:
"""Direct legacy delegated OAuth configurations to the admitted replacement."""
if server.auth_type != MCPAuth.oauth2:
@ -2604,7 +2617,10 @@ class MCPServerManager:
id_jag_resource=server_config.get("id_jag_resource", None),
client_private_key=server_config.get("client_private_key", None),
client_private_key_id=server_config.get("client_private_key_id", None),
client_assertion_signing_alg=server_config.get("client_assertion_signing_alg", "RS256"),
client_assertion_signing_alg=_stored_client_assertion_signing_alg(
server_config.get("client_assertion_signing_alg"),
str(server_config.get("alias") or server_config.get("server_name") or server_id),
),
token_exchange_profile=server_config.get("token_exchange_profile", "rfc8693"),
allow_sampling=bool(server_config.get("allow_sampling", False)),
allow_elicitation=bool(server_config.get("allow_elicitation", False)),
@ -3178,10 +3194,10 @@ class MCPServerManager:
credentials_are_encrypted,
),
client_private_key_id=(credentials_dict.get("client_private_key_id") if credentials_dict else None),
client_assertion_signing_alg=(
credentials_dict.get("client_assertion_signing_alg") if credentials_dict else None
)
or "RS256",
client_assertion_signing_alg=_stored_client_assertion_signing_alg(
credentials_dict.get("client_assertion_signing_alg") if credentials_dict else None,
name_for_prefix,
),
token_exchange_profile=mcp_server.token_exchange_profile
or (credentials_dict.get("token_exchange_profile") if credentials_dict else None)
or "rfc8693",

15
litellm/proxy/_experimental/mcp_server/oauth_identity_binding.py
View file

@ -23,20 +23,11 @@ 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.proxy.auth.jwt_algorithms import APPROVED_JWT_ALGORITHMS, jwks_keys_for
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",
)
_ALLOWED_ID_TOKEN_ALGORITHMS: Final = APPROVED_JWT_ALGORITHMS
_JWKS_CACHE_TTL_SECONDS: Final = 3600
_jwks_cache: Final = InMemoryCache(default_ttl=_JWKS_CACHE_TTL_SECONDS)
@ -126,7 +117,7 @@ async def _discover_jwks_url(issuer: str) -> str:
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:
for key in jwks_keys_for(keys, _ALLOWED_ID_TOKEN_ALGORITHMS):
if kid is None or key.get("kid") == kid:
return jwt.PyJWK(dict(key)) # mutable-ok: PyJWT requires a concrete JWK dictionary
return _BindingRejection(

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

@ -40,6 +40,7 @@ from litellm.proxy._experimental.mcp_server.outbound_credentials.result import (
Ok,
Result,
)
from litellm.proxy.auth.jwt_algorithms import ApprovedJwtAlgorithm
from litellm.types.mcp import (
DEFAULT_CREDENTIAL_HEADER,
DEFAULT_SUBJECT_TOKEN_TYPE,
@ -326,7 +327,7 @@ class PrivateKeyJwtAuth(BaseModel):
source: Literal["private_key_jwt"] = "private_key_jwt"
private_key: SecretStr
key_id: str | None = None
signing_alg: str = "RS256"
signing_alg: ApprovedJwtAlgorithm = "RS256"
class ClientSecretAuth(BaseModel):

44
litellm/proxy/_lazy_openapi_snapshot.json
View file

@ -29465,6 +29465,17 @@
"client_assertion_signing_alg": {
"anyOf": [
{
"enum": [
"RS256",
"RS384",
"RS512",
"PS256",
"PS384",
"PS512",
"ES256",
"ES384",
"ES512"
],
"type": "string"
},
{
@ -32166,6 +32177,17 @@
"client_assertion_signing_alg": {
"anyOf": [
{
"enum": [
"RS256",
"RS384",
"RS512",
"PS256",
"PS384",
"PS512",
"ES256",
"ES384",
"ES512"
],
"type": "string"
},
{
@ -35439,6 +35461,17 @@
"client_assertion_signing_alg": {
"anyOf": [
{
"enum": [
"RS256",
"RS384",
"RS512",
"PS256",
"PS384",
"PS512",
"ES256",
"ES384",
"ES512"
],
"type": "string"
},
{
@ -39289,6 +39322,17 @@
"client_assertion_signing_alg": {
"anyOf": [
{
"enum": [
"RS256",
"RS384",
"RS512",
"PS256",
"PS384",
"PS512",
"ES256",
"ES384",
"ES512"
],
"type": "string"
},
{

83
litellm/proxy/auth/handle_jwt.py
View file

@ -16,6 +16,7 @@ import re
import time
from collections.abc import Awaitable, Callable, Collection, Mapping, Sequence
from dataclasses import dataclass
from functools import lru_cache
from typing import Any, Final, Literal, NoReturn, Protocol, TypeVar, cast
import httpx
@ -53,6 +54,12 @@ from litellm.proxy._types import (
UserAPIKeyAuth,
)
from litellm.proxy.auth.auth_checks import can_team_access_model
from litellm.proxy.auth.jwt_algorithms import (
APPROVED_JWT_ALGORITHMS,
LEGACY_JWT_ALGORITHMS,
allowed_jwt_algorithms,
jwks_keys_for,
)
from litellm.proxy.auth.model_access_denied import (
ModelAccessDeniedHTTPException,
model_access_denied_client_message,
@ -60,6 +67,7 @@ from litellm.proxy.auth.model_access_denied import (
from litellm.proxy.auth.resolvers.grants import GrantResolver, UserLookup, canonical_user_id
from litellm.proxy.auth.route_checks import RouteChecks
from litellm.proxy.auth.team_grants import team_grants, team_model_aliases
from litellm.proxy.common_utils.fips import is_fips_mode
from litellm.proxy.common_utils.user_api_key_cache import (
UserApiKeyCache,
get_management_object_ttl,
@ -104,6 +112,14 @@ UNREACHABLE_CACHE_KEY_PREFIX: Final = "litellm_jwks_unreachable_"
_CachedValueT = TypeVar("_CachedValueT", bound=JWKKeyValue | str)
@lru_cache(maxsize=1)
def _log_eddsa_deprecation() -> None:
verbose_proxy_logger.warning(
"JWT Auth: accepted a token signed with EdDSA, which is deprecated and not FIPS 140-3 approved; "
"it is rejected when LITELLM_FIPS_MODE=true. Move the IdP signing key to RS256, PS256 or ES256"
)
class _JWTAuthSettings(Protocol):
"""The JWT auth settings block this handler reads back through ``getattr``, when one is configured."""
@ -209,18 +225,7 @@ class JWTHandler:
# Supported algos: https://pyjwt.readthedocs.io/en/stable/algorithms.html
# "Warning: Make sure not to mix symmetric and asymmetric algorithms that interpret
# the key in different ways (e.g. HS* and RS*)."
SUPPORTED_JWT_ALGORITHMS = [
"RS256",
"RS384",
"RS512",
"PS256",
"PS384",
"PS512",
"ES256",
"ES384",
"ES512",
"EdDSA",
]
SUPPORTED_JWT_ALGORITHMS = [*APPROVED_JWT_ALGORITHMS, *LEGACY_JWT_ALGORITHMS]
LITELLM_JWT_ISSUER_CLAIM = "_litellm_jwt_issuer"
LITELLM_USER_ID_CLAIM = "_litellm_user_id"
LITELLM_USER_EMAIL_CLAIM = "_litellm_user_email"
@ -240,13 +245,24 @@ class JWTHandler:
def __init__(
self,
fips_mode: Callable[[], bool] = is_fips_mode,
) -> None:
self._fips_mode: Final = fips_mode
self.http_handler = HTTPHandler()
self.leeway = 0
# Per-cache-key locks so a TTL lapse triggers one refresh instead of one per in-flight request.
self._refresh_locks: dict[str, asyncio.Lock] = {} # mutable-ok: lock registry, keyed by JWKS url
self.agent_lookup: AgentLookup = _NoRegisteredAgents()
def allowed_algorithms(self) -> tuple[str, ...]:
return allowed_jwt_algorithms(self._fips_mode())
def _warn_deprecated_signing_algorithm(self, token: str) -> None:
if self._fips_mode():
return
if jwt.get_unverified_header(token).get("alg") == "EdDSA":
_log_eddsa_deprecation()
def bind_agent_lookup(self, agent_lookup: AgentLookup) -> None:
self.agent_lookup = agent_lookup
@ -943,7 +959,12 @@ class JWTHandler:
log_context=f"kid={kid}",
)
public_key: Final = self.parse_keys(keys=keys, kid=kid)
usable_keys: Final[JWKKeyValue] = (
list(jwks_keys_for(keys, self.allowed_algorithms()))
if isinstance(keys, list)
else next(iter(jwks_keys_for((keys,), self.allowed_algorithms())), {})
)
public_key: Final = self.parse_keys(keys=usable_keys, kid=kid)
if public_key is not None:
return cast(dict, public_key)
@ -1216,30 +1237,32 @@ class JWTHandler:
if isinstance(public_key, dict):
public_key_obj: Final = PyJWK.from_dict(self._get_jwk_from_public_key(public_key=public_key)).key
return jwt.decode(
payload: Final = jwt.decode(
token,
public_key_obj,
algorithms=self.SUPPORTED_JWT_ALGORITHMS,
algorithms=self.allowed_algorithms(),
options=decode_options,
audience=audience,
issuer=issuer,
leeway=self.leeway,
)
cert: Final = x509.load_pem_x509_certificate(public_key.encode(), default_backend())
key: Final = cert.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
return jwt.decode(
token,
key,
algorithms=self.SUPPORTED_JWT_ALGORITHMS,
audience=audience,
issuer=issuer,
options=decode_options,
leeway=self.leeway,
)
else:
cert: Final = x509.load_pem_x509_certificate(public_key.encode(), default_backend())
key: Final = cert.public_key().public_bytes(
serialization.Encoding.PEM,
serialization.PublicFormat.SubjectPublicKeyInfo,
)
payload = jwt.decode(
token,
key,
algorithms=self.allowed_algorithms(),
audience=audience,
issuer=issuer,
options=decode_options,
leeway=self.leeway,
)
self._warn_deprecated_signing_algorithm(token)
return payload
async def _auth_jwt_with_issuer(self, token: str, issuer_config: JWTIssuerConfig, kid: str | None) -> dict:
try:

48
litellm/proxy/auth/jwt_algorithms.py Normal file
View file

@ -0,0 +1,48 @@
from collections.abc import Collection, Mapping, Sequence
from types import MappingProxyType
from typing import Final, Literal
ApprovedJwtAlgorithm = Literal["RS256", "RS384", "RS512", "PS256", "PS384", "PS512", "ES256", "ES384", "ES512"]
APPROVED_JWT_ALGORITHMS: Final[tuple[ApprovedJwtAlgorithm, ...]] = (
"RS256",
"RS384",
"RS512",
"PS256",
"PS384",
"PS512",
"ES256",
"ES384",
"ES512",
)
LEGACY_JWT_ALGORITHMS: Final = ("EdDSA",)
_KEY_TYPE_ALGORITHMS: Final = MappingProxyType(
{
"RSA": frozenset(APPROVED_JWT_ALGORITHMS[:6]),
"EC": frozenset(APPROVED_JWT_ALGORITHMS[6:]),
"OKP": frozenset(LEGACY_JWT_ALGORITHMS),
}
)
def allowed_jwt_algorithms(fips_mode: bool) -> tuple[str, ...]:
return APPROVED_JWT_ALGORITHMS if fips_mode else (*APPROVED_JWT_ALGORITHMS, *LEGACY_JWT_ALGORITHMS)
def jwks_keys_for(
keys: Sequence[Mapping[str, object]], algorithms: Collection[str]
) -> tuple[Mapping[str, object], ...]:
"""Keep keys whose declared alg is allowed; keys without alg are kept only when their kty can sign with an allowed algorithm."""
return tuple(key for key in keys if _key_allowed(key, frozenset(algorithms)))
def _key_allowed(key: Mapping[str, object], algorithms: frozenset[str]) -> bool:
alg: Final = key.get("alg")
if isinstance(alg, str):
return alg in algorithms
key_type: Final = key.get("kty")
if not isinstance(key_type, str):
return False
return bool(_KEY_TYPE_ALGORITHMS.get(key_type, frozenset()) & algorithms)

7
litellm/proxy/guardrails/guardrail_hooks/mcp_jwt_signer/mcp_jwt_signer.py
View file

@ -89,6 +89,7 @@ from litellm.integrations.custom_guardrail import (
log_guardrail_information,
)
from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.jwt_algorithms import APPROVED_JWT_ALGORITHMS, jwks_keys_for
from litellm.types.guardrail_base_init import GuardrailBaseInitKwargs
from litellm.types.guardrails import GuardrailEventHooks
from litellm.types.utils import CallTypesLiteral
@ -449,12 +450,16 @@ class MCPJWTSigner(CustomGuardrail):
unverified_header: Final = jwt.get_unverified_header(raw_token)
kid: Final = unverified_header.get("kid")
approved_keys: Final = jwks_keys_for(jwks_keys, APPROVED_JWT_ALGORITHMS)
if not approved_keys:
raise jwt.exceptions.PyJWKSetError(f"No JWKS key at {jwks_uri!r} uses an approved signing algorithm")
# Build a JWKS object and pick the matching key.
# PyJWT's PyJWKSet handles key-type parsing and kid matching correctly.
from jwt import PyJWKSet
try:
jwks_set: Final = PyJWKSet.from_dict({"keys": jwks_keys})
jwks_set: Final = PyJWKSet.from_dict({"keys": list(approved_keys)})
except Exception as exc:
raise jwt.exceptions.PyJWKSetError(f"Failed to parse JWKS from {jwks_uri!r}: {exc}") from exc

3
litellm/types/mcp.py
View file

@ -11,6 +11,7 @@ import httpx
from pydantic import BaseModel, ConfigDict, Field
from typing_extensions import TypedDict
from litellm.proxy.auth.jwt_algorithms import ApprovedJwtAlgorithm
from litellm.types.llms.base import HiddenParams
if TYPE_CHECKING:
@ -251,7 +252,7 @@ class MCPCredentials(TypedDict, total=False):
Key id (kid) advertised in the client_assertion JWT header
"""
client_assertion_signing_alg: str | None
client_assertion_signing_alg: ApprovedJwtAlgorithm | None
"""
Signing algorithm for the client_assertion JWT. Default: RS256
"""

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

@ -4,6 +4,7 @@ from typing import Any, Final, Literal
from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator
from typing_extensions import Self
from litellm.proxy.auth.jwt_algorithms import ApprovedJwtAlgorithm
from litellm.types.mcp import (
DEFAULT_SUBJECT_TOKEN_TYPE,
MCPAuth,
@ -145,7 +146,7 @@ class MCPServer(BaseModel):
id_jag_resource: str | None = None
client_private_key: str | None = None
client_private_key_id: str | None = None
client_assertion_signing_alg: str = "RS256"
client_assertion_signing_alg: ApprovedJwtAlgorithm = "RS256"
# Wire dialect: "rfc8693" (standard token-exchange grant) or "entra_obo" (Microsoft Entra
# On-Behalf-Of, the RFC 7523 jwt-bearer grant + requested_token_use extension)
token_exchange_profile: str = "rfc8693"

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

File diff suppressed because it is too large Load diff

32
tests/test_litellm/proxy/_experimental/mcp_server/test_oauth_identity_binding.py
View file

@ -665,3 +665,35 @@ async def test_audit_matching_login_without_nonce_does_not_report_failure(caplog
)
assert result is None
assert "oauth_identity_binding audit" not in caplog.text
def test_select_signing_key_rejects_kid_matching_key_with_non_approved_alg() -> None:
import json
from cryptography.hazmat.primitives.asymmetric import ed25519
from litellm.proxy._experimental.mcp_server.oauth_identity_binding import _BindingRejection
okp_jwk: Final = {
**json.loads(jwt.algorithms.OKPAlgorithm.to_jwk(ed25519.Ed25519PrivateKey.generate().public_key())),
"kid": KID,
"alg": "EdDSA",
}
result: Final = _select_signing_key(_sign_id_token({}), [okp_jwk])
assert isinstance(result, _BindingRejection)
def test_select_signing_key_picks_approved_key_when_kid_is_shared() -> None:
import json
from cryptography.hazmat.primitives.asymmetric import ed25519
okp_jwk: Final = {
**json.loads(jwt.algorithms.OKPAlgorithm.to_jwk(ed25519.Ed25519PrivateKey.generate().public_key())),
"kid": KID,
"alg": "EdDSA",
}
hs_jwk: Final = {"kty": "oct", "kid": KID, "alg": "HS256", "k": "c2VjcmV0"}
result: Final = _select_signing_key(_sign_id_token({}), [okp_jwk, hs_jwk, _PUBLIC_JWK])
assert isinstance(result, jwt.PyJWK)
assert result.key_id == KID

929
tests/test_litellm/proxy/auth/test_handle_jwt.py
View file

File diff suppressed because it is too large Load diff

151
tests/test_litellm/proxy/guardrails/test_mcp_jwt_signer.py
View file

@ -174,9 +174,7 @@ def test_jwks_public_key_can_verify_signed_jwt():
def test_build_claims_standard_fields():
"""_build_claims() populates iss, aud, iat, exp, nbf."""
signer = _make_signer(
issuer="https://litellm.example.com", audience="mcp", ttl_seconds=300
)
signer = _make_signer(issuer="https://litellm.example.com", audience="mcp", ttl_seconds=300)
user_dict = _make_user_api_key_dict()
data = {"mcp_tool_name": "get_weather"}
@ -347,17 +345,13 @@ async def test_hook_skips_non_mcp_call_types():
data=original_data,
call_type=call_type, # type: ignore[arg-type]
)
assert "extra_headers" not in (
result or {}
), f"extra_headers should not be set for {call_type}"
assert "extra_headers" not in (result or {}), f"extra_headers should not be set for {call_type}"
@pytest.mark.asyncio
async def test_hook_signs_list_mcp_tools():
"""async_pre_call_hook() signs JWT for list_mcp_tools with list scope."""
signer = _make_signer(
issuer="https://litellm.example.com", audience="mcp", ttl_seconds=300
)
signer = _make_signer(issuer="https://litellm.example.com", audience="mcp", ttl_seconds=300)
user_dict = _make_user_api_key_dict(user_id="alice", team_id="backend")
data = {"mcp_tool_name": "should_be_cleared"}
@ -382,9 +376,7 @@ async def test_hook_signs_list_mcp_tools():
@pytest.mark.asyncio
async def test_signed_token_is_verifiable():
"""The JWT injected by the hook can be verified against the JWKS public key."""
signer = _make_signer(
issuer="https://litellm.example.com", audience="mcp", ttl_seconds=300
)
signer = _make_signer(issuer="https://litellm.example.com", audience="mcp", ttl_seconds=300)
user_dict = _make_user_api_key_dict(user_id="alice", team_id="backend")
data = {"mcp_tool_name": "search"}
@ -600,9 +592,7 @@ async def test_channel_token_injected_when_configured():
assert isinstance(result, dict)
assert "x-mcp-channel-token" in result["extra_headers"]
channel_token = result["extra_headers"]["x-mcp-channel-token"].removeprefix(
"Bearer "
)
channel_token = result["extra_headers"]["x-mcp-channel-token"].removeprefix("Bearer ")
channel_payload = _decode_unverified(channel_token)
assert channel_payload["aud"] == "bedrock-gateway"
@ -818,9 +808,7 @@ def test_initialize_guardrail_passes_all_params():
litellm_params.issuer = "https://litellm.example.com"
litellm_params.audience = "mcp-test"
litellm_params.ttl_seconds = 120
litellm_params.access_token_discovery_uri = (
"https://idp.example.com/.well-known/openid-configuration"
)
litellm_params.access_token_discovery_uri = "https://idp.example.com/.well-known/openid-configuration"
litellm_params.token_introspection_endpoint = "https://idp.example.com/introspect"
litellm_params.verify_issuer = "https://idp.example.com"
litellm_params.verify_audience = "api://test"
@ -843,10 +831,7 @@ def test_initialize_guardrail_passes_all_params():
assert signer.issuer == "https://litellm.example.com"
assert signer.audience == "mcp-test"
assert signer.ttl_seconds == 120
assert (
signer.access_token_discovery_uri
== "https://idp.example.com/.well-known/openid-configuration"
)
assert signer.access_token_discovery_uri == "https://idp.example.com/.well-known/openid-configuration"
assert signer.token_introspection_endpoint == "https://idp.example.com/introspect"
assert signer.verify_issuer == "https://idp.example.com"
assert signer.verify_audience == "api://test"
@ -879,9 +864,7 @@ def _make_httpx_response(json_body: dict, status_code: int = 200):
if status_code >= 400:
from httpx import HTTPStatusError, Request, Response
mock_resp.raise_for_status.side_effect = HTTPStatusError(
"error", request=MagicMock(), response=MagicMock()
)
mock_resp.raise_for_status.side_effect = HTTPStatusError("error", request=MagicMock(), response=MagicMock())
return mock_resp
@ -940,9 +923,7 @@ async def test_fetch_jwks_uses_cache_on_second_call():
@pytest.mark.asyncio
async def test_get_oidc_discovery_caches_when_jwks_uri_present():
"""_get_oidc_discovery caches the doc when jwks_uri is in the response."""
signer = _make_signer(
access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration"
)
signer = _make_signer(access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration")
signer._oidc_discovery_doc = None # ensure fresh
discovery_doc = {
@ -964,9 +945,7 @@ async def test_get_oidc_discovery_caches_when_jwks_uri_present():
@pytest.mark.asyncio
async def test_get_oidc_discovery_does_not_cache_when_jwks_uri_absent():
"""_get_oidc_discovery does NOT cache a doc that is missing jwks_uri."""
signer = _make_signer(
access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration"
)
signer = _make_signer(access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration")
signer._oidc_discovery_doc = None
bad_doc = {"issuer": "https://idp.example.com"} # no jwks_uri
@ -1069,9 +1048,7 @@ async def test_verify_incoming_jwt_raises_on_expired_token():
@pytest.mark.asyncio
async def test_introspect_opaque_token_returns_claims_when_active():
"""_introspect_opaque_token returns the introspection payload for active tokens."""
signer = _make_signer(
token_introspection_endpoint="https://idp.example.com/introspect"
)
signer = _make_signer(token_introspection_endpoint="https://idp.example.com/introspect")
introspection_response = {
"active": True,
@ -1095,9 +1072,7 @@ async def test_introspect_opaque_token_returns_claims_when_active():
@pytest.mark.asyncio
async def test_introspect_opaque_token_raises_on_inactive_token():
"""_introspect_opaque_token raises ExpiredSignatureError when active=false."""
signer = _make_signer(
token_introspection_endpoint="https://idp.example.com/introspect"
)
signer = _make_signer(token_introspection_endpoint="https://idp.example.com/introspect")
fake_resp = _make_httpx_response({"active": False})
mock_client = MagicMock()
@ -1128,9 +1103,7 @@ async def test_hook_raises_401_when_jwt_verification_fails():
"""async_pre_call_hook raises HTTP 401 when incoming JWT verification fails."""
from fastapi import HTTPException
signer = _make_signer(
access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration"
)
signer = _make_signer(access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration")
with patch.object(
signer,
@ -1269,3 +1242,101 @@ async def test_inject_mcp_jwt_signs_for_tool_call_path():
scopes = set(decoded["scope"].split())
assert "mcp:tools/call" in scopes
assert "mcp:tools/search_web:call" in scopes
# ---------------------------------------------------------------------------
# B5: incoming-JWT JWKS allowlist (LIT-8429)
# ---------------------------------------------------------------------------
def _okp_idp_key_and_token(now: int):
import json
from cryptography.hazmat.primitives.asymmetric import ed25519
private_key = ed25519.Ed25519PrivateKey.generate()
jwk = {
**json.loads(jwt.algorithms.OKPAlgorithm.to_jwk(private_key.public_key())),
"kid": "ed",
"alg": "EdDSA",
}
token = jwt.encode(
{"sub": "idp-user", "iat": now, "exp": now + 300},
private_key,
algorithm="EdDSA",
headers={"kid": "ed"},
)
return jwk, token
def _oct_idp_key_and_token(now: int):
secret = b"integration-hs256-client-secret-0123456789abcdef"
jwk = {
"kty": "oct",
"kid": "sym",
"alg": "HS256",
"k": base64.urlsafe_b64encode(secret).rstrip(b"=").decode(),
}
token = jwt.encode(
{"sub": "idp-user", "iat": now, "exp": now + 300},
secret,
algorithm="HS256",
headers={"kid": "sym"},
)
return jwk, token
@pytest.mark.asyncio
@pytest.mark.parametrize(
"key_and_token", (_oct_idp_key_and_token, _okp_idp_key_and_token), ids=("oct-HS256", "OKP-EdDSA")
)
async def test_verify_incoming_jwt_rejects_jwks_without_approved_algorithms(key_and_token):
"""A JWKS whose only keys use non-approved algorithms can never verify the incoming token."""
jwks_key, incoming_token = key_and_token(int(time.time()))
signer = _make_signer(
access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration",
)
with patch.object(
signer,
"_get_oidc_discovery",
new_callable=AsyncMock,
return_value={"jwks_uri": "https://idp.example.com/jwks"},
):
with patch(
"litellm.proxy.guardrails.guardrail_hooks.mcp_jwt_signer.mcp_jwt_signer._fetch_jwks",
new_callable=AsyncMock,
return_value=[jwks_key],
):
with pytest.raises(jwt.exceptions.PyJWKSetError, match="approved signing algorithm"):
await signer._verify_incoming_jwt(incoming_token)
@pytest.mark.asyncio
async def test_verify_incoming_jwt_ignores_filtered_keys_sharing_kid():
"""An approved RS256 key still verifies when the JWKS also carries a non-approved key with the same kid."""
signer = _make_signer(
access_token_discovery_uri="https://idp.example.com/.well-known/openid-configuration",
)
now = int(time.time())
oct_jwk, _ = _oct_idp_key_and_token(now)
oct_jwk["kid"] = signer._kid
incoming_token = jwt.encode(
{"sub": "idp-user", "iat": now, "exp": now + 300},
signer._private_key,
algorithm="RS256",
headers={"kid": signer._kid},
)
jwks = [oct_jwk, *signer.get_jwks()["keys"]]
with patch.object(
signer,
"_get_oidc_discovery",
new_callable=AsyncMock,
return_value={"jwks_uri": "https://idp.example.com/jwks"},
):
with patch(
"litellm.proxy.guardrails.guardrail_hooks.mcp_jwt_signer.mcp_jwt_signer._fetch_jwks",
new_callable=AsyncMock,
return_value=jwks,
):
payload = await signer._verify_incoming_jwt(incoming_token)
assert payload["sub"] == "idp-user"

34
tests/test_litellm/types/test_mcp.py
View file

@ -53,12 +53,12 @@ def test_has_header_matches_any_casing() -> None:
@pytest.mark.parametrize(
"target,expected",
[
("https://upstream.example.com/other", False), # same origin
("https://upstream.example.com/other", False), # same origin
("https://upstream.example.com:443/other", False), # explicit default port
("https://attacker.example.com/collect", True), # different host
("http://upstream.example.com/collect", True), # scheme downgrade, same host
("https://attacker.example.com/collect", True), # different host
("http://upstream.example.com/collect", True), # scheme downgrade, same host
("https://upstream.example.com:8443/other", True), # different port, same host
("https://sub.upstream.example.com/x", True), # different host
("https://sub.upstream.example.com/x", True), # different host
],
)
def test_origin_is_scheme_host_and_port_not_host_alone(target: str, expected: bool) -> None:
@ -85,3 +85,29 @@ async def test_the_hook_drops_the_slot_only_once_the_origin_changes() -> None:
foreign = httpx.Request("GET", "https://attacker.example.com/x", headers={"esb-oauth": "Bearer x"})
await hook(foreign)
assert "esb-oauth" not in foreign.headers
def test_new_mcp_server_request_rejects_non_approved_client_assertion_signing_alg() -> None:
from pydantic import ValidationError
from litellm.proxy._types import NewMCPServerRequest
with pytest.raises(ValidationError) as exc:
NewMCPServerRequest(
transport="http",
url="https://mcp.example.com",
credentials={"client_assertion_signing_alg": "HS256"},
)
assert "client_assertion_signing_alg" in str(exc.value)
def test_new_mcp_server_request_accepts_approved_client_assertion_signing_alg() -> None:
from litellm.proxy._types import NewMCPServerRequest
request = NewMCPServerRequest(
transport="http",
url="https://mcp.example.com",
credentials={"client_assertion_signing_alg": "ES256"},
)
assert request.credentials is not None
assert request.credentials["client_assertion_signing_alg"] == "ES256"