mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
fix: allowlist JWT claims in multi-issuer normalization
Build the normalized token from standard registered claims plus the claims each issuer explicitly maps, instead of passing through all non-internal claims. The single-issuer path is unchanged
This commit is contained in:
parent
556e8f89c8
commit
bf87f7a8f0
2 changed files with 93 additions and 1 deletions
|
|
@ -113,6 +113,10 @@ class JWTHandler:
|
|||
LITELLM_ORG_ID_CLAIM,
|
||||
LITELLM_END_USER_ID_CLAIM,
|
||||
)
|
||||
# Registered JWT claims (RFC 7519) that are safe to carry through issuer
|
||||
# normalization regardless of mapping config: they identify/scope the token
|
||||
# itself and are never trusted for authorization by LiteLLM.
|
||||
STANDARD_JWT_CLAIMS = frozenset(("iss", "sub", "aud", "exp", "nbf", "iat", "jti"))
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
|
|
@ -941,8 +945,11 @@ class JWTHandler:
|
|||
def _apply_issuer_claim_mappings(
|
||||
self, token: dict, issuer_config: JWTIssuerConfig
|
||||
) -> dict:
|
||||
# Build the normalized token from standard registered claims; the
|
||||
# issuer's mapped claims are re-added below. Unmapped claims are not
|
||||
# carried across issuers.
|
||||
normalized: dict = {
|
||||
k: v for k, v in token.items() if k not in self.LITELLM_INTERNAL_CLAIMS
|
||||
k: v for k, v in token.items() if k in self.STANDARD_JWT_CLAIMS
|
||||
}
|
||||
normalized[self.LITELLM_JWT_ISSUER_CLAIM] = issuer_config.issuer
|
||||
claim_mappings = [
|
||||
|
|
|
|||
|
|
@ -4273,6 +4273,91 @@ async def test_multi_issuer_jwt_strips_unmapped_internal_claims(monkeypatch):
|
|||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_issuer_jwt_drops_unmapped_scope_claims(
|
||||
monkeypatch,
|
||||
):
|
||||
"""A token from a configured-but-low-trust issuer that carries a raw
|
||||
``scope`` claim must NOT escalate to proxy admin. The issuer config below
|
||||
maps only an identity field, so ``scope`` is unmapped and dropped during
|
||||
normalization; ``get_scopes``/``is_admin`` then see nothing.
|
||||
"""
|
||||
monkeypatch.delenv("JWT_AUDIENCE", raising=False)
|
||||
monkeypatch.delenv("JWT_PUBLIC_KEY_URL", raising=False)
|
||||
|
||||
issuer = "https://low-trust-issuer.example.com"
|
||||
jwks_url = f"{issuer}/keys"
|
||||
private_key, jwk = _get_rsa_key_and_jwk(kid="issuer-key")
|
||||
jwt_handler = _get_jwt_handler_with_issuer_keys(
|
||||
issuers=[
|
||||
{
|
||||
"issuer": issuer,
|
||||
"jwks_url": jwks_url,
|
||||
"audience": "expected-audience",
|
||||
"user_id_jwt_field": "sub",
|
||||
}
|
||||
],
|
||||
keys_by_url={jwks_url: [jwk]},
|
||||
)
|
||||
token = _encode_rsa_jwt(
|
||||
private_key=private_key,
|
||||
issuer=issuer,
|
||||
audience="expected-audience",
|
||||
kid="issuer-key",
|
||||
extra_claims={
|
||||
"scope": "litellm_proxy_admin",
|
||||
"roles": ["litellm_proxy_admin"],
|
||||
},
|
||||
)
|
||||
|
||||
claims = await jwt_handler.auth_jwt(token=token)
|
||||
|
||||
assert "scope" not in claims
|
||||
assert "roles" not in claims
|
||||
assert jwt_handler.get_scopes(token=claims) == []
|
||||
assert jwt_handler.is_admin(scopes=jwt_handler.get_scopes(token=claims)) is False
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_global_jwt_path_still_trusts_scope_for_admin(monkeypatch):
|
||||
"""The legacy single-issuer / global path (no ``issuers`` configured) must
|
||||
keep honoring the ``scope`` claim as the admin field. The cross-issuer
|
||||
allowlist only applies to issuer-scoped normalization, so this path is
|
||||
unchanged and a token carrying the configured admin scope still maps to
|
||||
proxy admin.
|
||||
"""
|
||||
from litellm.caching.dual_cache import DualCache
|
||||
|
||||
monkeypatch.delenv("JWT_AUDIENCE", raising=False)
|
||||
monkeypatch.delenv("JWT_ISSUER", raising=False)
|
||||
|
||||
jwks_url = "https://global-issuer.example.com/keys"
|
||||
monkeypatch.setenv("JWT_PUBLIC_KEY_URL", jwks_url)
|
||||
|
||||
private_key, jwk = _get_rsa_key_and_jwk(kid="global-key")
|
||||
cache = DualCache()
|
||||
cache.set_cache(key=f"litellm_jwt_auth_keys_{jwks_url}", value=[jwk])
|
||||
|
||||
jwt_handler = JWTHandler()
|
||||
jwt_handler.update_environment(
|
||||
prisma_client=None,
|
||||
user_api_key_cache=cache,
|
||||
litellm_jwtauth=LiteLLM_JWTAuth(),
|
||||
)
|
||||
token = _encode_rsa_jwt(
|
||||
private_key=private_key,
|
||||
issuer="https://global-issuer.example.com",
|
||||
audience="any-audience",
|
||||
kid="global-key",
|
||||
extra_claims={"scope": "litellm_proxy_admin"},
|
||||
)
|
||||
|
||||
claims = await jwt_handler.auth_jwt(token=token)
|
||||
|
||||
assert jwt_handler.get_scopes(token=claims) == ["litellm_proxy_admin"]
|
||||
assert jwt_handler.is_admin(scopes=jwt_handler.get_scopes(token=claims)) is True
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_issuer_jwt_does_not_emit_unscoped_global_warning(
|
||||
monkeypatch, caplog
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue