mirror of
https://github.com/BerriAI/litellm.git
synced 2026-09-12 23:01:41 +00:00
test: cover JWT multi-issuer claim normalization edge cases
This commit is contained in:
parent
bf87f7a8f0
commit
02f4fd1610
1 changed files with 171 additions and 0 deletions
|
|
@ -4417,3 +4417,174 @@ def test_build_decode_kwargs_warns_for_unscoped_global_fallback_in_mixed_deploym
|
|||
if "neither JWT_AUDIENCE nor JWT_ISSUER" in r.getMessage()
|
||||
]
|
||||
assert len(matching) == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_issuer_drops_unmapped_list_scope(monkeypatch):
|
||||
"""A low-trust issuer that maps only an identity field must have its raw
|
||||
``scope`` dropped even when ``scope`` arrives as a list rather than a
|
||||
space-separated string. The list shape must not slip past normalization
|
||||
into authorization."""
|
||||
monkeypatch.delenv("JWT_AUDIENCE", raising=False)
|
||||
monkeypatch.delenv("JWT_PUBLIC_KEY_URL", raising=False)
|
||||
|
||||
issuer = "https://list-scope-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", "openid"]},
|
||||
)
|
||||
|
||||
claims = await jwt_handler.auth_jwt(token=token)
|
||||
|
||||
assert "scope" 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_multi_issuer_drops_nested_unmapped_claim(monkeypatch):
|
||||
"""A nested claim object the issuer does not map (e.g. a Keycloak-style
|
||||
``resource_access`` carrying roles) must be dropped during normalization,
|
||||
not carried through where downstream role resolution could read it."""
|
||||
monkeypatch.delenv("JWT_AUDIENCE", raising=False)
|
||||
monkeypatch.delenv("JWT_PUBLIC_KEY_URL", raising=False)
|
||||
|
||||
issuer = "https://nested-claim-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={
|
||||
"resource_access": {
|
||||
"litellm-proxy": {"roles": ["litellm_proxy_admin"]}
|
||||
}
|
||||
},
|
||||
)
|
||||
|
||||
claims = await jwt_handler.auth_jwt(token=token)
|
||||
|
||||
assert "resource_access" not in claims
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_issuer_normalizes_token_without_optional_standard_claims(
|
||||
monkeypatch,
|
||||
):
|
||||
"""Normalization must not depend on the presence of optional registered
|
||||
claims. A valid token that omits ``exp``/``nbf``/``iat``/``jti`` must
|
||||
normalize without error and still carry the issuer's explicitly mapped
|
||||
identity field."""
|
||||
monkeypatch.delenv("JWT_AUDIENCE", raising=False)
|
||||
monkeypatch.delenv("JWT_PUBLIC_KEY_URL", raising=False)
|
||||
|
||||
import jwt as pyjwt
|
||||
from cryptography.hazmat.primitives import serialization
|
||||
|
||||
issuer = "https://minimal-claims-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_email_jwt_field": "email",
|
||||
}
|
||||
],
|
||||
keys_by_url={jwks_url: [jwk]},
|
||||
)
|
||||
|
||||
private_key_pem = private_key.private_bytes(
|
||||
encoding=serialization.Encoding.PEM,
|
||||
format=serialization.PrivateFormat.PKCS8,
|
||||
encryption_algorithm=serialization.NoEncryption(),
|
||||
)
|
||||
token = pyjwt.encode(
|
||||
{
|
||||
"sub": "test-subject",
|
||||
"iss": issuer,
|
||||
"aud": "expected-audience",
|
||||
"email": "real-user@example.com",
|
||||
},
|
||||
private_key_pem,
|
||||
algorithm="RS256",
|
||||
headers={"kid": "issuer-key"},
|
||||
)
|
||||
|
||||
claims = await jwt_handler.auth_jwt(token=token)
|
||||
|
||||
assert "exp" not in claims
|
||||
assert "iat" not in claims
|
||||
assert jwt_handler.get_user_email(token=claims, default_value=None) == (
|
||||
"real-user@example.com"
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multi_issuer_preserves_explicitly_mapped_role_field(monkeypatch):
|
||||
"""The allowlist is permissive for fields the issuer explicitly maps: when
|
||||
an issuer config maps a scope/role-bearing field to a team claim, that
|
||||
mapped value must survive normalization and resolve to the team."""
|
||||
monkeypatch.delenv("JWT_AUDIENCE", raising=False)
|
||||
monkeypatch.delenv("JWT_PUBLIC_KEY_URL", raising=False)
|
||||
|
||||
issuer = "https://mapped-role-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",
|
||||
"team_ids_jwt_field": "groups",
|
||||
}
|
||||
],
|
||||
keys_by_url={jwks_url: [jwk]},
|
||||
)
|
||||
token = _encode_rsa_jwt(
|
||||
private_key=private_key,
|
||||
issuer=issuer,
|
||||
audience="expected-audience",
|
||||
kid="issuer-key",
|
||||
extra_claims={"groups": ["team-alpha", "team-beta"]},
|
||||
)
|
||||
|
||||
claims = await jwt_handler.auth_jwt(token=token)
|
||||
|
||||
assert claims[JWTHandler.LITELLM_TEAM_IDS_CLAIM] == ["team-alpha", "team-beta"]
|
||||
assert jwt_handler.get_team_ids_from_jwt(token=claims) == [
|
||||
"team-alpha",
|
||||
"team-beta",
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Reference in a new issue