From 03faccea172a3c0e5b17b76e5e18c6ad68de83db Mon Sep 17 00:00:00 2001 From: Yassin Kortam Date: Wed, 10 Jun 2026 17:59:29 -0700 Subject: [PATCH] test(auth_v2): add test suite for the standards-based auth module Cover every layer of litellm/auth_v2 with tests that fail when the behavior regresses, not just for coverage. Highlights: - authenticators: JwtVerifier enforces signature, aud, iss, exp, required claims, and at+jwt typ via an injected jwks_client (real RS256 against an in-test RSA keypair, no monkeypatching); per-scheme apiKey/http-bearer/ http-basic/oauth2/oidc/mTLS extraction and fail-fast on present-but-invalid - security: OR precedence first-match-wins, a present-but-invalid api key does not fall through to a valid bearer, scope -> 403 insufficient_scope, role -> 403, missing credential -> 401 with WWW-Authenticate, network wired onto the principal - resolver: sha256 api-key lookup (wrong key never resolves), claims-driven principal build (groups -> teams, roles filtered to the Role enum), mTLS -> service account - network: trusted-proxy XFF honored only from a trusted peer, right-to-left parse skips chained proxies, spoofed XFF from an untrusted peer ignored - scim: Users/Groups create/get/patch/list/delete round-trip plus malformed body -> SCIM 400 Error and discovery endpoints - oidc: userinfo -> scim2_models.User mapping and the upsert seam - saml: a real pysaml2 IdP mints a signed assertion; ACS provisions the user, sets a session cookie, and authenticates with method=saml, while tampered and unsigned assertions are rejected (skipped when xmlsec1 is absent) - models/rbac/config: frozen Credential, Role validation, scope/role helpers, SamlConfig metadata validation A mutation spot-check confirmed the suite fails when JWT verification or the api-key hash lookup is broken. --- tests/test_litellm/auth_v2/auth_v2_helpers.py | 90 +++++ tests/test_litellm/auth_v2/conftest.py | 51 +++ .../auth_v2/test_authenticators.py | 338 +++++++++++++++++ tests/test_litellm/auth_v2/test_config.py | 52 +++ tests/test_litellm/auth_v2/test_models.py | 93 +++++ tests/test_litellm/auth_v2/test_network.py | 76 ++++ tests/test_litellm/auth_v2/test_oidc.py | 50 +++ tests/test_litellm/auth_v2/test_rbac.py | 45 +++ tests/test_litellm/auth_v2/test_resolver.py | 119 ++++++ tests/test_litellm/auth_v2/test_saml.py | 356 ++++++++++++++++++ tests/test_litellm/auth_v2/test_scim.py | 146 +++++++ tests/test_litellm/auth_v2/test_security.py | 203 ++++++++++ 12 files changed, 1619 insertions(+) create mode 100644 tests/test_litellm/auth_v2/auth_v2_helpers.py create mode 100644 tests/test_litellm/auth_v2/conftest.py create mode 100644 tests/test_litellm/auth_v2/test_authenticators.py create mode 100644 tests/test_litellm/auth_v2/test_config.py create mode 100644 tests/test_litellm/auth_v2/test_models.py create mode 100644 tests/test_litellm/auth_v2/test_network.py create mode 100644 tests/test_litellm/auth_v2/test_oidc.py create mode 100644 tests/test_litellm/auth_v2/test_rbac.py create mode 100644 tests/test_litellm/auth_v2/test_resolver.py create mode 100644 tests/test_litellm/auth_v2/test_saml.py create mode 100644 tests/test_litellm/auth_v2/test_scim.py create mode 100644 tests/test_litellm/auth_v2/test_security.py diff --git a/tests/test_litellm/auth_v2/auth_v2_helpers.py b/tests/test_litellm/auth_v2/auth_v2_helpers.py new file mode 100644 index 00000000000..d2e503ea93c --- /dev/null +++ b/tests/test_litellm/auth_v2/auth_v2_helpers.py @@ -0,0 +1,90 @@ +from __future__ import annotations + +import time +from typing import Any, Dict, List, Optional, Tuple + +import jwt +from fastapi import Request + +TEST_ISSUER = "https://idp.test.litellm.ai" +TEST_AUDIENCE = "litellm-proxy" + + +class _StaticSigningKey: + def __init__(self, key: Any) -> None: + self.key = key + + +class FakeJwksClient: + """Stands in for PyJWKClient. Returns one fixed key for every token so + JwtVerifier performs a real RS256 signature check against it via PyJWT.""" + + def __init__(self, public_key: Any) -> None: + self._public_key = public_key + self.calls = 0 + + def get_signing_key_from_jwt(self, token: str) -> _StaticSigningKey: + self.calls += 1 + return _StaticSigningKey(self._public_key) + + +class TokenFactory: + def __init__(self, private_pem: bytes) -> None: + self._private_pem = private_pem + + def mint( + self, + *, + issuer: str = TEST_ISSUER, + audience: Any = TEST_AUDIENCE, + subject: str = "user-1", + expires_in: int = 3600, + headers: Optional[Dict[str, Any]] = None, + private_pem: Optional[bytes] = None, + **extra_claims: Any, + ) -> str: + now = int(time.time()) + claims: Dict[str, Any] = { + "iss": issuer, + "aud": audience, + "sub": subject, + "iat": now, + "exp": now + expires_in, + } + claims.update(extra_claims) + return jwt.encode( + claims, + private_pem or self._private_pem, + algorithm="RS256", + headers=headers or {}, + ) + + +def make_request( + *, + headers: Optional[Dict[str, str]] = None, + cookies: Optional[Dict[str, str]] = None, + client: Optional[Tuple[str, int]] = ("203.0.113.7", 5555), + scope_extra: Optional[Dict[str, Any]] = None, +) -> Request: + raw_headers: List[Tuple[bytes, bytes]] = [] + for key, value in (headers or {}).items(): + raw_headers.append((key.lower().encode(), value.encode())) + if cookies: + cookie_header = "; ".join(f"{k}={v}" for k, v in cookies.items()) + raw_headers.append((b"cookie", cookie_header.encode())) + scope: Dict[str, Any] = { + "type": "http", + "http_version": "1.1", + "method": "GET", + "path": "/", + "raw_path": b"/", + "query_string": b"", + "headers": raw_headers, + "client": client, + "server": ("testserver", 80), + "scheme": "http", + } + if scope_extra: + scope.update(scope_extra) + return Request(scope) diff --git a/tests/test_litellm/auth_v2/conftest.py b/tests/test_litellm/auth_v2/conftest.py new file mode 100644 index 00000000000..3a1b8af831c --- /dev/null +++ b/tests/test_litellm/auth_v2/conftest.py @@ -0,0 +1,51 @@ +from __future__ import annotations + +from typing import Any, Tuple + +import pytest +from cryptography.hazmat.primitives import serialization +from cryptography.hazmat.primitives.asymmetric import rsa + +from litellm.auth_v2.authenticators import JwtVerifier +from litellm.auth_v2.config import OidcProviderConfig + +from auth_v2_helpers import TEST_AUDIENCE, TEST_ISSUER, FakeJwksClient, TokenFactory + + +def _generate_keypair() -> Tuple[bytes, Any]: + private_key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + private_pem = private_key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.PKCS8, + serialization.NoEncryption(), + ) + return private_pem, private_key.public_key() + + +@pytest.fixture(scope="session") +def rsa_keypair() -> Tuple[bytes, Any]: + return _generate_keypair() + + +@pytest.fixture(scope="session") +def other_rsa_keypair() -> Tuple[bytes, Any]: + return _generate_keypair() + + +@pytest.fixture +def token_factory(rsa_keypair: Tuple[bytes, Any]) -> TokenFactory: + private_pem, _ = rsa_keypair + return TokenFactory(private_pem) + + +@pytest.fixture +def oidc_provider() -> OidcProviderConfig: + return OidcProviderConfig(issuer=TEST_ISSUER, audience=[TEST_AUDIENCE]) + + +@pytest.fixture +def jwt_verifier( + rsa_keypair: Tuple[bytes, Any], oidc_provider: OidcProviderConfig +) -> JwtVerifier: + _, public_key = rsa_keypair + return JwtVerifier(oidc_provider, jwks_client=FakeJwksClient(public_key)) diff --git a/tests/test_litellm/auth_v2/test_authenticators.py b/tests/test_litellm/auth_v2/test_authenticators.py new file mode 100644 index 00000000000..ab05e50a896 --- /dev/null +++ b/tests/test_litellm/auth_v2/test_authenticators.py @@ -0,0 +1,338 @@ +from __future__ import annotations + +import base64 +from typing import Any + +import pytest + +from litellm.auth_v2.authenticators import ( + ApiKeyAuthenticator, + HttpAuthenticator, + JwtVerifier, + MutualTlsAuthenticator, + OAuth2Authenticator, + OidcAuthenticator, + build_authenticators, +) +from litellm.auth_v2.config import ( + ApiKeySchemeConfig, + AuthConfig, + HttpBasicConfig, + MutualTlsConfig, + OidcProviderConfig, +) +from litellm.auth_v2.errors import AuthError +from litellm.auth_v2.models import AuthMethod, SecuritySchemeType + +from auth_v2_helpers import ( + TEST_AUDIENCE, + TEST_ISSUER, + FakeJwksClient, + make_request, +) + +# --------------------------------------------------------------------------- # +# JwtVerifier: every RFC 7519 check must be enforced. +# --------------------------------------------------------------------------- # + + +def test_jwt_verifier_accepts_valid_token(jwt_verifier, token_factory): + claims = jwt_verifier.verify(token_factory.mint(subject="alice", scope="a b")) + assert claims["sub"] == "alice" + assert claims["aud"] == TEST_AUDIENCE + + +def test_jwt_verifier_rejects_bad_signature( + rsa_keypair, other_rsa_keypair, oidc_provider, token_factory +): + _, public_key = rsa_keypair + verifier = JwtVerifier(oidc_provider, jwks_client=FakeJwksClient(public_key)) + other_pem, _ = other_rsa_keypair + forged = token_factory.mint(private_pem=other_pem) + with pytest.raises(AuthError) as exc: + verifier.verify(forged) + assert exc.value.status_code == 401 + + +def test_jwt_verifier_rejects_wrong_audience(jwt_verifier, token_factory): + with pytest.raises(AuthError) as exc: + jwt_verifier.verify(token_factory.mint(audience="some-other-app")) + assert exc.value.status_code == 401 + + +def test_jwt_verifier_rejects_wrong_issuer(jwt_verifier, token_factory): + with pytest.raises(AuthError) as exc: + jwt_verifier.verify(token_factory.mint(issuer="https://evil.example.com")) + assert exc.value.status_code == 401 + + +def test_jwt_verifier_rejects_expired(jwt_verifier, token_factory): + with pytest.raises(AuthError) as exc: + jwt_verifier.verify(token_factory.mint(expires_in=-30)) + assert exc.value.status_code == 401 + + +def test_jwt_verifier_requires_exp_iss_aud(jwt_verifier, rsa_keypair): + import jwt as pyjwt + + private_pem, _ = rsa_keypair + # token deliberately missing exp/iss/aud + token = pyjwt.encode({"sub": "x"}, private_pem, algorithm="RS256") + with pytest.raises(AuthError) as exc: + jwt_verifier.verify(token) + assert exc.value.status_code == 401 + + +def test_jwt_verifier_enforces_at_jwt_typ(jwt_verifier, token_factory): + without_typ = token_factory.mint() + with pytest.raises(AuthError): + jwt_verifier.verify(without_typ, require_at_jwt=True) + + with_typ = token_factory.mint(headers={"typ": "at+jwt"}) + claims = jwt_verifier.verify(with_typ, require_at_jwt=True) + assert claims["sub"] == "user-1" + + +# --------------------------------------------------------------------------- # +# ApiKeyAuthenticator +# --------------------------------------------------------------------------- # + + +async def test_api_key_authenticator_extracts_header(): + auth = ApiKeyAuthenticator(ApiKeySchemeConfig(header_name="x-litellm-api-key")) + request = make_request(headers={"x-litellm-api-key": "sk-secret-value"}) + credential = await auth.authenticate(request) + assert credential is not None + assert credential.method == AuthMethod.API_KEY + assert credential.subject == "sk-secret-value" + assert credential.claims["_raw_api_key"] == "sk-secret-value" + assert credential.credential_ref.key_id == "sk-secret-" + + +async def test_api_key_authenticator_returns_none_when_absent(): + auth = ApiKeyAuthenticator(ApiKeySchemeConfig()) + assert await auth.authenticate(make_request()) is None + + +# --------------------------------------------------------------------------- # +# HttpAuthenticator (bearer-JWT + basic) +# --------------------------------------------------------------------------- # + + +def _http_auth(public_key: Any, *, basic: HttpBasicConfig = None) -> HttpAuthenticator: + verifier = JwtVerifier( + OidcProviderConfig(issuer=TEST_ISSUER, audience=[TEST_AUDIENCE]), + jwks_client=FakeJwksClient(public_key), + ) + return HttpAuthenticator(basic or HttpBasicConfig(), [verifier]) + + +async def test_http_bearer_valid_token_resolves_credential(rsa_keypair, token_factory): + _, public_key = rsa_keypair + auth = _http_auth(public_key) + token = token_factory.mint(subject="bob", scope="models:read chat:write") + request = make_request(headers={"authorization": f"Bearer {token}"}) + credential = await auth.authenticate(request) + assert credential is not None + assert credential.method == AuthMethod.BEARER_JWT + assert credential.subject == "bob" + assert credential.issuer == TEST_ISSUER + assert credential.audience == [TEST_AUDIENCE] + assert credential.scopes == ["models:read", "chat:write"] + + +async def test_http_bearer_present_but_invalid_fails_fast(rsa_keypair, token_factory): + _, public_key = rsa_keypair + auth = _http_auth(public_key) + # issuer with no configured verifier -> must raise, not return None + token = token_factory.mint(issuer="https://unconfigured.example.com") + request = make_request(headers={"authorization": f"Bearer {token}"}) + with pytest.raises(AuthError) as exc: + await auth.authenticate(request) + assert exc.value.status_code == 401 + + +async def test_http_no_authorization_header_returns_none(rsa_keypair): + _, public_key = rsa_keypair + assert await _http_auth(public_key).authenticate(make_request()) is None + + +async def test_http_basic_disabled_ignores_basic_scheme(rsa_keypair): + _, public_key = rsa_keypair + auth = _http_auth(public_key, basic=HttpBasicConfig(enabled=False)) + creds = base64.b64encode(b"alice:pw").decode() + request = make_request(headers={"authorization": f"Basic {creds}"}) + assert await auth.authenticate(request) is None + + +async def test_http_basic_enabled_decodes_username(rsa_keypair): + _, public_key = rsa_keypair + auth = _http_auth(public_key, basic=HttpBasicConfig(enabled=True)) + creds = base64.b64encode(b"alice:supersecret").decode() + request = make_request(headers={"authorization": f"Basic {creds}"}) + credential = await auth.authenticate(request) + assert credential is not None + assert credential.method == AuthMethod.HTTP_BASIC + assert credential.subject == "alice" + + +async def test_http_basic_malformed_payload_raises(rsa_keypair): + _, public_key = rsa_keypair + auth = _http_auth(public_key, basic=HttpBasicConfig(enabled=True)) + request = make_request(headers={"authorization": "Basic !!!not-base64!!!"}) + with pytest.raises(AuthError) as exc: + await auth.authenticate(request) + assert exc.value.status_code == 401 + + +def test_http_challenge_advertises_basic_only_when_enabled(rsa_keypair): + _, public_key = rsa_keypair + assert "Basic" not in _http_auth(public_key).challenge() + enabled = _http_auth(public_key, basic=HttpBasicConfig(enabled=True)) + assert "Basic" in enabled.challenge() + assert "Bearer" in enabled.challenge() + + +# --------------------------------------------------------------------------- # +# OAuth2Authenticator (at+jwt enforcement + opaque token path) +# --------------------------------------------------------------------------- # + + +def _oauth2(public_key: Any) -> OAuth2Authenticator: + verifier = JwtVerifier( + OidcProviderConfig(issuer=TEST_ISSUER, audience=[TEST_AUDIENCE]), + jwks_client=FakeJwksClient(public_key), + ) + return OAuth2Authenticator([verifier], introspection=None) + + +async def test_oauth2_rejects_jwt_without_at_jwt_typ(rsa_keypair, token_factory): + _, public_key = rsa_keypair + token = token_factory.mint() # no typ header + request = make_request(headers={"authorization": f"Bearer {token}"}) + with pytest.raises(AuthError) as exc: + await _oauth2(public_key).authenticate(request) + assert exc.value.status_code == 401 + + +async def test_oauth2_accepts_at_jwt(rsa_keypair, token_factory): + _, public_key = rsa_keypair + token = token_factory.mint(headers={"typ": "at+jwt"}, subject="svc-1") + request = make_request(headers={"authorization": f"Bearer {token}"}) + credential = await _oauth2(public_key).authenticate(request) + assert credential is not None + assert credential.subject == "svc-1" + + +async def test_oauth2_opaque_token_without_introspection_raises(rsa_keypair): + _, public_key = rsa_keypair + request = make_request(headers={"authorization": "Bearer opaque-not-a-jwt"}) + with pytest.raises(AuthError) as exc: + await _oauth2(public_key).authenticate(request) + assert exc.value.status_code == 401 + + +async def test_oauth2_no_bearer_returns_none(rsa_keypair): + _, public_key = rsa_keypair + assert await _oauth2(public_key).authenticate(make_request()) is None + + +# --------------------------------------------------------------------------- # +# OidcAuthenticator +# --------------------------------------------------------------------------- # + + +def _oidc(public_key: Any) -> OidcAuthenticator: + verifier = JwtVerifier( + OidcProviderConfig(issuer=TEST_ISSUER, audience=[TEST_AUDIENCE]), + jwks_client=FakeJwksClient(public_key), + ) + return OidcAuthenticator([verifier]) + + +async def test_oidc_valid_token_sets_oidc_method(rsa_keypair, token_factory): + _, public_key = rsa_keypair + token = token_factory.mint(subject="carol", email="carol@example.com") + request = make_request(headers={"authorization": f"Bearer {token}"}) + credential = await _oidc(public_key).authenticate(request) + assert credential is not None + assert credential.method == AuthMethod.OIDC + assert credential.subject == "carol" + assert credential.claims["email"] == "carol@example.com" + + +async def test_oidc_unknown_issuer_raises(rsa_keypair, token_factory): + _, public_key = rsa_keypair + token = token_factory.mint(issuer="https://other.example.com") + request = make_request(headers={"authorization": f"Bearer {token}"}) + with pytest.raises(AuthError): + await _oidc(public_key).authenticate(request) + + +# --------------------------------------------------------------------------- # +# MutualTlsAuthenticator +# --------------------------------------------------------------------------- # + + +async def test_mtls_reads_forwarded_subject_header(): + auth = MutualTlsAuthenticator( + MutualTlsConfig(enabled=True, forwarded_subject_header="x-client-dn") + ) + request = make_request(headers={"x-client-dn": "CN=svc-a,O=Co,C=US"}) + credential = await auth.authenticate(request) + assert credential is not None + assert credential.method == AuthMethod.MUTUAL_TLS + assert credential.subject == "CN=svc-a,O=Co,C=US" + assert credential.client_certificate.subject_dn == "CN=svc-a,O=Co,C=US" + + +async def test_mtls_forwarded_header_absent_returns_none(): + auth = MutualTlsAuthenticator( + MutualTlsConfig(enabled=True, forwarded_subject_header="x-client-dn") + ) + assert await auth.authenticate(make_request()) is None + + +async def test_mtls_reads_asgi_tls_extension(): + auth = MutualTlsAuthenticator(MutualTlsConfig(enabled=True)) + request = make_request( + scope_extra={"extensions": {"tls": {"client_cert_name": "CN=from-asgi"}}} + ) + credential = await auth.authenticate(request) + assert credential is not None + assert credential.subject == "CN=from-asgi" + + +async def test_mtls_no_cert_returns_none(): + auth = MutualTlsAuthenticator(MutualTlsConfig(enabled=True)) + assert await auth.authenticate(make_request()) is None + + +# --------------------------------------------------------------------------- # +# build_authenticators: ordering and inclusion follow config +# --------------------------------------------------------------------------- # + + +def test_build_authenticators_follows_scheme_order(): + config = AuthConfig() + authenticators = build_authenticators(config) + schemes = [a.scheme for a in authenticators] + # mutual_tls disabled by default -> excluded + assert schemes == [ + SecuritySchemeType.API_KEY, + SecuritySchemeType.HTTP, + SecuritySchemeType.OPENID_CONNECT, + SecuritySchemeType.OAUTH2, + ] + + +def test_build_authenticators_omits_api_key_when_unconfigured(): + config = AuthConfig(api_key=None) + schemes = [a.scheme for a in build_authenticators(config)] + assert SecuritySchemeType.API_KEY not in schemes + + +def test_build_authenticators_includes_mtls_when_enabled(): + config = AuthConfig(mutual_tls=MutualTlsConfig(enabled=True)) + schemes = [a.scheme for a in build_authenticators(config)] + assert SecuritySchemeType.MUTUAL_TLS == schemes[-1] diff --git a/tests/test_litellm/auth_v2/test_config.py b/tests/test_litellm/auth_v2/test_config.py new file mode 100644 index 00000000000..8050ef3eba0 --- /dev/null +++ b/tests/test_litellm/auth_v2/test_config.py @@ -0,0 +1,52 @@ +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from litellm.auth_v2.config import ( + OAuth2IntrospectionConfig, + OidcProviderConfig, + SamlConfig, +) + + +def test_saml_config_requires_idp_metadata_when_enabled(): + with pytest.raises(ValidationError): + SamlConfig(enabled=True, entity_id="sp", acs_url="https://sp/acs") + + +def test_saml_config_allows_empty_metadata_when_disabled(): + config = SamlConfig(enabled=False, entity_id="sp", acs_url="https://sp/acs") + assert config.idp_metadata == "" + + +def test_saml_config_accepts_inline_metadata(): + config = SamlConfig( + enabled=True, + entity_id="sp", + acs_url="https://sp/acs", + idp_metadata="", + ) + assert config.idp_metadata == "" + + +def test_oidc_provider_requires_audience(): + with pytest.raises(ValidationError): + OidcProviderConfig(issuer="https://idp.example.com") + + +def test_oidc_provider_defaults_to_rs256(): + provider = OidcProviderConfig(issuer="https://idp.example.com", audience=["x"]) + assert provider.algorithms == ["RS256"] + assert provider.require_at_jwt is False + + +def test_introspection_client_secret_is_secret(): + config = OAuth2IntrospectionConfig( + introspection_endpoint="https://idp.example.com/introspect", + client_id="rp", + client_secret="hunter2", + ) + # SecretStr never leaks the value in its repr + assert "hunter2" not in repr(config) + assert config.client_secret.get_secret_value() == "hunter2" diff --git a/tests/test_litellm/auth_v2/test_models.py b/tests/test_litellm/auth_v2/test_models.py new file mode 100644 index 00000000000..f77bf0f3cbc --- /dev/null +++ b/tests/test_litellm/auth_v2/test_models.py @@ -0,0 +1,93 @@ +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from litellm.auth_v2.models import ( + AuthMethod, + Credential, + Principal, + PrincipalType, + SecuritySchemeType, + TeamIdentity, + TeamRole, + UserIdentity, +) +from litellm.auth_v2.rbac import Role + + +def _credential() -> Credential: + return Credential( + scheme=SecuritySchemeType.API_KEY, + method=AuthMethod.API_KEY, + subject="sk-test", + ) + + +def test_credential_is_frozen(): + credential = _credential() + with pytest.raises(ValidationError): + credential.subject = "mutated" + + +def test_credential_defaults_are_independent_instances(): + a = _credential() + b = _credential() + assert a.audience == [] and a.scopes == [] and a.claims == {} + assert a.audience is not b.audience + assert a.claims is not b.claims + + +def test_principal_requires_identity_core_fields(): + with pytest.raises(ValidationError): + Principal(subject="u1") # missing principal_type + auth_method + + +def test_principal_roles_are_validated_against_role_enum(): + principal = Principal( + principal_type=PrincipalType.HUMAN, + subject="u1", + auth_method=AuthMethod.OIDC, + roles=["org_admin"], + ) + assert principal.roles == [Role.ORG_ADMIN] + assert isinstance(principal.roles[0], Role) + + with pytest.raises(ValidationError): + Principal( + principal_type=PrincipalType.HUMAN, + subject="u1", + auth_method=AuthMethod.OIDC, + roles=["not_a_real_role"], + ) + + +def test_principal_default_network_and_collections(): + principal = Principal( + principal_type=PrincipalType.SERVICE_ACCOUNT, + subject="svc", + auth_method=AuthMethod.MUTUAL_TLS, + ) + assert principal.teams == [] + assert principal.scopes == [] + assert principal.network.client_ip is None + assert principal.network.via_trusted_proxy is False + + +def test_team_identity_defaults_to_member_role(): + team = TeamIdentity(id="g1") + assert team.role == TeamRole.MEMBER + + +def test_security_scheme_values_match_openapi_spec(): + assert SecuritySchemeType.API_KEY.value == "apiKey" + assert SecuritySchemeType.HTTP.value == "http" + assert SecuritySchemeType.OAUTH2.value == "oauth2" + assert SecuritySchemeType.OPENID_CONNECT.value == "openIdConnect" + assert SecuritySchemeType.MUTUAL_TLS.value == "mutualTLS" + + +def test_user_identity_optional_fields_default_none(): + user = UserIdentity(id="u1") + assert user.email is None + assert user.external_id is None diff --git a/tests/test_litellm/auth_v2/test_network.py b/tests/test_litellm/auth_v2/test_network.py new file mode 100644 index 00000000000..968130d2d6a --- /dev/null +++ b/tests/test_litellm/auth_v2/test_network.py @@ -0,0 +1,76 @@ +from __future__ import annotations + +from litellm.auth_v2.config import TrustedProxyConfig +from litellm.auth_v2.network import resolve_client_ip, resolve_network_context + +from auth_v2_helpers import make_request + +TRUSTED = TrustedProxyConfig(use_forwarded_for=True, trusted_proxy_cidrs=["10.0.0.0/8"]) + + +def test_xff_ignored_when_forwarding_disabled(): + config = TrustedProxyConfig( + use_forwarded_for=False, trusted_proxy_cidrs=["10.0.0.0/8"] + ) + request = make_request( + headers={"x-forwarded-for": "203.0.113.9"}, client=("10.0.0.1", 1) + ) + ip, via_proxy = resolve_client_ip(request, config) + assert ip == "10.0.0.1" + assert via_proxy is False + + +def test_xff_honored_from_trusted_peer(): + request = make_request( + headers={"x-forwarded-for": "203.0.113.9, 10.0.0.5"}, client=("10.0.0.1", 1) + ) + ip, via_proxy = resolve_client_ip(request, TRUSTED) + assert ip == "203.0.113.9" + assert via_proxy is True + + +def test_spoofed_xff_from_untrusted_peer_is_ignored(): + request = make_request( + headers={"x-forwarded-for": "203.0.113.9"}, client=("8.8.8.8", 1) + ) + ip, via_proxy = resolve_client_ip(request, TRUSTED) + assert ip == "8.8.8.8" + assert via_proxy is False + + +def test_right_to_left_parse_skips_chained_trusted_proxies(): + request = make_request( + headers={"x-forwarded-for": "198.51.100.4, 10.1.1.1, 10.0.0.9"}, + client=("10.0.0.1", 1), + ) + ip, via_proxy = resolve_client_ip(request, TRUSTED) + assert ip == "198.51.100.4" + assert via_proxy is True + + +def test_all_trusted_hops_fall_back_to_peer(): + request = make_request( + headers={"x-forwarded-for": "10.1.1.1, 10.0.0.9"}, client=("10.0.0.1", 1) + ) + ip, via_proxy = resolve_client_ip(request, TRUSTED) + assert ip == "10.0.0.1" + assert via_proxy is True + + +def test_invalid_xff_token_is_skipped(): + request = make_request( + headers={"x-forwarded-for": "not-an-ip, 203.0.113.50"}, client=("10.0.0.1", 1) + ) + ip, _ = resolve_client_ip(request, TRUSTED) + assert ip == "203.0.113.50" + + +def test_network_context_captures_host_and_proxy_flag(): + request = make_request( + headers={"x-forwarded-for": "203.0.113.9", "host": "proxy.litellm.ai"}, + client=("10.0.0.1", 1), + ) + ctx = resolve_network_context(request, TRUSTED) + assert ctx.client_ip == "203.0.113.9" + assert ctx.host == "proxy.litellm.ai" + assert ctx.via_trusted_proxy is True diff --git a/tests/test_litellm/auth_v2/test_oidc.py b/tests/test_litellm/auth_v2/test_oidc.py new file mode 100644 index 00000000000..c2b04c571f2 --- /dev/null +++ b/tests/test_litellm/auth_v2/test_oidc.py @@ -0,0 +1,50 @@ +from __future__ import annotations + +from litellm.auth_v2.config import OidcProviderConfig +from litellm.auth_v2.oidc import _provider_key, _user_from_userinfo +from litellm.auth_v2.resolver import InMemoryIdentityStore + + +def test_userinfo_maps_standard_claims_to_scim_user(): + user = _user_from_userinfo( + { + "sub": "idp-subject-123", + "preferred_username": "dana", + "email": "dana@example.com", + "name": "Dana D", + } + ) + assert user.external_id == "idp-subject-123" + assert user.user_name == "dana" + assert user.display_name == "Dana D" + + +def test_userinfo_falls_back_to_email_when_no_preferred_username(): + user = _user_from_userinfo({"sub": "s1", "email": "eve@example.com"}) + assert user.user_name == "eve@example.com" + + +def test_provider_key_sanitizes_issuer_url(): + key = _provider_key( + OidcProviderConfig(issuer="https://Login.Example.com/realm", audience=["x"]) + ) + assert key == "https-login-example-com-realm" + assert " " not in key + + +async def test_callback_seam_upserts_userinfo_into_store(): + store = InMemoryIdentityStore() + userinfo = { + "sub": "idp-subject-123", + "preferred_username": "dana", + "email": "dana@example.com", + "name": "Dana D", + } + # this is exactly what the OIDC callback does: map userinfo -> SCIM user -> upsert + stored = await store.upsert_user(_user_from_userinfo(userinfo)) + + assert stored.id # store assigned an id + fetched = await store.get_user(stored.id) + assert fetched is not None + assert fetched.external_id == "idp-subject-123" + assert fetched.user_name == "dana" diff --git a/tests/test_litellm/auth_v2/test_rbac.py b/tests/test_litellm/auth_v2/test_rbac.py new file mode 100644 index 00000000000..98232fbdc5c --- /dev/null +++ b/tests/test_litellm/auth_v2/test_rbac.py @@ -0,0 +1,45 @@ +from __future__ import annotations + +from fastapi.security import SecurityScopes + +from litellm.auth_v2.models import AuthMethod, Principal, PrincipalType +from litellm.auth_v2.rbac import Role, has_any_role, has_required_scopes + + +def _principal(*, scopes=None, roles=None) -> Principal: + return Principal( + principal_type=PrincipalType.HUMAN, + subject="u1", + auth_method=AuthMethod.OIDC, + scopes=scopes or [], + roles=roles or [], + ) + + +def test_required_scopes_is_subset_check(): + principal = _principal(scopes=["models:read", "chat:write", "scim:write"]) + assert has_required_scopes(SecurityScopes(["models:read"]), principal) + assert has_required_scopes(SecurityScopes(["models:read", "chat:write"]), principal) + + +def test_missing_required_scope_fails(): + principal = _principal(scopes=["models:read"]) + assert not has_required_scopes(SecurityScopes(["chat:write"]), principal) + + +def test_empty_required_scopes_always_passes(): + assert has_required_scopes(SecurityScopes([]), _principal()) + + +def test_has_any_role_matches_one_of_allowed(): + principal = _principal(roles=[Role.TEAM_MEMBER, Role.ORG_VIEWER]) + assert has_any_role(principal, (Role.ORG_VIEWER, Role.PLATFORM_ADMIN)) + + +def test_has_any_role_rejects_when_no_overlap(): + principal = _principal(roles=[Role.TEAM_MEMBER]) + assert not has_any_role(principal, (Role.PLATFORM_ADMIN, Role.ORG_ADMIN)) + + +def test_has_any_role_false_when_principal_has_no_roles(): + assert not has_any_role(_principal(), (Role.PLATFORM_ADMIN,)) diff --git a/tests/test_litellm/auth_v2/test_resolver.py b/tests/test_litellm/auth_v2/test_resolver.py new file mode 100644 index 00000000000..6526f314819 --- /dev/null +++ b/tests/test_litellm/auth_v2/test_resolver.py @@ -0,0 +1,119 @@ +from __future__ import annotations + +import pytest + +from litellm.auth_v2.errors import AuthError +from litellm.auth_v2.models import ( + AuthMethod, + ClientCertificate, + Credential, + Principal, + PrincipalType, + SecuritySchemeType, +) +from litellm.auth_v2.rbac import Role +from litellm.auth_v2.resolver import InMemoryIdentityStore, _hash_api_key + + +def _api_key_credential(raw: str) -> Credential: + return Credential( + scheme=SecuritySchemeType.API_KEY, + method=AuthMethod.API_KEY, + subject=raw, + claims={"_raw_api_key": raw}, + ) + + +def _principal(subject: str = "user-1") -> Principal: + return Principal( + principal_type=PrincipalType.HUMAN, + subject=subject, + auth_method=AuthMethod.API_KEY, + ) + + +async def test_api_key_resolves_only_for_exact_key(): + raw = "sk-live-correct-horse" + store = InMemoryIdentityStore(api_keys={_hash_api_key(raw): _principal("svc-a")}) + + resolved = await store.resolve(_api_key_credential(raw)) + assert resolved.subject == "svc-a" + + +async def test_wrong_api_key_never_resolves(): + raw = "sk-live-correct-horse" + store = InMemoryIdentityStore(api_keys={_hash_api_key(raw): _principal()}) + with pytest.raises(AuthError) as exc: + await store.resolve(_api_key_credential("sk-live-wrong-key")) + assert exc.value.status_code == 401 + + +async def test_api_key_lookup_is_keyed_on_sha256_not_raw(): + raw = "sk-live-correct-horse" + # store keyed by the raw value (not its hash) must NOT resolve: resolver hashes first + store = InMemoryIdentityStore(api_keys={raw: _principal()}) + with pytest.raises(AuthError): + await store.resolve(_api_key_credential(raw)) + + +async def test_missing_raw_api_key_claim_is_rejected(): + store = InMemoryIdentityStore(api_keys={}) + credential = Credential( + scheme=SecuritySchemeType.API_KEY, + method=AuthMethod.API_KEY, + subject="sk-x", + ) + with pytest.raises(AuthError): + await store.resolve(credential) + + +async def test_subject_lookup_prefers_stored_principal(): + stored = _principal("from-store") + store = InMemoryIdentityStore(subjects={"https://idp|sub-9": stored}) + credential = Credential( + scheme=SecuritySchemeType.OPENID_CONNECT, + method=AuthMethod.OIDC, + subject="sub-9", + issuer="https://idp", + ) + resolved = await store.resolve(credential) + assert resolved.subject == "from-store" + + +async def test_self_describing_token_builds_principal_from_claims(): + store = InMemoryIdentityStore() + credential = Credential( + scheme=SecuritySchemeType.OPENID_CONNECT, + method=AuthMethod.OIDC, + subject="sub-42", + issuer="https://idp", + scopes=["models:read"], + claims={ + "email": "dana@example.com", + "preferred_username": "dana", + "name": "Dana D", + "groups": ["eng", "oncall"], + "roles": ["org_admin", "bogus_role"], + }, + ) + principal = await store.resolve(credential) + assert principal.user.email == "dana@example.com" + assert principal.user.user_name == "dana" + assert [team.id for team in principal.teams] == ["eng", "oncall"] + # invalid role strings are filtered out, valid ones become Role enums + assert principal.roles == [Role.ORG_ADMIN] + assert principal.scopes == ["models:read"] + + +async def test_mtls_credential_resolves_to_service_account(): + store = InMemoryIdentityStore() + credential = Credential( + scheme=SecuritySchemeType.MUTUAL_TLS, + method=AuthMethod.MUTUAL_TLS, + subject="CN=svc-a,O=Co", + client_certificate=ClientCertificate(subject_dn="CN=svc-a,O=Co"), + ) + principal = await store.resolve(credential) + assert principal.principal_type == PrincipalType.SERVICE_ACCOUNT + assert principal.user is None + assert principal.subject == "CN=svc-a,O=Co" diff --git a/tests/test_litellm/auth_v2/test_saml.py b/tests/test_litellm/auth_v2/test_saml.py new file mode 100644 index 00000000000..0c05c2301bd --- /dev/null +++ b/tests/test_litellm/auth_v2/test_saml.py @@ -0,0 +1,356 @@ +from __future__ import annotations + +import base64 +import datetime +import shutil +from dataclasses import dataclass +from pathlib import Path +from typing import Any, Dict, Optional + +import pytest +from fastapi import FastAPI, Security +from fastapi.testclient import TestClient + +xmlsec1 = shutil.which("xmlsec1") +pytestmark = pytest.mark.skipif( + xmlsec1 is None, reason="SAML SP requires the xmlsec1 binary on PATH" +) + +SP_ENTITY_ID = "https://sp.test.litellm.ai/auth/saml/metadata" +ACS_URL = "https://sp.test.litellm.ai/auth/saml/acs" +IDP_ENTITY_ID = "https://idp.test.litellm.ai/idp" +IDP_SSO_URL = "https://idp.test.litellm.ai/sso" + + +def _gen_cert(directory: Path, prefix: str) -> tuple[str, str]: + from cryptography import x509 + from cryptography.hazmat.primitives import hashes, serialization + from cryptography.hazmat.primitives.asymmetric import rsa + from cryptography.x509.oid import NameOID + + key = rsa.generate_private_key(public_exponent=65537, key_size=2048) + subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, prefix)]) + cert = ( + x509.CertificateBuilder() + .subject_name(subject) + .issuer_name(subject) + .public_key(key.public_key()) + .serial_number(x509.random_serial_number()) + .not_valid_before(datetime.datetime(2020, 1, 1)) + .not_valid_after(datetime.datetime(2035, 1, 1)) + .sign(key, hashes.SHA256()) + ) + key_path = directory / f"{prefix}.key" + cert_path = directory / f"{prefix}.crt" + key_path.write_bytes( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + ) + cert_path.write_bytes(cert.public_bytes(serialization.Encoding.PEM)) + return str(key_path), str(cert_path) + + +@dataclass +class SamlEnv: + config: Any + idp: Any # saml2.server.Server + name_id_value: str = "alice@example.com" + + def mint_response( + self, + *, + identity: Optional[Dict[str, Any]] = None, + sign_assertion: bool = True, + ) -> str: + from saml2.authn_context import PASSWORD + from saml2.saml import NAMEID_FORMAT_EMAILADDRESS, NameID + + name_id = NameID(format=NAMEID_FORMAT_EMAILADDRESS, text=self.name_id_value) + response = self.idp.create_authn_response( + identity=identity + or { + "email": ["alice@example.com"], + "displayName": ["Alice Anderson"], + "groups": ["eng", "admins"], + }, + in_response_to=None, + destination=ACS_URL, + sp_entity_id=SP_ENTITY_ID, + name_id=name_id, + sign_assertion=sign_assertion, + authn={"class_ref": PASSWORD, "authn_auth": IDP_ENTITY_ID}, + ) + return base64.b64encode(str(response).encode()).decode() + + +@pytest.fixture +def saml_env(tmp_path: Path) -> SamlEnv: + from saml2 import BINDING_HTTP_POST, BINDING_HTTP_REDIRECT + from saml2.config import IdPConfig, SPConfig + from saml2.metadata import entity_descriptor + from saml2.saml import NAMEID_FORMAT_EMAILADDRESS + from saml2.server import Server + + from litellm.auth_v2.config import SamlConfig + + idp_key, idp_cert = _gen_cert(tmp_path, "idp") + sp_key, sp_cert = _gen_cert(tmp_path, "sp") + + sp_conf = SPConfig() + sp_conf.load( + { + "entityid": SP_ENTITY_ID, + "service": { + "sp": { + "endpoints": { + "assertion_consumer_service": [(ACS_URL, BINDING_HTTP_POST)] + }, + "allow_unsolicited": True, + "authn_requests_signed": False, + "want_assertions_signed": True, + "want_response_signed": False, + } + }, + "allow_unknown_attributes": True, + "xmlsec_binary": xmlsec1, + } + ) + sp_metadata_path = tmp_path / "sp_metadata.xml" + sp_metadata_path.write_text(str(entity_descriptor(sp_conf))) + + idp_conf = IdPConfig() + idp_conf.load( + { + "entityid": IDP_ENTITY_ID, + "service": { + "idp": { + "endpoints": { + "single_sign_on_service": [(IDP_SSO_URL, BINDING_HTTP_REDIRECT)] + }, + "name_id_format": [NAMEID_FORMAT_EMAILADDRESS], + } + }, + "metadata": {"local": [str(sp_metadata_path)]}, + "key_file": idp_key, + "cert_file": idp_cert, + "xmlsec_binary": xmlsec1, + } + ) + idp = Server(config=idp_conf) + idp_metadata = str(entity_descriptor(idp.config)) + + config = SamlConfig( + enabled=True, + entity_id=SP_ENTITY_ID, + acs_url=ACS_URL, + idp_metadata=idp_metadata, + sp_key_file=sp_key, + sp_cert_file=sp_cert, + xmlsec_binary=xmlsec1, + ) + return SamlEnv(config=config, idp=idp) + + +def _build_app(saml_env: SamlEnv): + from litellm.auth_v2.config import AuthConfig + from litellm.auth_v2.models import Principal + from litellm.auth_v2.resolver import InMemoryIdentityStore + from litellm.auth_v2.security import get_current_principal, install_auth + + app = FastAPI() + store = InMemoryIdentityStore() + install_auth( + app, + AuthConfig(saml=saml_env.config), + store, + mount_scim=False, + mount_oidc=False, + mount_saml=True, + ) + + @app.get("/whoami") + async def whoami( + principal: "Principal" = Security(get_current_principal), + ): + return { + "subject": principal.subject, + "auth_method": principal.auth_method.value, + "email": principal.user.email if principal.user else None, + } + + return app, store + + +# --------------------------------------------------------------------------- # +# Metadata + login redirect +# --------------------------------------------------------------------------- # + + +def test_metadata_endpoint_serves_sp_descriptor(saml_env): + app, _ = _build_app(saml_env) + client = TestClient(app) + response = client.get("/auth/saml/metadata") + assert response.status_code == 200 + assert "EntityDescriptor" in response.text + assert SP_ENTITY_ID in response.text + assert ACS_URL in response.text + + +def test_login_redirects_to_idp_sso(saml_env): + app, _ = _build_app(saml_env) + client = TestClient(app) + response = client.get("/auth/saml/login", follow_redirects=False) + assert response.status_code == 303 + assert response.headers["location"].startswith(IDP_SSO_URL) + + +# --------------------------------------------------------------------------- # +# ACS: signed assertion provisions + authenticates; tampering is rejected +# --------------------------------------------------------------------------- # + + +def test_acs_accepts_signed_assertion_and_provisions_user(saml_env): + app, store = _build_app(saml_env) + client = TestClient(app) + saml_response = saml_env.mint_response() + + acs = client.post( + "/auth/saml/acs", + data={"SAMLResponse": saml_response}, + follow_redirects=False, + ) + assert acs.status_code == 303 + assert "saml_session" in acs.cookies + + # user was provisioned into the ProvisioningStore via the shared upsert seam + users = list(store._users.values()) + assert len(users) == 1 + assert users[0].external_id == "alice@example.com" + assert users[0].emails[0].value == "alice@example.com" + + +def test_session_cookie_authenticates_with_saml_method(saml_env): + app, _ = _build_app(saml_env) + client = TestClient(app) + acs = client.post( + "/auth/saml/acs", + data={"SAMLResponse": saml_env.mint_response()}, + follow_redirects=False, + ) + client.cookies.set("saml_session", acs.cookies["saml_session"]) + + whoami = client.get("/whoami") + assert whoami.status_code == 200 + body = whoami.json() + assert body["auth_method"] == "saml" + assert body["subject"] == "alice@example.com" + assert body["email"] == "alice@example.com" + + +def test_acs_rejects_tampered_assertion(saml_env): + app, store = _build_app(saml_env) + client = TestClient(app) + valid = saml_env.mint_response() + decoded = base64.b64decode(valid).decode() + tampered = decoded.replace("alice@example.com", "attacker@evil.com") + tampered_b64 = base64.b64encode(tampered.encode()).decode() + + response = client.post( + "/auth/saml/acs", + data={"SAMLResponse": tampered_b64}, + follow_redirects=False, + ) + assert response.status_code == 401 + assert store._users == {} + + +def test_acs_rejects_unsigned_assertion(saml_env): + app, store = _build_app(saml_env) + client = TestClient(app) + unsigned = saml_env.mint_response(sign_assertion=False) + response = client.post( + "/auth/saml/acs", + data={"SAMLResponse": unsigned}, + follow_redirects=False, + ) + assert response.status_code == 401 + assert store._users == {} + + +def test_acs_missing_response_is_rejected(saml_env): + app, _ = _build_app(saml_env) + client = TestClient(app) + response = client.post("/auth/saml/acs", data={}, follow_redirects=False) + assert response.status_code == 400 + + +# --------------------------------------------------------------------------- # +# Pure helpers (no xmlsec1 required) - attribute mapping + open-redirect guard +# --------------------------------------------------------------------------- # + + +def test_map_attributes_applies_attribute_map(): + from litellm.auth_v2.config import DEFAULT_SAML_ATTRIBUTE_MAP + from litellm.auth_v2.saml import _map_attributes + + ava = { + "email": ["alice@example.com"], + "givenName": ["Alice"], + "surname": ["Anderson"], + "groups": ["eng", "admins"], + } + mapped = _map_attributes(ava, dict(DEFAULT_SAML_ATTRIBUTE_MAP)) + assert mapped["email"] == "alice@example.com" + assert mapped["given_name"] == "Alice" + assert mapped["family_name"] == "Anderson" + assert mapped["groups"] == ["eng", "admins"] + + +def test_user_from_mapped_builds_name_and_email(): + from litellm.auth_v2.saml import _user_from_mapped + + user = _user_from_mapped( + "alice@example.com", + { + "given_name": "Alice", + "family_name": "Anderson", + "email": "alice@example.com", + }, + ) + assert user.external_id == "alice@example.com" + assert user.display_name == "Alice Anderson" + assert user.emails[0].value == "alice@example.com" + assert user.name.given_name == "Alice" + + +@pytest.mark.parametrize( + "candidate,expected", + [ + ("/dashboard", "/dashboard"), + ("//evil.com", "/"), + ("https://evil.com", "/"), + ("/path\\with-backslash", "/"), + (None, "/"), + ], +) +def test_safe_relay_state_blocks_open_redirects(candidate, expected): + from litellm.auth_v2.saml import _safe_relay_state + + assert _safe_relay_state(candidate, "/") == expected + + +@pytest.mark.parametrize( + "metadata,expected_key", + [ + ("", "inline"), + ("https://idp.example.com/metadata", "remote"), + ("/etc/saml/idp.xml", "local"), + ], +) +def test_metadata_source_classifies_input(metadata, expected_key): + from litellm.auth_v2.saml import _metadata_source + + assert expected_key in _metadata_source(metadata) diff --git a/tests/test_litellm/auth_v2/test_scim.py b/tests/test_litellm/auth_v2/test_scim.py new file mode 100644 index 00000000000..d00df336457 --- /dev/null +++ b/tests/test_litellm/auth_v2/test_scim.py @@ -0,0 +1,146 @@ +from __future__ import annotations + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient + +from litellm.auth_v2.config import AuthConfig +from litellm.auth_v2.resolver import InMemoryIdentityStore +from litellm.auth_v2.security import install_auth + +USER_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:User" +GROUP_SCHEMA = "urn:ietf:params:scim:schemas:core:2.0:Group" +ERROR_SCHEMA = "urn:ietf:params:scim:api:messages:2.0:Error" + + +@pytest.fixture +def client() -> TestClient: + app = FastAPI() + install_auth( + app, + AuthConfig(), + InMemoryIdentityStore(), + mount_scim=True, + mount_oidc=False, + mount_saml=False, + ) + return TestClient(app) + + +def _create_user(client: TestClient, user_name="alice@example.com", display="Alice"): + return client.post( + "/scim/v2/Users", + json={"schemas": [USER_SCHEMA], "userName": user_name, "displayName": display}, + ) + + +def test_create_user_returns_201_with_id(client): + response = _create_user(client) + assert response.status_code == 201 + body = response.json() + assert body["id"] + assert body["userName"] == "alice@example.com" + assert USER_SCHEMA in body["schemas"] + + +def test_get_user_round_trips(client): + user_id = _create_user(client).json()["id"] + response = client.get(f"/scim/v2/Users/{user_id}") + assert response.status_code == 200 + assert response.json()["userName"] == "alice@example.com" + + +def test_get_unknown_user_returns_scim_404(client): + response = client.get("/scim/v2/Users/does-not-exist") + assert response.status_code == 404 + assert ERROR_SCHEMA in response.json()["schemas"] + + +def test_patch_replace_display_name(client): + user_id = _create_user(client).json()["id"] + response = client.patch( + f"/scim/v2/Users/{user_id}", + json={ + "schemas": ["urn:ietf:params:scim:api:messages:2.0:PatchOp"], + "Operations": [ + {"op": "replace", "path": "displayName", "value": "Alice B"} + ], + }, + ) + assert response.status_code == 200 + assert response.json()["displayName"] == "Alice B" + # persisted + assert client.get(f"/scim/v2/Users/{user_id}").json()["displayName"] == "Alice B" + + +def test_list_users_returns_list_response(client): + _create_user(client, user_name="a@example.com") + _create_user(client, user_name="b@example.com") + response = client.get("/scim/v2/Users") + assert response.status_code == 200 + body = response.json() + assert body["totalResults"] == 2 + user_names = {r["userName"] for r in body["Resources"]} + assert user_names == {"a@example.com", "b@example.com"} + + +def test_deactivate_user_sets_active_false(client): + user_id = _create_user(client).json()["id"] + assert client.delete(f"/scim/v2/Users/{user_id}").status_code == 204 + assert client.get(f"/scim/v2/Users/{user_id}").json()["active"] is False + + +def test_malformed_user_returns_scim_400_error(client): + # userName is required for a SCIM User creation request + response = client.post( + "/scim/v2/Users", json={"schemas": [USER_SCHEMA], "displayName": "No Username"} + ) + assert response.status_code == 400 + body = response.json() + assert ERROR_SCHEMA in body["schemas"] + assert body["status"] == "400" + + +def test_group_membership_round_trips(client): + response = client.post( + "/scim/v2/Groups", + json={ + "schemas": [GROUP_SCHEMA], + "displayName": "Engineering", + "members": [{"value": "user-1", "display": "Alice"}], + }, + ) + assert response.status_code == 201 + group_id = response.json()["id"] + + fetched = client.get(f"/scim/v2/Groups/{group_id}").json() + assert fetched["displayName"] == "Engineering" + assert fetched["members"][0]["value"] == "user-1" + + +def test_delete_group_removes_it(client): + group_id = client.post( + "/scim/v2/Groups", + json={"schemas": [GROUP_SCHEMA], "displayName": "Temp"}, + ).json()["id"] + assert client.delete(f"/scim/v2/Groups/{group_id}").status_code == 204 + assert client.get(f"/scim/v2/Groups/{group_id}").status_code == 404 + + +def test_service_provider_config_advertises_patch(client): + response = client.get("/scim/v2/ServiceProviderConfig") + assert response.status_code == 200 + assert response.json()["patch"]["supported"] is True + + +def test_resource_types_lists_user_and_group(client): + response = client.get("/scim/v2/ResourceTypes") + assert response.status_code == 200 + names = {r["name"] for r in response.json()["Resources"]} + assert names == {"User", "Group"} + + +def test_schemas_endpoint_returns_user_and_group(client): + response = client.get("/scim/v2/Schemas") + assert response.status_code == 200 + assert response.json()["totalResults"] == 2 diff --git a/tests/test_litellm/auth_v2/test_security.py b/tests/test_litellm/auth_v2/test_security.py new file mode 100644 index 00000000000..f47466f2d29 --- /dev/null +++ b/tests/test_litellm/auth_v2/test_security.py @@ -0,0 +1,203 @@ +from __future__ import annotations + +from typing import Annotated, Any, Tuple + +import pytest +from fastapi import FastAPI, Security +from fastapi.testclient import TestClient + +from litellm.auth_v2.authenticators import ( + ApiKeyAuthenticator, + HttpAuthenticator, + JwtVerifier, +) +from litellm.auth_v2.config import ( + ApiKeySchemeConfig, + AuthConfig, + HttpBasicConfig, + OidcProviderConfig, +) +from litellm.auth_v2.models import AuthMethod, Principal, PrincipalType +from litellm.auth_v2.rbac import Role +from litellm.auth_v2.resolver import InMemoryIdentityStore, _hash_api_key +from litellm.auth_v2.security import ( + AuthContext, + get_current_principal, + require_roles, +) + +from auth_v2_helpers import TEST_AUDIENCE, TEST_ISSUER, FakeJwksClient + +ADMIN_KEY = "sk-admin-key" +READER_KEY = "sk-reader-key" +NOSCOPE_KEY = "sk-noscope-key" + + +def _principal(subject: str, *, scopes=None, roles=None) -> Principal: + return Principal( + principal_type=PrincipalType.HUMAN, + subject=subject, + auth_method=AuthMethod.API_KEY, + scopes=scopes or [], + roles=roles or [], + ) + + +def _build_app(public_key: Any) -> Tuple[FastAPI, InMemoryIdentityStore]: + verifier = JwtVerifier( + OidcProviderConfig(issuer=TEST_ISSUER, audience=[TEST_AUDIENCE]), + jwks_client=FakeJwksClient(public_key), + ) + authenticators = [ + ApiKeyAuthenticator(ApiKeySchemeConfig()), + HttpAuthenticator(HttpBasicConfig(), [verifier]), + ] + resolver = InMemoryIdentityStore( + api_keys={ + _hash_api_key(ADMIN_KEY): _principal( + "admin-principal", scopes=["models:read"], roles=[Role.ORG_ADMIN] + ), + _hash_api_key(READER_KEY): _principal( + "reader-principal", scopes=["models:read"] + ), + _hash_api_key(NOSCOPE_KEY): _principal("noscope-principal"), + } + ) + ctx = AuthContext(AuthConfig(), authenticators, resolver) + + app = FastAPI() + app.state.auth_v2 = ctx + + @app.get("/open") + async def open_route( + principal: Annotated[Principal, Security(get_current_principal)], + ): + return { + "subject": principal.subject, + "auth_method": principal.auth_method.value, + "network_host": principal.network.host, + } + + @app.get("/scoped") + async def scoped_route( + principal: Annotated[ + Principal, Security(get_current_principal, scopes=["models:read"]) + ], + ): + return {"subject": principal.subject} + + @app.get("/admin") + async def admin_route( + principal: Annotated[Principal, Security(require_roles(Role.ORG_ADMIN))], + ): + return {"subject": principal.subject} + + return app, resolver + + +@pytest.fixture +def client(rsa_keypair) -> TestClient: + _, public_key = rsa_keypair + app, _ = _build_app(public_key) + return TestClient(app) + + +def _bearer(token_factory, **claims) -> dict: + return {"Authorization": f"Bearer {token_factory.mint(**claims)}"} + + +# --------------------------------------------------------------------------- # +# Missing credential +# --------------------------------------------------------------------------- # + + +def test_no_credential_returns_401_with_challenge(client): + response = client.get("/open") + assert response.status_code == 401 + assert "WWW-Authenticate" in response.headers + assert "Bearer" in response.headers["WWW-Authenticate"] + + +# --------------------------------------------------------------------------- # +# Single-scheme success + network wiring +# --------------------------------------------------------------------------- # + + +def test_valid_api_key_authenticates(client): + response = client.get("/open", headers={"x-litellm-api-key": READER_KEY}) + assert response.status_code == 200 + body = response.json() + assert body["subject"] == "reader-principal" + assert body["network_host"] # resolve_network_context wired into principal + + +def test_valid_bearer_authenticates_from_claims(client, token_factory): + response = client.get( + "/open", headers=_bearer(token_factory, subject="jwt-user", scope="models:read") + ) + assert response.status_code == 200 + assert response.json()["subject"] == "jwt-user" + + +# --------------------------------------------------------------------------- # +# OR precedence: first match wins, present-but-invalid fails fast +# --------------------------------------------------------------------------- # + + +def test_first_match_wins_api_key_before_bearer(client, token_factory): + response = client.get( + "/open", + headers={ + "x-litellm-api-key": READER_KEY, + "Authorization": f"Bearer {token_factory.mint(subject='jwt-user')}", + }, + ) + assert response.status_code == 200 + # api key is earlier in scheme_order, so it resolves; bearer is never consulted + assert response.json()["subject"] == "reader-principal" + + +def test_present_but_invalid_api_key_does_not_fall_through_to_bearer( + client, token_factory +): + response = client.get( + "/open", + headers={ + "x-litellm-api-key": "sk-totally-unknown", + "Authorization": f"Bearer {token_factory.mint(subject='jwt-user', scope='models:read')}", + }, + ) + assert response.status_code == 401 + assert "jwt-user" not in response.text + + +# --------------------------------------------------------------------------- # +# Scope enforcement (RFC 6750: 403 insufficient_scope) +# --------------------------------------------------------------------------- # + + +def test_scope_satisfied_returns_200(client): + response = client.get("/scoped", headers={"x-litellm-api-key": READER_KEY}) + assert response.status_code == 200 + + +def test_missing_scope_returns_403_insufficient_scope(client): + response = client.get("/scoped", headers={"x-litellm-api-key": NOSCOPE_KEY}) + assert response.status_code == 403 + assert "insufficient_scope" in response.headers.get("WWW-Authenticate", "") + + +# --------------------------------------------------------------------------- # +# Role enforcement +# --------------------------------------------------------------------------- # + + +def test_required_role_present_returns_200(client): + response = client.get("/admin", headers={"x-litellm-api-key": ADMIN_KEY}) + assert response.status_code == 200 + assert response.json()["subject"] == "admin-principal" + + +def test_required_role_missing_returns_403(client): + response = client.get("/admin", headers={"x-litellm-api-key": READER_KEY}) + assert response.status_code == 403