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