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

@@ -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)

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))

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]

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="<EntityDescriptor/>",
+    )
+    assert config.idp_metadata == "<EntityDescriptor/>"
+
+
+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"

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

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

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"

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,))

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"

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",
+    [
+        ("<EntityDescriptor/>", "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)

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

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