refactor: authv2

backend/auth/routers/__init__.py

@@ -0,0 +1,5 @@
+from .oidc import router as oidc_router
+from .saml import router as saml_router
+from .scim import router as scim_router
+
+__all__ = ["oidc_router", "saml_router", "scim_router"]

backend/auth/routers/dependencies.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Tuple, cast
+
+from authlib.integrations.starlette_client import OAuth
+from fastapi import Request
+from fastapi.security import SecurityScopes
+from saml2.client import Saml2Client
+
+from litellm.proxy.auth_v2.models import Principal
+from litellm.proxy.auth_v2.resolvers import ProvisioningStore
+from litellm.proxy.auth_v2.security import AuthSecurity
+
+from ..services.oidc import build_oauth_registry
+from ..services.saml import SAMLProtocolStore, build_sp_client
+
+
+def get_auth(request: Request) -> AuthSecurity:
+    return request.app.state.auth_v2
+
+
+def get_oauth_registry(request: Request) -> OAuth:
+    cached = getattr(request.app.state, "oidc_oauth", None)
+    if cached is None:
+        cached = build_oauth_registry(get_auth(request).config.oidc_providers)
+        request.app.state.oidc_oauth = cached
+    return cached
+
+
+def get_saml_runtime(request: Request) -> Tuple[Saml2Client, SAMLProtocolStore]:
+    state = request.app.state
+    client = getattr(state, "saml_client", None)
+    if client is None:
+        auth = get_auth(request)
+        config = auth.config.saml
+        assert config is not None
+        client = build_sp_client(config)
+        state.saml_client = client
+        state.saml_protocol = SAMLProtocolStore(auth.config.session.ttl_seconds)
+    return client, state.saml_protocol
+
+
+async def scim_principal(request: Request) -> Principal:
+    auth = get_auth(request)
+    return await auth.principal(SecurityScopes(scopes=["scim:write"]), request)
+
+
+def scim_store(request: Request) -> ProvisioningStore:
+    return cast(ProvisioningStore, get_auth(request).resolver)

backend/auth/routers/oidc.py

@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import secrets
+from typing import cast
+
+from authlib.integrations.starlette_client import OAuth
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import RedirectResponse
+
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.authenticators import apply_role_policy
+from litellm.proxy.auth_v2.models import AuthMethod
+from ..services.redirects import safe_relay_state
+from litellm.proxy.auth_v2.resolvers import ProvisioningStore
+from litellm.proxy.auth_v2.security import AuthSecurity
+from litellm.proxy.auth_v2.sessions.schemas import OAuthTransaction, SessionState
+
+from ..services.oidc import mapped_claims, providers_by_key, user_from_userinfo
+from .dependencies import get_auth, get_oauth_registry
+
+router = APIRouter(prefix="/auth/oidc", tags=["oidc"])
+
+
+@router.get("/{provider}/login")
+async def login(
+    provider: str,
+    request: Request,
+    auth: AuthSecurity = Depends(get_auth),
+    oauth: OAuth = Depends(get_oauth_registry),
+) -> RedirectResponse:
+    session = auth.config.session
+    client = oauth.create_client(provider)
+    if client is None:
+        raise errors.unknown_provider()
+    redirect_uri = str(request.url_for("oidc_callback", provider=provider))
+    relay = safe_relay_state(request.query_params.get("next"), session.default_redirect_path)
+    authorization = await client.create_authorization_url(redirect_uri)
+    txn_id = secrets.token_urlsafe(32)
+    await auth.oauth_txn_store.set(
+        txn_id,
+        OAuthTransaction(
+            provider=provider,
+            state=authorization["state"],
+            redirect_uri=redirect_uri,
+            relay=relay,
+            nonce=authorization.get("nonce"),
+            code_verifier=authorization.get("code_verifier"),
+        ),
+    )
+    response = RedirectResponse(authorization["url"], status_code=303)
+    response.set_cookie(
+        session.login_cookie,
+        txn_id,
+        httponly=True,
+        samesite="lax",
+        secure=session.secure,
+        max_age=session.login_state_ttl,
+    )
+    return response
+
+
+@router.get("/{provider}/callback", name="oidc_callback")
+async def callback(
+    provider: str,
+    request: Request,
+    auth: AuthSecurity = Depends(get_auth),
+    oauth: OAuth = Depends(get_oauth_registry),
+) -> RedirectResponse:
+    session = auth.config.session
+    client = oauth.create_client(provider)
+    if client is None:
+        raise errors.unknown_provider()
+    txn_id = request.cookies.get(session.login_cookie)
+    txn = await auth.oauth_txn_store.pop(txn_id) if txn_id else None
+    if txn is None or txn["provider"] != provider:
+        raise errors.invalid_login_state()
+    returned_state = request.query_params.get("state")
+    if not returned_state or returned_state != txn["state"]:
+        raise errors.state_mismatch()
+    error = request.query_params.get("error")
+    if error:
+        raise errors.oidc_provider_error(error)
+    code = request.query_params.get("code")
+    if not code:
+        raise errors.missing_authorization_code()
+
+    token = await client.fetch_access_token(
+        redirect_uri=txn["redirect_uri"],
+        code=code,
+        code_verifier=txn["code_verifier"],
+        state=txn["state"],
+    )
+    if token.get("id_token"):
+        userinfo = await client.parse_id_token(token, nonce=txn["nonce"])
+    else:
+        userinfo = await client.userinfo(token=token)
+
+    info = dict(userinfo)
+    provider_config = providers_by_key(auth.config.oidc_providers)[provider]
+
+    store = cast(ProvisioningStore, auth.resolver)
+    await store.upsert_user(user_from_userinfo(info))
+
+    claims = mapped_claims(info)
+    apply_role_policy(claims, provider_config)
+    session_id = secrets.token_urlsafe(32)
+    await auth.session_store.set(
+        session_id,
+        SessionState(
+            method=AuthMethod.OIDC.value,
+            subject=info.get("sub", ""),
+            issuer=info.get("iss") or provider_config.issuer,
+            claims=claims,
+        ),
+    )
+    target = safe_relay_state(txn["relay"], session.default_redirect_path)
+    response = RedirectResponse(target, status_code=303)
+    response.set_cookie(
+        session.cookie,
+        session_id,
+        httponly=True,
+        samesite="lax",
+        secure=session.secure,
+    )
+    return response

backend/auth/routers/saml.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import secrets
+from typing import Tuple, cast
+
+from fastapi import APIRouter, Depends, Request
+from fastapi.responses import RedirectResponse, Response
+from saml2 import BINDING_HTTP_POST
+from saml2.client import Saml2Client
+from saml2.metadata import entity_descriptor
+
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.models import AuthMethod
+from litellm.proxy.auth_v2.authorization import filter_claim_roles
+from ..services.redirects import safe_relay_state
+from litellm.proxy.auth_v2.resolvers import ProvisioningStore
+from litellm.proxy.auth_v2.security import AuthSecurity
+from litellm.proxy.auth_v2.sessions.schemas import SessionState
+
+from ..services.saml import (
+    SAMLProtocolStore,
+    claims_from_mapped,
+    map_attributes,
+    user_from_mapped,
+)
+from .dependencies import get_auth, get_saml_runtime
+
+router = APIRouter(prefix="/auth/saml", tags=["saml"])
+
+
+@router.get("/metadata")
+async def metadata(
+    runtime: Tuple[Saml2Client, SAMLProtocolStore] = Depends(get_saml_runtime),
+) -> Response:
+    client, _ = runtime
+    return Response(
+        content=str(entity_descriptor(client.config)),
+        media_type="application/samlmetadata+xml",
+    )
+
+
+@router.get("/login")
+async def login(
+    request: Request,
+    auth: AuthSecurity = Depends(get_auth),
+    runtime: Tuple[Saml2Client, SAMLProtocolStore] = Depends(get_saml_runtime),
+) -> RedirectResponse:
+    session = auth.config.session
+    client, protocol = runtime
+    relay_state = safe_relay_state(request.query_params.get("next"), session.default_redirect_path)
+    request_id, info = client.prepare_for_authenticate(relay_state=relay_state)
+    protocol.remember_request(request_id, relay_state)
+    location = dict(info["headers"]).get("Location")
+    if not location:
+        raise errors.saml_redirect_failed()
+    return RedirectResponse(location, status_code=303)
+
+
+@router.post("/acs")
+async def assertion_consumer_service(
+    request: Request,
+    auth: AuthSecurity = Depends(get_auth),
+    runtime: Tuple[Saml2Client, SAMLProtocolStore] = Depends(get_saml_runtime),
+) -> Response:
+    config = auth.config.saml
+    assert config is not None
+    session = auth.config.session
+    client, protocol = runtime
+
+    form = await request.form()
+    saml_response = form.get("SAMLResponse")
+    if not isinstance(saml_response, str):
+        raise errors.missing_saml_response()
+    try:
+        authn_response = client.parse_authn_request_response(
+            saml_response,
+            BINDING_HTTP_POST,
+            outstanding=protocol.outstanding_relays() or None,
+        )
+    except Exception as exc:
+        raise errors.invalid_saml_response() from exc
+    if authn_response is None:
+        raise errors.invalid_saml_response()
+
+    in_response_to = getattr(authn_response, "in_response_to", None)
+    bound_relay = protocol.consume_request(in_response_to) if in_response_to else None
+
+    assertion = getattr(authn_response, "assertion", None)
+    assertion_id = getattr(assertion, "id", None)
+    if assertion_id and not protocol.consume_assertion(assertion_id):
+        raise errors.saml_assertion_replay()
+
+    name_id = authn_response.get_subject().text
+    ava = authn_response.get_identity() or {}
+    mapped = map_attributes(ava, config.attribute_map)
+    mapped["roles"] = filter_claim_roles(mapped.get("roles"), config.allowed_roles, config.allow_platform_roles)
+    user = user_from_mapped(name_id, mapped)
+
+    store = cast(ProvisioningStore, auth.resolver)
+    await store.upsert_user(user)
+
+    session_id = secrets.token_urlsafe(32)
+    await auth.session_store.set(
+        session_id,
+        SessionState(
+            method=AuthMethod.SAML.value,
+            subject=name_id,
+            issuer=authn_response.issuer(),
+            claims=claims_from_mapped(mapped),
+        ),
+    )
+    target = safe_relay_state(bound_relay, session.default_redirect_path)
+    response = RedirectResponse(target, status_code=303)
+    response.set_cookie(
+        session.cookie,
+        session_id,
+        httponly=True,
+        samesite="lax",
+        secure=session.secure,
+    )
+    return response

backend/auth/routers/scim.py

@@ -0,0 +1,143 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import APIRouter, Depends, Query, Request, Response, status
+from fastapi.responses import JSONResponse
+from pydantic import ValidationError
+from scim2_models import Group, User
+
+from ..services import scim
+from .dependencies import scim_principal, scim_store
+
+router = APIRouter(prefix="/scim/v2", tags=["scim"], route_class=scim.ScimErrorRoute)
+_protected = [Depends(scim_principal)]
+
+
+@router.post("/Users", status_code=status.HTTP_201_CREATED, dependencies=_protected)
+async def create_user(request: Request) -> Response:
+    try:
+        user = scim.parse_resource(await request.json(), User)
+    except ValidationError as exc:
+        return scim.scim_error(status.HTTP_400_BAD_REQUEST, str(exc))
+    stored = await scim_store(request).upsert_user(user)
+    return JSONResponse(
+        status_code=status.HTTP_201_CREATED,
+        content=scim.creation_response(stored),
+    )
+
+
+@router.get("/Users/{resource_id}", dependencies=_protected)
+async def get_user(resource_id: str, request: Request) -> Response:
+    user = await scim_store(request).get_user(resource_id)
+    if user is None:
+        return scim.scim_error(status.HTTP_404_NOT_FOUND, f"User {resource_id} not found")
+    return JSONResponse(content=scim.query_response(user))
+
+
+@router.patch("/Users/{resource_id}", dependencies=_protected)
+async def patch_user(resource_id: str, request: Request) -> Response:
+    store = scim_store(request)
+    user = await store.get_user(resource_id)
+    if user is None:
+        return scim.scim_error(status.HTTP_404_NOT_FOUND, f"User {resource_id} not found")
+    try:
+        patched = scim.apply_patch(user, await request.json())
+    except (ValidationError, ValueError) as exc:
+        return scim.scim_error(status.HTTP_400_BAD_REQUEST, str(exc))
+    updated = await store.upsert_user(patched)
+    return JSONResponse(content=scim.patch_response(updated))
+
+
+@router.delete(
+    "/Users/{resource_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    dependencies=_protected,
+)
+async def deactivate_user(resource_id: str, request: Request) -> Response:
+    store = scim_store(request)
+    if await store.get_user(resource_id) is None:
+        return scim.scim_error(status.HTTP_404_NOT_FOUND, f"User {resource_id} not found")
+    await store.deactivate_user(resource_id)
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.get("/Users", dependencies=_protected)
+async def list_users(
+    request: Request,
+    filter_expr: Optional[str] = Query(default=None, alias="filter"),
+) -> Response:
+    users = await scim_store(request).list_users(filter_expr)
+    return JSONResponse(content=scim.list_response(User, users))
+
+
+@router.post("/Groups", status_code=status.HTTP_201_CREATED, dependencies=_protected)
+async def create_group(request: Request) -> Response:
+    try:
+        group = scim.parse_resource(await request.json(), Group)
+    except ValidationError as exc:
+        return scim.scim_error(status.HTTP_400_BAD_REQUEST, str(exc))
+    stored = await scim_store(request).upsert_group(group)
+    return JSONResponse(
+        status_code=status.HTTP_201_CREATED,
+        content=scim.creation_response(stored),
+    )
+
+
+@router.get("/Groups/{resource_id}", dependencies=_protected)
+async def get_group(resource_id: str, request: Request) -> Response:
+    group = await scim_store(request).get_group(resource_id)
+    if group is None:
+        return scim.scim_error(status.HTTP_404_NOT_FOUND, f"Group {resource_id} not found")
+    return JSONResponse(content=scim.query_response(group))
+
+
+@router.patch("/Groups/{resource_id}", dependencies=_protected)
+async def patch_group(resource_id: str, request: Request) -> Response:
+    store = scim_store(request)
+    group = await store.get_group(resource_id)
+    if group is None:
+        return scim.scim_error(status.HTTP_404_NOT_FOUND, f"Group {resource_id} not found")
+    try:
+        patched = scim.apply_patch(group, await request.json())
+    except (ValidationError, ValueError) as exc:
+        return scim.scim_error(status.HTTP_400_BAD_REQUEST, str(exc))
+    updated = await store.upsert_group(patched)
+    return JSONResponse(content=scim.patch_response(updated))
+
+
+@router.delete(
+    "/Groups/{resource_id}",
+    status_code=status.HTTP_204_NO_CONTENT,
+    dependencies=_protected,
+)
+async def delete_group(resource_id: str, request: Request) -> Response:
+    store = scim_store(request)
+    if await store.get_group(resource_id) is None:
+        return scim.scim_error(status.HTTP_404_NOT_FOUND, f"Group {resource_id} not found")
+    await store.delete_group(resource_id)
+    return Response(status_code=status.HTTP_204_NO_CONTENT)
+
+
+@router.get("/Groups", dependencies=_protected)
+async def list_groups(
+    request: Request,
+    filter_expr: Optional[str] = Query(default=None, alias="filter"),
+) -> Response:
+    groups = await scim_store(request).list_groups(filter_expr)
+    return JSONResponse(content=scim.list_response(Group, groups))
+
+
+@router.get("/ServiceProviderConfig")
+async def service_provider_config() -> Response:
+    return JSONResponse(content=scim.service_provider_config())
+
+
+@router.get("/ResourceTypes")
+async def resource_types() -> Response:
+    return JSONResponse(content=scim.resource_types())
+
+
+@router.get("/Schemas")
+async def schemas() -> Response:
+    return JSONResponse(content=scim.schemas())

backend/auth/services/oidc.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+import re
+from typing import Any, Dict, List
+
+from authlib.integrations.starlette_client import OAuth
+from scim2_models import User as ScimUser
+
+from litellm.proxy.auth_v2.config import OIDCProviderConfig
+
+CLAIM_KEYS = ("email", "preferred_username", "name", "groups", "roles")
+
+
+def provider_key(provider: OIDCProviderConfig) -> str:
+    return re.sub(r"[^a-z0-9]+", "-", provider.issuer.lower()).strip("-")
+
+
+def providers_by_key(
+    providers: List[OIDCProviderConfig],
+) -> Dict[str, OIDCProviderConfig]:
+    return {provider_key(provider): provider for provider in providers}
+
+
+def user_from_userinfo(userinfo: Dict[str, Any]) -> ScimUser:
+    return ScimUser(
+        external_id=userinfo.get("sub"),
+        user_name=userinfo.get("preferred_username") or userinfo.get("email"),
+        display_name=userinfo.get("name"),
+    )
+
+
+def mapped_claims(userinfo: Dict[str, Any]) -> Dict[str, Any]:
+    return {key: userinfo[key] for key in CLAIM_KEYS if userinfo.get(key) is not None}
+
+
+def build_oauth_registry(providers: List[OIDCProviderConfig]) -> OAuth:
+    oauth = OAuth()
+    for provider in providers:
+        oauth.register(
+            name=provider_key(provider),
+            server_metadata_url=f"{provider.issuer.rstrip('/')}/.well-known/openid-configuration",
+            client_id=provider.client_id,
+            client_secret=(provider.client_secret.get_secret_value() if provider.client_secret else None),
+            client_kwargs={
+                "scope": " ".join(provider.login_scopes),
+                "code_challenge_method": "S256",
+            },
+        )
+    return oauth

backend/auth/services/redirects.py

@@ -0,0 +1,15 @@
+from __future__ import annotations
+
+from typing import Optional
+
+
+def safe_relay_state(target: Optional[str], default: str) -> str:
+    """Return ``target`` only if it's a safe same-site path, else ``default``.
+
+    Guards the post-login redirect against open-redirect: the target must be a
+    relative path (single leading slash, no scheme, no protocol-relative ``//``
+    or backslash tricks).
+    """
+    if target and target.startswith("/") and not target.startswith("//") and "://" not in target and "\\" not in target:
+        return target
+    return default

backend/auth/services/saml.py

@@ -1,24 +1,15 @@
 from __future__ import annotations
 
 import time
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, cast
+from typing import Any, Dict, List, Optional, Tuple
 
-from fastapi import APIRouter, HTTPException, Request
-from fastapi.responses import RedirectResponse, Response
 from saml2 import BINDING_HTTP_POST
 from saml2.client import Saml2Client
 from saml2.config import SPConfig
-from saml2.metadata import entity_descriptor
 from scim2_models import Email, Name
 from scim2_models import User as ScimUser
 
-from .config import SAMLConfig
-from ..rbac import filter_claim_roles
-from ..resolver import ProvisioningStore
-from ..session import safe_relay_state
-
-if TYPE_CHECKING:
-    from ..security import AuthSecurity
+from litellm.proxy.auth_v2.config import SAMLConfig
 
 _SINGLE_VALUE_TARGETS = {
     "email",
@@ -30,9 +21,7 @@ _SINGLE_VALUE_TARGETS = {
 _MULTI_VALUE_TARGETS = ("groups", "roles")
 
 
-def _map_attributes(
-    ava: Dict[str, Any], attribute_map: Dict[str, str]
-) -> Dict[str, Any]:
+def map_attributes(ava: Dict[str, Any], attribute_map: Dict[str, str]) -> Dict[str, Any]:
     mapped: Dict[str, Any] = {}
     for saml_attr, target in attribute_map.items():
         if saml_attr not in ava:
@@ -49,13 +38,11 @@ def _map_attributes(
 def _formatted_name(mapped: Dict[str, Any]) -> Optional[str]:
     if mapped.get("display_name"):
         return mapped["display_name"]
-    parts: List[str] = [
-        part for part in (mapped.get("given_name"), mapped.get("family_name")) if part
-    ]
+    parts: List[str] = [part for part in (mapped.get("given_name"), mapped.get("family_name")) if part]
     return " ".join(parts) if parts else None
 
 
-def _user_from_mapped(name_id: str, mapped: Dict[str, Any]) -> ScimUser:
+def user_from_mapped(name_id: str, mapped: Dict[str, Any]) -> ScimUser:
     display = _formatted_name(mapped)
     user = ScimUser(
         external_id=name_id,
@@ -73,7 +60,7 @@ def _user_from_mapped(name_id: str, mapped: Dict[str, Any]) -> ScimUser:
     return user
 
 
-def _claims_from_mapped(mapped: Dict[str, Any]) -> Dict[str, Any]:
+def claims_from_mapped(mapped: Dict[str, Any]) -> Dict[str, Any]:
     claims: Dict[str, Any] = {}
     if mapped.get("email"):
         claims["email"] = mapped["email"]
@@ -102,9 +89,7 @@ def _sp_config_dict(config: SAMLConfig) -> Dict[str, Any]:
         "entityid": config.entity_id,
         "service": {
             "sp": {
-                "endpoints": {
-                    "assertion_consumer_service": [(config.acs_url, BINDING_HTTP_POST)]
-                },
+                "endpoints": {"assertion_consumer_service": [(config.acs_url, BINDING_HTTP_POST)]},
                 "allow_unsolicited": config.allow_unsolicited,
                 "authn_requests_signed": False,
                 "want_assertions_signed": True,
@@ -149,9 +134,7 @@ class SAMLProtocolStore:
 
     def outstanding_relays(self) -> Dict[str, str]:
         now = time.time()
-        return {
-            rid: relay for rid, (exp, relay) in self._outstanding.items() if exp >= now
-        }
+        return {rid: relay for rid, (exp, relay) in self._outstanding.items() if exp >= now}
 
     def consume_request(self, request_id: str) -> Optional[str]:
         entry = self._outstanding.pop(request_id, None)
@@ -171,99 +154,8 @@ class SAMLProtocolStore:
 
     def consume_assertion(self, assertion_id: str) -> bool:
         now = time.time()
-        self._seen_assertions = {
-            aid: exp for aid, exp in self._seen_assertions.items() if exp >= now
-        }
+        self._seen_assertions = {aid: exp for aid, exp in self._seen_assertions.items() if exp >= now}
         if assertion_id in self._seen_assertions:
             return False
         self._seen_assertions[assertion_id] = now + self._replay_ttl
         return True
-
-
-def build_saml_router(auth: "AuthSecurity") -> APIRouter:
-    config = auth.config.saml
-    assert config is not None
-    session = auth.config.session
-    client = build_sp_client(config)
-    protocol = SAMLProtocolStore(session.ttl_seconds)
-    router = APIRouter(prefix="/auth/saml", tags=["saml"])
-
-    @router.get("/metadata")
-    async def metadata() -> Response:
-        return Response(
-            content=str(entity_descriptor(client.config)),
-            media_type="application/samlmetadata+xml",
-        )
-
-    @router.get("/login")
-    async def login(request: Request) -> RedirectResponse:
-        relay_state = safe_relay_state(
-            request.query_params.get("next"), session.default_redirect_path
-        )
-        request_id, info = client.prepare_for_authenticate(relay_state=relay_state)
-        protocol.remember_request(request_id, relay_state)
-        location = dict(info["headers"]).get("Location")
-        if not location:
-            raise HTTPException(status_code=500, detail="no SAML redirect produced")
-        return RedirectResponse(location, status_code=303)
-
-    @router.post("/acs")
-    async def assertion_consumer_service(request: Request) -> Response:
-        form = await request.form()
-        saml_response = form.get("SAMLResponse")
-        if not isinstance(saml_response, str):
-            raise HTTPException(status_code=400, detail="missing SAMLResponse")
-        try:
-            authn_response = client.parse_authn_request_response(
-                saml_response,
-                BINDING_HTTP_POST,
-                outstanding=protocol.outstanding_relays() or None,
-            )
-        except Exception as exc:
-            raise HTTPException(
-                status_code=401, detail="invalid SAML response"
-            ) from exc
-        if authn_response is None:
-            raise HTTPException(status_code=401, detail="invalid SAML response")
-
-        in_response_to = getattr(authn_response, "in_response_to", None)
-        bound_relay = (
-            protocol.consume_request(in_response_to) if in_response_to else None
-        )
-
-        assertion = getattr(authn_response, "assertion", None)
-        assertion_id = getattr(assertion, "id", None)
-        if assertion_id and not protocol.consume_assertion(assertion_id):
-            raise HTTPException(status_code=401, detail="SAML assertion replay")
-
-        name_id = authn_response.get_subject().text
-        ava = authn_response.get_identity() or {}
-        mapped = _map_attributes(ava, config.attribute_map)
-        mapped["roles"] = filter_claim_roles(
-            mapped.get("roles"), config.allowed_roles, config.allow_platform_roles
-        )
-        user = _user_from_mapped(name_id, mapped)
-
-        store = cast(ProvisioningStore, auth.resolver)
-        await store.upsert_user(user)
-
-        session_id = auth.session_store.create_session(
-            {
-                "method": "saml",
-                "subject": name_id,
-                "issuer": authn_response.issuer(),
-                "claims": _claims_from_mapped(mapped),
-            }
-        )
-        target = safe_relay_state(bound_relay, session.default_redirect_path)
-        response = RedirectResponse(target, status_code=303)
-        response.set_cookie(
-            session.cookie,
-            session_id,
-            httponly=True,
-            samesite="lax",
-            secure=session.secure,
-        )
-        return response
-
-    return router

backend/auth/services/scim.py

@@ -0,0 +1,178 @@
+from __future__ import annotations
+
+from typing import Any, Callable, Coroutine, Dict, List, Type, TypeVar
+
+from fastapi import HTTPException, Request, Response, status
+from fastapi.responses import JSONResponse
+from fastapi.routing import APIRoute
+from scim2_models import (
+    Bulk,
+    ChangePassword,
+    Context,
+    Error,
+    Filter,
+    Group,
+    ListResponse,
+    Patch,
+    PatchOp,
+    Resource,
+    ResourceType,
+    Schema,
+    ServiceProviderConfig,
+    Sort,
+    User,
+)
+
+R = TypeVar("R", bound=Resource)
+
+
+def scim_error(status_code: int, detail: str) -> JSONResponse:
+    return JSONResponse(
+        status_code=status_code,
+        content=Error(status=str(status_code), detail=detail).model_dump(),
+    )
+
+
+class ScimErrorRoute(APIRoute):
+    """Render authentication failures with the SCIM Error schema (RFC 7644)."""
+
+    def get_route_handler(  # type: ignore[override]
+        self,
+    ) -> Callable[[Request], Coroutine[Any, Any, Response]]:
+        handler = super().get_route_handler()
+
+        async def scim_handler(request: Request) -> Response:
+            try:
+                return await handler(request)
+            except HTTPException as exc:
+                if exc.status_code not in (
+                    status.HTTP_401_UNAUTHORIZED,
+                    status.HTTP_403_FORBIDDEN,
+                ):
+                    raise
+                response = scim_error(exc.status_code, str(exc.detail))
+                if exc.headers:
+                    response.headers.update(exc.headers)
+                return response
+
+        return scim_handler
+
+
+def parse_resource(body: Any, model: Type[R]) -> R:
+    return model.model_validate(body, scim_ctx=Context.RESOURCE_CREATION_REQUEST)
+
+
+def _set_path(data: Dict[str, Any], path: str, value: Any) -> None:
+    keys = path.split(".")
+    node = data
+    for key in keys[:-1]:
+        child = node.get(key)
+        if not isinstance(child, dict):
+            child = {}
+            node[key] = child
+        node = child
+    node[keys[-1]] = value
+
+
+def _remove_path(data: Dict[str, Any], path: str) -> None:
+    keys = path.split(".")
+    node = data
+    for key in keys[:-1]:
+        child = node.get(key)
+        if not isinstance(child, dict):
+            return
+        node = child
+    node.pop(keys[-1], None)
+
+
+def _targets_read_only_id(op: Any) -> bool:
+    if op.path is not None:
+        return op.path.split(".")[0].strip().lower() == "id"
+    return isinstance(op.value, dict) and any(str(k).lower() == "id" for k in op.value)
+
+
+def apply_patch(resource: R, body: Any) -> R:
+    patch = PatchOp[type(resource)].model_validate(body)
+    data: Dict[str, Any] = resource.model_dump()
+    for op in patch.operations:
+        action = op.op.value if hasattr(op.op, "value") else str(op.op)
+        if op.path is not None and ("[" in op.path or "]" in op.path):
+            raise ValueError(f"unsupported SCIM patch path filter: {op.path}")
+        if _targets_read_only_id(op):
+            raise ValueError("the SCIM id attribute is read-only")
+        if action == "remove":
+            if op.path:
+                _remove_path(data, op.path)
+            continue
+        if op.path is None and isinstance(op.value, dict):
+            data.update(op.value)
+        elif op.path is not None:
+            _set_path(data, op.path, op.value)
+    return type(resource).model_validate(data)
+
+
+def creation_response(resource: Resource) -> Dict[str, Any]:
+    return resource.model_dump(scim_ctx=Context.RESOURCE_CREATION_RESPONSE)
+
+
+def query_response(resource: Resource) -> Dict[str, Any]:
+    return resource.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
+
+
+def patch_response(resource: Resource) -> Dict[str, Any]:
+    return resource.model_dump(scim_ctx=Context.RESOURCE_PATCH_RESPONSE)
+
+
+def list_response(model: Type[R], items: List[R]) -> Dict[str, Any]:
+    listing = ListResponse[model](
+        total_results=len(items),
+        start_index=1,
+        items_per_page=len(items),
+        resources=items or None,
+    )
+    return listing.model_dump(scim_ctx=Context.RESOURCE_QUERY_RESPONSE)
+
+
+def service_provider_config() -> Dict[str, Any]:
+    return ServiceProviderConfig(
+        patch=Patch(supported=True),
+        bulk=Bulk(supported=False, max_operations=0, max_payload_size=0),
+        filter=Filter(supported=False, max_results=0),
+        change_password=ChangePassword(supported=False),
+        sort=Sort(supported=False),
+        etag=None,
+        authentication_schemes=[],
+    ).model_dump()
+
+
+def resource_types() -> Dict[str, Any]:
+    types = [
+        ResourceType(
+            id="User",
+            name="User",
+            endpoint="/Users",
+            schema="urn:ietf:params:scim:schemas:core:2.0:User",
+        ),
+        ResourceType(
+            id="Group",
+            name="Group",
+            endpoint="/Groups",
+            schema="urn:ietf:params:scim:schemas:core:2.0:Group",
+        ),
+    ]
+    return ListResponse[ResourceType](
+        total_results=len(types),
+        start_index=1,
+        items_per_page=len(types),
+        resources=types,
+    ).model_dump()
+
+
+def schemas() -> Dict[str, Any]:
+    resources = [User.to_schema(), Group.to_schema()]
+    return ListResponse[Schema](
+        total_results=len(resources),
+        start_index=1,
+        items_per_page=len(resources),
+        resources=resources,
+    ).model_dump()

litellm/proxy/auth_v2/__init__.py

@@ -1,19 +1,18 @@
-from .config import (
+from litellm.proxy.auth_v2.config import (
     ApiKeySchemeConfig,
     AuthConfig,
     HttpBasicConfig,
     MutualTLSConfig,
     OAuth2IntrospectionConfig,
+    OIDCProviderConfig,
+    SAMLConfig,
+    SessionConfig,
     TrustedProxyConfig,
 )
-from .models import Principal
-from .oidc import OIDCProviderConfig, build_oidc_router
-from .rbac import Role
-from .resolver import IdentityResolver, InMemoryIdentityStore, ProvisioningStore
-from .saml import SAMLConfig, build_saml_router
-from .scim import build_scim_router
-from .security import AuthSecurity
-from .session import SessionConfig
+from litellm.proxy.auth_v2.models import Principal
+from litellm.proxy.auth_v2.authorization import Role
+from litellm.proxy.auth_v2.resolvers import IdentityResolver, InMemoryIdentityStore, ProvisioningStore
+from litellm.proxy.auth_v2.security import AuthSecurity
 
 __all__ = [
     "AuthSecurity",
@@ -31,7 +30,4 @@ __all__ = [
     "TrustedProxyConfig",
     "SessionConfig",
     "SAMLConfig",
-    "build_saml_router",
-    "build_scim_router",
-    "build_oidc_router",
 ]

litellm/proxy/auth_v2/authenticators.py

@@ -1,441 +0,0 @@
-from __future__ import annotations
-
-import base64
-import binascii
-import functools
-import hashlib
-import hmac
-import secrets
-from typing import Any, Callable, Dict, List, Optional, Protocol, runtime_checkable
-
-import jwt
-from fastapi import Request
-from jwt import PyJWKClient
-from jwt import decode as jwt_decode
-from starlette.concurrency import run_in_threadpool
-
-from . import errors
-from .config import (
-    ApiKeySchemeConfig,
-    AuthConfig,
-    HttpBasicConfig,
-    MutualTLSConfig,
-    OAuth2IntrospectionConfig,
-    TrustedProxyConfig,
-)
-from .models import (
-    AuthMethod,
-    ClientCertificate,
-    Credential,
-    CredentialRef,
-    SecuritySchemeType,
-)
-from .network import ip_in_trusted_proxies
-from .oidc.config import OIDCProviderConfig
-from .rbac import filter_claim_roles
-
-AT_JWT_TYPES = {"at+jwt", "application/at+jwt"}
-
-
-def _apply_role_policy(claims: Dict[str, Any], provider: OIDCProviderConfig) -> None:
-    claims["roles"] = filter_claim_roles(
-        claims.get("roles"), provider.allowed_roles, provider.allow_platform_roles
-    )
-
-
-@runtime_checkable
-class Authenticator(Protocol):
-    async def authenticate(self, request: Request) -> Optional[Credential]: ...
-
-    def challenge(self) -> str: ...
-
-
-@runtime_checkable
-class BasicAuthVerifier(Protocol):
-    def verify(self, username: str, password: str) -> bool: ...
-
-
-_PBKDF2_ITERATIONS = 600_000
-
-
-def hash_basic_password(password: str, salt: Optional[str] = None) -> str:
-    salt = salt or secrets.token_hex(16)
-    digest = hashlib.pbkdf2_hmac(
-        "sha256", password.encode(), bytes.fromhex(salt), _PBKDF2_ITERATIONS
-    ).hex()
-    return f"pbkdf2_sha256${_PBKDF2_ITERATIONS}${salt}${digest}"
-
-
-class InMemoryBasicAuthStore:
-    def __init__(self, credentials: Dict[str, str]) -> None:
-        self._credentials = credentials
-
-    def verify(self, username: str, password: str) -> bool:
-        stored = self._credentials.get(username)
-        if stored is None:
-            return False
-        try:
-            _algorithm, iterations, salt, expected = stored.split("$")
-            candidate = hashlib.pbkdf2_hmac(
-                "sha256", password.encode(), bytes.fromhex(salt), int(iterations)
-            ).hex()
-        except ValueError:
-            return False
-        return hmac.compare_digest(candidate, expected)
-
-
-def _extract_bearer(request: Request) -> Optional[str]:
-    header = request.headers.get("authorization")
-    if not header:
-        return None
-    scheme, _, value = header.partition(" ")
-    if scheme.lower() != "bearer" or not value:
-        return None
-    return value
-
-
-def _looks_like_jwt(token: str) -> bool:
-    return token.count(".") == 2
-
-
-def _normalize_audience(value: Any) -> List[str]:
-    if isinstance(value, str):
-        return [value]
-    if isinstance(value, list):
-        return [str(item) for item in value]
-    return []
-
-
-def _split_scope(value: Any) -> List[str]:
-    return value.split() if isinstance(value, str) else []
-
-
-def _credential_from_claims(
-    scheme: SecuritySchemeType,
-    method: AuthMethod,
-    token: str,
-    claims: Dict[str, Any],
-) -> Credential:
-    header = jwt.get_unverified_header(token)
-    return Credential(
-        scheme=scheme,
-        method=method,
-        subject=str(claims.get("sub", "")),
-        issuer=claims.get("iss"),
-        audience=_normalize_audience(claims.get("aud")),
-        scopes=_split_scope(claims.get("scope")),
-        claims=claims,
-        credential_ref=CredentialRef(
-            key_id=header.get("kid"), token_id=claims.get("jti")
-        ),
-    )
-
-
-class JWTVerifier:
-    def __init__(
-        self,
-        provider: OIDCProviderConfig,
-        jwks_client: Optional[PyJWKClient] = None,
-    ) -> None:
-        self.provider = provider
-        if jwks_client is not None:
-            self._jwks_client = jwks_client
-            return
-        jwks_uri = (
-            str(provider.jwks_uri) if provider.jwks_uri else self._discover_jwks()
-        )
-        self._jwks_client = PyJWKClient(
-            jwks_uri,
-            cache_keys=True,
-            cache_jwk_set=True,
-            lifespan=300,
-            timeout=10,
-        )
-
-    def _discover_jwks(self) -> str:
-        import httpx
-
-        url = f"{self.provider.issuer.rstrip('/')}/.well-known/openid-configuration"
-        response = httpx.get(url, timeout=10.0)
-        response.raise_for_status()
-        jwks_uri = response.json().get("jwks_uri")
-        if not jwks_uri:
-            raise ValueError(f"discovery document missing jwks_uri: {url}")
-        return str(jwks_uri)
-
-    def verify(
-        self, token: str, *, require_at_jwt: Optional[bool] = None
-    ) -> Dict[str, Any]:
-        enforce = (
-            self.provider.require_at_jwt if require_at_jwt is None else require_at_jwt
-        )
-        if enforce:
-            header = jwt.get_unverified_header(token)
-            if str(header.get("typ", "")).lower() not in AT_JWT_TYPES:
-                raise errors.invalid_token("token typ must be at+jwt")
-        try:
-            signing_key = self._jwks_client.get_signing_key_from_jwt(token)
-            return jwt_decode(
-                token,
-                signing_key.key,
-                algorithms=self.provider.algorithms,
-                audience=self.provider.audience,
-                issuer=self.provider.issuer,
-                options={"verify_exp": True, "require": ["exp", "iss", "aud"]},
-            )
-        except jwt.PyJWTError as exc:
-            raise errors.invalid_token("token verification failed") from exc
-
-
-async def _verify_jwt_off_loop(
-    verifier: JWTVerifier, token: str, *, require_at_jwt: Optional[bool] = None
-) -> Dict[str, Any]:
-    return await run_in_threadpool(
-        functools.partial(verifier.verify, token, require_at_jwt=require_at_jwt)
-    )
-
-
-def _select_verifier(token: str, verifiers: List[JWTVerifier]) -> Optional[JWTVerifier]:
-    if not verifiers:
-        return None
-    try:
-        issuer = jwt.decode(token, options={"verify_signature": False}).get("iss")
-    except jwt.PyJWTError:
-        return None
-    for verifier in verifiers:
-        if verifier.provider.issuer == issuer:
-            return verifier
-    return None
-
-
-async def _authenticate_bearer_jwt(
-    token: str,
-    verifiers: List[JWTVerifier],
-    scheme: SecuritySchemeType,
-    method: AuthMethod,
-    *,
-    require_at_jwt: bool = False,
-) -> Credential:
-    verifier = _select_verifier(token, verifiers)
-    if verifier is None:
-        raise errors.invalid_token("no issuer match")
-    claims = await _verify_jwt_off_loop(verifier, token, require_at_jwt=require_at_jwt)
-    _apply_role_policy(claims, verifier.provider)
-    return _credential_from_claims(scheme, method, token, claims)
-
-
-class APIKeyAuthenticator:
-    def __init__(self, config: ApiKeySchemeConfig) -> None:
-        self._header_name = config.header_name
-
-    async def authenticate(self, request: Request) -> Optional[Credential]:
-        raw = request.headers.get(self._header_name)
-        if not raw:
-            return None
-        return Credential(
-            scheme=SecuritySchemeType.API_KEY,
-            method=AuthMethod.API_KEY,
-            subject=raw,
-            credential_ref=CredentialRef(key_id=raw[:10]),
-            claims={"_raw_api_key": raw},
-        )
-
-    def challenge(self) -> str:
-        return ""
-
-
-class HttpAuthenticator:
-    def __init__(
-        self,
-        basic: HttpBasicConfig,
-        jwt_verifiers: List[JWTVerifier],
-        basic_verifier: Optional[BasicAuthVerifier] = None,
-    ) -> None:
-        self._basic = basic
-        self._verifiers = jwt_verifiers
-        self._basic_verifier = basic_verifier
-
-    async def authenticate(self, request: Request) -> Optional[Credential]:
-        header = request.headers.get("authorization")
-        if not header:
-            return None
-        scheme, _, value = header.partition(" ")
-        scheme_lower = scheme.lower()
-        if scheme_lower == "bearer" and value:
-            return await _authenticate_bearer_jwt(
-                value, self._verifiers, SecuritySchemeType.HTTP, AuthMethod.BEARER_JWT
-            )
-        if scheme_lower == "basic" and self._basic.enabled and value:
-            return self._verify_basic(value)
-        return None
-
-    def _verify_basic(self, value: str) -> Credential:
-        challenge = errors.basic_challenge(self._basic.realm)
-        try:
-            decoded = base64.b64decode(value).decode("utf-8")
-        except (binascii.Error, UnicodeDecodeError) as exc:
-            raise errors.unauthenticated(challenge) from exc
-        username, separator, password = decoded.partition(":")
-        if (
-            not username
-            or separator != ":"
-            or self._basic_verifier is None
-            or not self._basic_verifier.verify(username, password)
-        ):
-            raise errors.unauthenticated(challenge)
-        return Credential(
-            scheme=SecuritySchemeType.HTTP,
-            method=AuthMethod.HTTP_BASIC,
-            subject=username,
-        )
-
-    def challenge(self) -> str:
-        bearer = errors.bearer_challenge()
-        if self._basic.enabled:
-            return f"{bearer}, {errors.basic_challenge(self._basic.realm)}"
-        return bearer
-
-
-def _default_introspection_client() -> Any:
-    from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
-    from litellm.types.llms.custom_http import httpxSpecialProvider
-
-    return get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
-
-
-class OAuth2Authenticator:
-    def __init__(
-        self,
-        jwt_verifiers: List[JWTVerifier],
-        introspection: Optional[OAuth2IntrospectionConfig],
-        client_factory: Optional[Callable[[], Any]] = None,
-    ) -> None:
-        self._verifiers = jwt_verifiers
-        self._introspection = introspection
-        self._client_factory = client_factory or _default_introspection_client
-
-    async def authenticate(self, request: Request) -> Optional[Credential]:
-        token = _extract_bearer(request)
-        if token is None:
-            return None
-        if _looks_like_jwt(token):
-            return await _authenticate_bearer_jwt(
-                token,
-                self._verifiers,
-                SecuritySchemeType.OAUTH2,
-                AuthMethod.BEARER_JWT,
-                require_at_jwt=True,
-            )
-        if self._introspection is not None:
-            return await self._introspect(token)
-        raise errors.invalid_token()
-
-    async def _introspect(self, token: str) -> Credential:
-        config = self._introspection
-        assert config is not None
-        basic = base64.b64encode(
-            f"{config.client_id}:{config.client_secret.get_secret_value()}".encode()
-        ).decode()
-        client = self._client_factory()
-        response = await client.post(
-            str(config.introspection_endpoint),
-            data={"token": token},
-            headers={"Authorization": f"Basic {basic}"},
-            timeout=10.0,
-        )
-        if response.status_code != 200:
-            raise errors.invalid_token("introspection failed")
-        try:
-            body = response.json()
-        except ValueError as exc:
-            raise errors.invalid_token("introspection failed") from exc
-        if not isinstance(body, dict) or body.get("active") is not True:
-            raise errors.invalid_token("token inactive")
-        token_audience = _normalize_audience(body.get("aud"))
-        if config.audience and not set(token_audience) & set(config.audience):
-            raise errors.invalid_token("audience mismatch")
-        if config.issuer is not None and body.get("iss") != config.issuer:
-            raise errors.invalid_token("issuer mismatch")
-        claims = {key: value for key, value in body.items() if key != "roles"}
-        return Credential(
-            scheme=SecuritySchemeType.OAUTH2,
-            method=AuthMethod.OAUTH2_INTROSPECTION,
-            subject=str(body.get(config.subject_field, "")),
-            issuer=body.get("iss"),
-            audience=token_audience,
-            scopes=_split_scope(body.get("scope")),
-            claims=claims,
-        )
-
-    def challenge(self) -> str:
-        return errors.bearer_challenge()
-
-
-class OIDCAuthenticator:
-    def __init__(self, jwt_verifiers: List[JWTVerifier]) -> None:
-        self._verifiers = jwt_verifiers
-
-    async def authenticate(self, request: Request) -> Optional[Credential]:
-        token = _extract_bearer(request)
-        if token is None:
-            return None
-        return await _authenticate_bearer_jwt(
-            token, self._verifiers, SecuritySchemeType.OPENID_CONNECT, AuthMethod.OIDC
-        )
-
-    def challenge(self) -> str:
-        return errors.bearer_challenge()
-
-
-class MutualTLSAuthenticator:
-    def __init__(self, config: MutualTLSConfig, network: TrustedProxyConfig) -> None:
-        self._config = config
-        self._network = network
-
-    async def authenticate(self, request: Request) -> Optional[Credential]:
-        cert = self._read_client_cert(request)
-        if cert is None:
-            return None
-        return Credential(
-            scheme=SecuritySchemeType.MUTUAL_TLS,
-            method=AuthMethod.MUTUAL_TLS,
-            subject=cert.subject_dn,
-            client_certificate=cert,
-        )
-
-    def _read_client_cert(self, request: Request) -> Optional[ClientCertificate]:
-        tls = request.scope.get("extensions", {}).get("tls", {})
-        verified_dn = tls.get("client_cert_name")
-        if verified_dn:
-            return ClientCertificate(subject_dn=verified_dn)
-        if self._config.forwarded_subject_header:
-            peer = request.client.host if request.client else None
-            if not ip_in_trusted_proxies(peer, self._network):
-                return None
-            dn = request.headers.get(self._config.forwarded_subject_header)
-            return ClientCertificate(subject_dn=dn) if dn else None
-        return None
-
-    def challenge(self) -> str:
-        return ""
-
-
-def build_authenticators(
-    config: AuthConfig, *, basic_verifier: Optional[BasicAuthVerifier] = None
-) -> List[Authenticator]:
-    verifiers = [JWTVerifier(provider) for provider in config.oidc_providers]
-    by_scheme: Dict[SecuritySchemeType, Authenticator] = {}
-    if config.api_key is not None:
-        by_scheme[SecuritySchemeType.API_KEY] = APIKeyAuthenticator(config.api_key)
-    by_scheme[SecuritySchemeType.HTTP] = HttpAuthenticator(
-        config.http_basic, verifiers, basic_verifier
-    )
-    by_scheme[SecuritySchemeType.OPENID_CONNECT] = OIDCAuthenticator(verifiers)
-    by_scheme[SecuritySchemeType.OAUTH2] = OAuth2Authenticator(
-        verifiers, config.oauth2_introspection
-    )
-    if config.mutual_tls.enabled:
-        by_scheme[SecuritySchemeType.MUTUAL_TLS] = MutualTLSAuthenticator(
-            config.mutual_tls, config.network
-        )
-    return [by_scheme[scheme] for scheme in config.scheme_order if scheme in by_scheme]

litellm/proxy/auth_v2/authenticators/__init__.py

@@ -0,0 +1,23 @@
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+from litellm.proxy.auth_v2.authenticators.config import build_authenticators
+from litellm.proxy.auth_v2.authenticators.http import HttpAuthenticator, hash_basic_password
+from litellm.proxy.auth_v2.authenticators.key import APIKeyAuthenticator
+from litellm.proxy.auth_v2.authenticators.mtls import MutualTLSAuthenticator
+from litellm.proxy.auth_v2.authenticators.oauth import OAuth2Authenticator
+from litellm.proxy.auth_v2.authenticators.oidc import OIDCAuthenticator
+from litellm.proxy.auth_v2.authenticators.types import BasicAuthVerifier
+from litellm.proxy.auth_v2.authenticators.utils import JWTVerifier, apply_role_policy
+
+__all__ = [
+    "Authenticator",
+    "BasicAuthVerifier",
+    "JWTVerifier",
+    "APIKeyAuthenticator",
+    "HttpAuthenticator",
+    "OAuth2Authenticator",
+    "OIDCAuthenticator",
+    "MutualTLSAuthenticator",
+    "hash_basic_password",
+    "apply_role_policy",
+    "build_authenticators",
+]

litellm/proxy/auth_v2/authenticators/base.py

@@ -0,0 +1,14 @@
+from __future__ import annotations
+
+from typing import Optional, Protocol, runtime_checkable
+
+from fastapi import Request
+
+from litellm.proxy.auth_v2.models import Credential
+
+
+@runtime_checkable
+class Authenticator(Protocol):
+    async def authenticate(self, request: Request) -> Optional[Credential]: ...
+
+    def challenge(self) -> str: ...

litellm/proxy/auth_v2/authenticators/config.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Dict, List, Optional
+
+from litellm.proxy.auth_v2.config import AuthConfig
+from litellm.proxy.auth_v2.models import SecuritySchemeType
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+from litellm.proxy.auth_v2.authenticators.http import HttpAuthenticator
+from litellm.proxy.auth_v2.authenticators.key import APIKeyAuthenticator
+from litellm.proxy.auth_v2.authenticators.mtls import MutualTLSAuthenticator
+from litellm.proxy.auth_v2.authenticators.oauth import OAuth2Authenticator
+from litellm.proxy.auth_v2.authenticators.oidc import OIDCAuthenticator
+from litellm.proxy.auth_v2.authenticators.types import BasicAuthVerifier
+from litellm.proxy.auth_v2.authenticators.utils import JWTVerifier
+
+
+def build_authenticators(
+    config: AuthConfig, *, basic_verifier: Optional[BasicAuthVerifier] = None
+) -> List[Authenticator]:
+    verifiers = [JWTVerifier(provider) for provider in config.oidc_providers]
+    by_scheme: Dict[SecuritySchemeType, Authenticator] = {}
+    if config.api_key is not None:
+        by_scheme[SecuritySchemeType.API_KEY] = APIKeyAuthenticator(config.api_key)
+    by_scheme[SecuritySchemeType.HTTP] = HttpAuthenticator(config.http_basic, verifiers, basic_verifier)
+    by_scheme[SecuritySchemeType.OPENID_CONNECT] = OIDCAuthenticator(verifiers)
+    by_scheme[SecuritySchemeType.OAUTH2] = OAuth2Authenticator(verifiers, config.oauth2_introspection)
+    if config.mutual_tls.enabled:
+        by_scheme[SecuritySchemeType.MUTUAL_TLS] = MutualTLSAuthenticator(config.mutual_tls, config.network)
+    return [by_scheme[scheme] for scheme in config.scheme_order if scheme in by_scheme]

litellm/proxy/auth_v2/authenticators/http.py

@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import secrets
+from typing import List, Optional
+
+from fastapi import Request
+
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.config import HttpBasicConfig
+from litellm.proxy.auth_v2.models import AuthMethod, Credential, SecuritySchemeType
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+from litellm.proxy.auth_v2.authenticators.types import BasicAuthVerifier
+from litellm.proxy.auth_v2.authenticators.utils import JWTVerifier, authenticate_bearer_jwt
+
+_PBKDF2_ITERATIONS = 600_000
+
+
+def hash_basic_password(password: str, salt: Optional[str] = None) -> str:
+    salt = salt or secrets.token_hex(16)
+    digest = hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), _PBKDF2_ITERATIONS).hex()
+    return f"pbkdf2_sha256${_PBKDF2_ITERATIONS}${salt}${digest}"
+
+
+class HttpAuthenticator(Authenticator):
+    def __init__(
+        self,
+        basic: HttpBasicConfig,
+        jwt_verifiers: List[JWTVerifier],
+        basic_verifier: Optional[BasicAuthVerifier] = None,
+    ) -> None:
+        self._basic = basic
+        self._verifiers = jwt_verifiers
+        self._basic_verifier = basic_verifier
+
+    async def authenticate(self, request: Request) -> Optional[Credential]:
+        header = request.headers.get("authorization")
+        if not header:
+            return None
+        scheme, _, value = header.partition(" ")
+        scheme_lower = scheme.lower()
+        if scheme_lower == "bearer" and value:
+            return await authenticate_bearer_jwt(value, self._verifiers, SecuritySchemeType.HTTP, AuthMethod.BEARER_JWT)
+        if scheme_lower == "basic" and self._basic.enabled and value:
+            return self._verify_basic(value)
+        return None
+
+    def _verify_basic(self, value: str) -> Credential:
+        challenge = errors.basic_challenge(self._basic.realm)
+        try:
+            decoded = base64.b64decode(value).decode("utf-8")
+        except (binascii.Error, UnicodeDecodeError) as exc:
+            raise errors.unauthenticated(challenge) from exc
+        username, separator, password = decoded.partition(":")
+        if (
+            not username
+            or separator != ":"
+            or self._basic_verifier is None
+            or not self._basic_verifier.verify(username, password)
+        ):
+            raise errors.unauthenticated(challenge)
+        return Credential(
+            scheme=SecuritySchemeType.HTTP,
+            method=AuthMethod.HTTP_BASIC,
+            subject=username,
+        )
+
+    def challenge(self) -> str:
+        bearer = errors.bearer_challenge()
+        if self._basic.enabled:
+            return f"{bearer}, {errors.basic_challenge(self._basic.realm)}"
+        return bearer

litellm/proxy/auth_v2/authenticators/key.py

@@ -0,0 +1,29 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import Request
+
+from litellm.proxy.auth_v2.config import ApiKeySchemeConfig
+from litellm.proxy.auth_v2.models import AuthMethod, Credential, CredentialRef, SecuritySchemeType
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+
+
+class APIKeyAuthenticator(Authenticator):
+    def __init__(self, config: ApiKeySchemeConfig) -> None:
+        self._header_name = config.header_name
+
+    async def authenticate(self, request: Request) -> Optional[Credential]:
+        raw = request.headers.get(self._header_name)
+        if not raw:
+            return None
+        return Credential(
+            scheme=SecuritySchemeType.API_KEY,
+            method=AuthMethod.API_KEY,
+            subject=raw,
+            credential_ref=CredentialRef(key_id=raw[:10]),
+            claims={"_raw_api_key": raw},
+        )
+
+    def challenge(self) -> str:
+        return ""

litellm/proxy/auth_v2/authenticators/mtls.py

@@ -0,0 +1,43 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import Request
+
+from litellm.proxy.auth_v2.config import MutualTLSConfig, TrustedProxyConfig
+from litellm.proxy.auth_v2.models import AuthMethod, ClientCertificate, Credential, SecuritySchemeType
+from litellm.proxy.auth_v2.network import ip_in_trusted_proxies
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+
+
+class MutualTLSAuthenticator(Authenticator):
+    def __init__(self, config: MutualTLSConfig, network: TrustedProxyConfig) -> None:
+        self._config = config
+        self._network = network
+
+    async def authenticate(self, request: Request) -> Optional[Credential]:
+        cert = self._read_client_cert(request)
+        if cert is None:
+            return None
+        return Credential(
+            scheme=SecuritySchemeType.MUTUAL_TLS,
+            method=AuthMethod.MUTUAL_TLS,
+            subject=cert.subject_dn,
+            client_certificate=cert,
+        )
+
+    def _read_client_cert(self, request: Request) -> Optional[ClientCertificate]:
+        tls = request.scope.get("extensions", {}).get("tls", {})
+        verified_dn = tls.get("client_cert_name")
+        if verified_dn:
+            return ClientCertificate(subject_dn=verified_dn)
+        if self._config.forwarded_subject_header:
+            peer = request.client.host if request.client else None
+            if not ip_in_trusted_proxies(peer, self._network):
+                return None
+            dn = request.headers.get(self._config.forwarded_subject_header)
+            return ClientCertificate(subject_dn=dn) if dn else None
+        return None
+
+    def challenge(self) -> str:
+        return ""

litellm/proxy/auth_v2/authenticators/oauth.py

@@ -0,0 +1,93 @@
+from __future__ import annotations
+
+import base64
+from typing import List, Optional
+
+from fastapi import Request
+
+from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
+from litellm.types.llms.custom_http import httpxSpecialProvider
+
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.config import OAuth2IntrospectionConfig
+from litellm.proxy.auth_v2.models import AuthMethod, Credential, SecuritySchemeType
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+from litellm.proxy.auth_v2.authenticators.types import IntrospectionClient, IntrospectionClientFactory
+from litellm.proxy.auth_v2.authenticators.utils import (
+    JWTVerifier,
+    authenticate_bearer_jwt,
+    extract_bearer,
+    looks_like_jwt,
+    normalize_audience,
+    split_scope,
+)
+
+
+def _default_introspection_client() -> IntrospectionClient:
+    return get_async_httpx_client(llm_provider=httpxSpecialProvider.Oauth2Check)
+
+
+class OAuth2Authenticator(Authenticator):
+    def __init__(
+        self,
+        jwt_verifiers: List[JWTVerifier],
+        introspection: Optional[OAuth2IntrospectionConfig],
+        client_factory: Optional[IntrospectionClientFactory] = None,
+    ) -> None:
+        self._verifiers = jwt_verifiers
+        self._introspection = introspection
+        self._client_factory = client_factory or _default_introspection_client
+
+    async def authenticate(self, request: Request) -> Optional[Credential]:
+        token = extract_bearer(request)
+        if token is None:
+            return None
+        if looks_like_jwt(token):
+            return await authenticate_bearer_jwt(
+                token,
+                self._verifiers,
+                SecuritySchemeType.OAUTH2,
+                AuthMethod.BEARER_JWT,
+                require_at_jwt=True,
+            )
+        if self._introspection is not None:
+            return await self._introspect(token)
+        raise errors.invalid_token()
+
+    async def _introspect(self, token: str) -> Credential:
+        config = self._introspection
+        assert config is not None
+        basic = base64.b64encode(f"{config.client_id}:{config.client_secret.get_secret_value()}".encode()).decode()
+        client = self._client_factory()
+        response = await client.post(
+            str(config.introspection_endpoint),
+            data={"token": token},
+            headers={"Authorization": f"Basic {basic}"},
+            timeout=10.0,
+        )
+        if response.status_code != 200:
+            raise errors.invalid_token("introspection failed")
+        try:
+            body = response.json()
+        except ValueError as exc:
+            raise errors.invalid_token("introspection failed") from exc
+        if not isinstance(body, dict) or body.get("active") is not True:
+            raise errors.invalid_token("token inactive")
+        token_audience = normalize_audience(body.get("aud"))
+        if config.audience and not set(token_audience) & set(config.audience):
+            raise errors.invalid_token("audience mismatch")
+        if config.issuer is not None and body.get("iss") != config.issuer:
+            raise errors.invalid_token("issuer mismatch")
+        claims = {key: value for key, value in body.items() if key != "roles"}
+        return Credential(
+            scheme=SecuritySchemeType.OAUTH2,
+            method=AuthMethod.OAUTH2_INTROSPECTION,
+            subject=str(body.get(config.subject_field, "")),
+            issuer=body.get("iss"),
+            audience=token_audience,
+            scopes=split_scope(body.get("scope")),
+            claims=claims,
+        )
+
+    def challenge(self) -> str:
+        return errors.bearer_challenge()

litellm/proxy/auth_v2/authenticators/oidc.py

@@ -0,0 +1,24 @@
+from __future__ import annotations
+
+from typing import List, Optional
+
+from fastapi import Request
+
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.models import AuthMethod, Credential, SecuritySchemeType
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+from litellm.proxy.auth_v2.authenticators.utils import JWTVerifier, authenticate_bearer_jwt, extract_bearer
+
+
+class OIDCAuthenticator(Authenticator):
+    def __init__(self, jwt_verifiers: List[JWTVerifier]) -> None:
+        self._verifiers = jwt_verifiers
+
+    async def authenticate(self, request: Request) -> Optional[Credential]:
+        token = extract_bearer(request)
+        if token is None:
+            return None
+        return await authenticate_bearer_jwt(token, self._verifiers, SecuritySchemeType.OPENID_CONNECT, AuthMethod.OIDC)
+
+    def challenge(self) -> str:
+        return errors.bearer_challenge()

litellm/proxy/auth_v2/authenticators/session.py

@@ -0,0 +1,40 @@
+from __future__ import annotations
+
+from typing import Optional
+
+from fastapi import Request
+
+from litellm.proxy.auth_v2.authenticators.base import Authenticator
+from litellm.proxy.auth_v2.models import (
+    AuthMethod,
+    Credential,
+    CredentialRef,
+    SecuritySchemeType,
+)
+from litellm.proxy.auth_v2.sessions import StateStore
+from litellm.proxy.auth_v2.sessions.schemas import SessionState
+
+
+class SessionAuthenticator(Authenticator):
+    def __init__(self, cookie_name: str, store: "StateStore[SessionState]") -> None:
+        self._cookie_name = cookie_name
+        self._store = store
+
+    async def authenticate(self, request: Request) -> Optional[Credential]:
+        session_id = request.cookies.get(self._cookie_name)
+        if not session_id:
+            return None
+        identity = await self._store.get(session_id)
+        if identity is None:
+            return None
+        return Credential(
+            scheme=SecuritySchemeType.API_KEY,
+            method=AuthMethod(identity["method"]),
+            subject=identity["subject"],
+            issuer=identity.get("issuer"),
+            claims=identity.get("claims", {}),
+            credential_ref=CredentialRef(token_id=session_id),
+        )
+
+    def challenge(self) -> str:
+        return ""

litellm/proxy/auth_v2/authenticators/types.py

@@ -0,0 +1,30 @@
+from __future__ import annotations
+
+from typing import Callable, Dict, Mapping, Protocol, runtime_checkable
+
+Claims = Dict[str, object]
+
+
+@runtime_checkable
+class BasicAuthVerifier(Protocol):
+    def verify(self, username: str, password: str) -> bool: ...
+
+
+class IntrospectionResponse(Protocol):
+    status_code: int
+
+    def json(self) -> object: ...
+
+
+class IntrospectionClient(Protocol):
+    async def post(
+        self,
+        url: str,
+        *,
+        data: Mapping[str, str],
+        headers: Mapping[str, str],
+        timeout: float,
+    ) -> IntrospectionResponse: ...
+
+
+IntrospectionClientFactory = Callable[[], IntrospectionClient]

litellm/proxy/auth_v2/authenticators/utils.py

@@ -0,0 +1,150 @@
+from __future__ import annotations
+
+import functools
+from typing import List, Optional
+
+import httpx
+import jwt
+from fastapi import Request
+from jwt import PyJWKClient
+from jwt import decode as jwt_decode
+from starlette.concurrency import run_in_threadpool
+
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.models import AuthMethod, Credential, CredentialRef, SecuritySchemeType
+from litellm.proxy.auth_v2.config import OIDCProviderConfig
+from litellm.proxy.auth_v2.authorization import filter_claim_roles
+from litellm.proxy.auth_v2.authenticators.types import Claims
+
+AT_JWT_TYPES = {"at+jwt", "application/at+jwt"}
+
+
+def apply_role_policy(claims: Claims, provider: OIDCProviderConfig) -> None:
+    claims["roles"] = filter_claim_roles(claims.get("roles"), provider.allowed_roles, provider.allow_platform_roles)
+
+
+def extract_bearer(request: Request) -> Optional[str]:
+    header = request.headers.get("authorization")
+    if not header:
+        return None
+    scheme, _, value = header.partition(" ")
+    if scheme.lower() != "bearer" or not value:
+        return None
+    return value
+
+
+def looks_like_jwt(token: str) -> bool:
+    return token.count(".") == 2
+
+
+def normalize_audience(value: object) -> List[str]:
+    if isinstance(value, str):
+        return [value]
+    if isinstance(value, list):
+        return [str(item) for item in value]
+    return []
+
+
+def split_scope(value: object) -> List[str]:
+    return value.split() if isinstance(value, str) else []
+
+
+def credential_from_claims(
+    scheme: SecuritySchemeType,
+    method: AuthMethod,
+    token: str,
+    claims: Claims,
+) -> Credential:
+    header = jwt.get_unverified_header(token)
+    issuer = claims.get("iss")
+    return Credential(
+        scheme=scheme,
+        method=method,
+        subject=str(claims.get("sub", "")),
+        issuer=issuer if isinstance(issuer, str) else None,
+        audience=normalize_audience(claims.get("aud")),
+        scopes=split_scope(claims.get("scope")),
+        claims=claims,
+        credential_ref=CredentialRef(key_id=header.get("kid"), token_id=claims.get("jti")),
+    )
+
+
+class JWTVerifier:
+    def __init__(
+        self,
+        provider: OIDCProviderConfig,
+        jwks_client: Optional[PyJWKClient] = None,
+    ) -> None:
+        self.provider = provider
+        if jwks_client is not None:
+            self._jwks_client = jwks_client
+            return
+        jwks_uri = str(provider.jwks_uri) if provider.jwks_uri else self._discover_jwks()
+        self._jwks_client = PyJWKClient(
+            jwks_uri,
+            cache_keys=True,
+            cache_jwk_set=True,
+            lifespan=300,
+            timeout=10,
+        )
+
+    def _discover_jwks(self) -> str:
+        url = f"{self.provider.issuer.rstrip('/')}/.well-known/openid-configuration"
+        response = httpx.get(url, timeout=10.0)
+        response.raise_for_status()
+        jwks_uri = response.json().get("jwks_uri")
+        if not jwks_uri:
+            raise ValueError(f"discovery document missing jwks_uri: {url}")
+        return str(jwks_uri)
+
+    def verify(self, token: str, *, require_at_jwt: Optional[bool] = None) -> Claims:
+        enforce = self.provider.require_at_jwt if require_at_jwt is None else require_at_jwt
+        if enforce:
+            header = jwt.get_unverified_header(token)
+            if str(header.get("typ", "")).lower() not in AT_JWT_TYPES:
+                raise errors.invalid_token("token typ must be at+jwt")
+        try:
+            signing_key = self._jwks_client.get_signing_key_from_jwt(token)
+            return jwt_decode(
+                token,
+                signing_key.key,
+                algorithms=self.provider.algorithms,
+                audience=self.provider.audience,
+                issuer=self.provider.issuer,
+                options={"verify_exp": True, "require": ["exp", "iss", "aud"]},
+            )
+        except jwt.PyJWTError as exc:
+            raise errors.invalid_token("token verification failed") from exc
+
+
+async def _verify_jwt_off_loop(verifier: JWTVerifier, token: str, *, require_at_jwt: Optional[bool] = None) -> Claims:
+    return await run_in_threadpool(functools.partial(verifier.verify, token, require_at_jwt=require_at_jwt))
+
+
+def _select_verifier(token: str, verifiers: List[JWTVerifier]) -> Optional[JWTVerifier]:
+    if not verifiers:
+        return None
+    try:
+        issuer = jwt.decode(token, options={"verify_signature": False}).get("iss")
+    except jwt.PyJWTError:
+        return None
+    for verifier in verifiers:
+        if verifier.provider.issuer == issuer:
+            return verifier
+    return None
+
+
+async def authenticate_bearer_jwt(
+    token: str,
+    verifiers: List[JWTVerifier],
+    scheme: SecuritySchemeType,
+    method: AuthMethod,
+    *,
+    require_at_jwt: bool = False,
+) -> Credential:
+    verifier = _select_verifier(token, verifiers)
+    if verifier is None:
+        raise errors.invalid_token("no issuer match")
+    claims = await _verify_jwt_off_loop(verifier, token, require_at_jwt=require_at_jwt)
+    apply_role_policy(claims, verifier.provider)
+    return credential_from_claims(scheme, method, token, claims)

litellm/proxy/auth_v2/authorization/__init__.py

@@ -0,0 +1,12 @@
+from litellm.proxy.auth_v2.authorization.base import Authorizer
+from litellm.proxy.auth_v2.authorization.rbac import RBACEngine
+from litellm.proxy.auth_v2.authorization.roles import Role, filter_claim_roles
+from litellm.proxy.auth_v2.authorization.scopes import has_required_scopes
+
+__all__ = [
+    "Authorizer",
+    "RBACEngine",
+    "Role",
+    "filter_claim_roles",
+    "has_required_scopes",
+]

litellm/proxy/auth_v2/authorization/base.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Protocol, Tuple, runtime_checkable
+
+if TYPE_CHECKING:
+    from litellm.proxy.auth_v2.authorization.roles import Role
+    from litellm.proxy.auth_v2.models import Principal
+
+
+@runtime_checkable
+class Authorizer(Protocol):
+    """Decides what an authenticated principal is allowed to do.
+
+    This is the extension point for authorization methods. RBAC is the only
+    implementation today; add others (ABAC, ReBAC, an external PDP, ...) by
+    implementing this protocol and passing the instance to
+    ``AuthSecurity(..., authorizer=...)``.
+    """
+
+    def enforce(self, principal: "Principal", obj: str, act: str) -> bool:
+        """Return True if ``principal`` may perform ``act`` on resource ``obj``."""
+        ...
+
+    def has_any_role(self, principal: "Principal", allowed: "Tuple[Role, ...]") -> bool:
+        """Return True if ``principal`` holds (or inherits) any of ``allowed``."""
+        ...

litellm/proxy/auth_v2/authorization/rbac.py

@@ -1,44 +1,14 @@
 from __future__ import annotations
 
-from enum import Enum
-from typing import TYPE_CHECKING, Any, List, Optional, Tuple
+from typing import TYPE_CHECKING, List, Optional, Tuple
 
 import casbin
-from fastapi.security import SecurityScopes
+
+from litellm.proxy.auth_v2.authorization.base import Authorizer
+from litellm.proxy.auth_v2.authorization.roles import Role
 
 if TYPE_CHECKING:
-    from .models import Principal
-
-
-class Role(str, Enum):
-    PLATFORM_ADMIN = "platform_admin"
-    PLATFORM_VIEWER = "platform_viewer"
-    ORG_ADMIN = "org_admin"
-    ORG_VIEWER = "org_viewer"
-    TEAM_ADMIN = "team_admin"
-    TEAM_MEMBER = "team_member"
-
-
-_PLATFORM_ROLE_VALUES = {Role.PLATFORM_ADMIN.value, Role.PLATFORM_VIEWER.value}
-
-
-def filter_claim_roles(
-    roles: Any, allowed_roles: List[str], allow_platform_roles: bool
-) -> List[str]:
-    if not isinstance(roles, list):
-        return []
-    allowed = set(allowed_roles)
-    filtered = [role for role in roles if role in allowed]
-    if not allow_platform_roles:
-        filtered = [role for role in filtered if role not in _PLATFORM_ROLE_VALUES]
-    return filtered
-
-
-def has_required_scopes(
-    security_scopes: SecurityScopes, principal: "Principal"
-) -> bool:
-    return set(security_scopes.scopes).issubset(set(principal.scopes))
-
+    from litellm.proxy.auth_v2.models import Principal
 
 _MODEL_TEXT = """
 [request_definition]
@@ -72,7 +42,7 @@ _DEFAULT_POLICY: List[Tuple[str, str, str]] = [
 ]
 
 
-class RBACEngine:
+class RBACEngine(Authorizer):
     def __init__(self, policy_path: Optional[str] = None) -> None:
         model = casbin.Model()
         model.load_model_from_text(_MODEL_TEXT)
@@ -86,9 +56,7 @@ class RBACEngine:
             self._enforcer.add_policy(*rule)
 
     def enforce(self, principal: "Principal", obj: str, act: str) -> bool:
-        return any(
-            self._enforcer.enforce(role.value, obj, act) for role in principal.roles
-        )
+        return any(self._enforcer.enforce(role.value, obj, act) for role in principal.roles)
 
     def has_any_role(self, principal: "Principal", allowed: Tuple[Role, ...]) -> bool:
         allowed_values = {role.value for role in allowed}

litellm/proxy/auth_v2/authorization/roles.py

@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Any, List
+
+
+class Role(str, Enum):
+    PLATFORM_ADMIN = "platform_admin"
+    PLATFORM_VIEWER = "platform_viewer"
+    ORG_ADMIN = "org_admin"
+    ORG_VIEWER = "org_viewer"
+    TEAM_ADMIN = "team_admin"
+    TEAM_MEMBER = "team_member"
+
+
+_PLATFORM_ROLE_VALUES = {Role.PLATFORM_ADMIN.value, Role.PLATFORM_VIEWER.value}
+
+
+def filter_claim_roles(roles: Any, allowed_roles: List[str], allow_platform_roles: bool) -> List[str]:
+    if not isinstance(roles, list):
+        return []
+    allowed = set(allowed_roles)
+    filtered = [role for role in roles if role in allowed]
+    if not allow_platform_roles:
+        filtered = [role for role in filtered if role not in _PLATFORM_ROLE_VALUES]
+    return filtered

litellm/proxy/auth_v2/authorization/scopes.py

@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from fastapi.security import SecurityScopes
+
+if TYPE_CHECKING:
+    from litellm.proxy.auth_v2.models import Principal
+
+
+def has_required_scopes(security_scopes: SecurityScopes, principal: "Principal") -> bool:
+    return set(security_scopes.scopes).issubset(set(principal.scopes))

litellm/proxy/auth_v2/config.py

@@ -1,11 +1,84 @@
-from typing import List, Optional
+from typing import Dict, List, Optional
 
-from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr, field_validator
+from pydantic import (
+    AnyHttpUrl,
+    BaseModel,
+    Field,
+    SecretStr,
+    field_validator,
+    model_validator,
+)
 
-from .models import SecuritySchemeType, require_secure_url
-from .oidc.config import OIDCProviderConfig
-from .saml.config import SAMLConfig
-from .session import SessionConfig
+from litellm.proxy.auth_v2.models import SecuritySchemeType, require_secure_url
+
+
+class SessionConfig(BaseModel):
+    cookie: str = "litellm_session"
+    secure: bool = True
+    ttl_seconds: int = 3600
+    max_size: int = 10000
+    default_redirect_path: str = "/"
+    login_cookie: str = "litellm_oidc_txn"
+    login_state_ttl: int = 300
+
+
+DEFAULT_SAML_ATTRIBUTE_MAP = {
+    "email": "email",
+    "mail": "email",
+    "givenName": "given_name",
+    "surname": "family_name",
+    "sn": "family_name",
+    "displayName": "display_name",
+    "userName": "user_name",
+    "uid": "user_name",
+    "groups": "groups",
+    "roles": "roles",
+}
+
+
+class OIDCProviderConfig(BaseModel):
+    issuer: str
+    audience: List[str]
+    jwks_uri: Optional[AnyHttpUrl] = None
+    algorithms: List[str] = Field(default_factory=lambda: ["RS256"])
+    require_at_jwt: bool = False
+    client_id: Optional[str] = None
+    client_secret: Optional[SecretStr] = None
+    login_scopes: List[str] = Field(default_factory=lambda: ["openid", "email", "profile"])
+    allowed_roles: List[str] = Field(default_factory=list)
+    allow_platform_roles: bool = False
+
+    @field_validator("issuer")
+    @classmethod
+    def _issuer_https(cls, value: str) -> str:
+        return require_secure_url(value)
+
+    @field_validator("jwks_uri")
+    @classmethod
+    def _jwks_https(cls, value: Optional[AnyHttpUrl]) -> Optional[AnyHttpUrl]:
+        if value is not None:
+            require_secure_url(str(value))
+        return value
+
+
+class SAMLConfig(BaseModel):
+    enabled: bool = False
+    entity_id: str
+    acs_url: str
+    idp_metadata: str = ""
+    sp_key_file: Optional[str] = None
+    sp_cert_file: Optional[str] = None
+    allow_unsolicited: bool = False
+    xmlsec_binary: Optional[str] = None
+    attribute_map: Dict[str, str] = Field(default_factory=lambda: dict(DEFAULT_SAML_ATTRIBUTE_MAP))
+    allowed_roles: List[str] = Field(default_factory=list)
+    allow_platform_roles: bool = False
+
+    @model_validator(mode="after")
+    def _require_idp_metadata(self) -> "SAMLConfig":
+        if self.enabled and not self.idp_metadata.strip():
+            raise ValueError("SAML enabled but idp_metadata is empty (inline XML, local path, or URL)")
+        return self
 
 
 class ApiKeySchemeConfig(BaseModel):

litellm/proxy/auth_v2/errors.py

@@ -6,16 +6,12 @@ from fastapi import HTTPException
 
 
 class AuthError(HTTPException):
-    def __init__(
-        self, status_code: int, detail: str, challenge: Optional[str] = None
-    ) -> None:
+    def __init__(self, status_code: int, detail: str, challenge: Optional[str] = None) -> None:
         headers = {"WWW-Authenticate": challenge} if challenge else None
         super().__init__(status_code=status_code, detail=detail, headers=headers)
 
 
-def bearer_challenge(
-    error: Optional[str] = None, description: Optional[str] = None
-) -> str:
+def bearer_challenge(error: Optional[str] = None, description: Optional[str] = None) -> str:
     parts = ['Bearer realm="litellm"']
     if error:
         parts.append(f'error="{error}"')
@@ -33,9 +29,7 @@ def unauthenticated(challenge: str) -> AuthError:
 
 
 def invalid_token(description: Optional[str] = None) -> AuthError:
-    return AuthError(
-        401, "Invalid token", bearer_challenge("invalid_token", description)
-    )
+    return AuthError(401, "Invalid token", bearer_challenge("invalid_token", description))
 
 
 def insufficient_scope() -> AuthError:
@@ -52,3 +46,39 @@ def forbidden_permission() -> AuthError:
 
 def account_disabled() -> AuthError:
     return AuthError(403, "Account disabled")
+
+
+def unknown_provider() -> HTTPException:
+    return HTTPException(status_code=404, detail="unknown provider")
+
+
+def invalid_login_state() -> HTTPException:
+    return HTTPException(status_code=400, detail="invalid or expired login state")
+
+
+def state_mismatch() -> HTTPException:
+    return HTTPException(status_code=400, detail="state mismatch")
+
+
+def oidc_provider_error(error: str) -> HTTPException:
+    return HTTPException(status_code=400, detail=error)
+
+
+def missing_authorization_code() -> HTTPException:
+    return HTTPException(status_code=400, detail="missing authorization code")
+
+
+def saml_redirect_failed() -> HTTPException:
+    return HTTPException(status_code=500, detail="no SAML redirect produced")
+
+
+def missing_saml_response() -> HTTPException:
+    return HTTPException(status_code=400, detail="missing SAMLResponse")
+
+
+def invalid_saml_response() -> HTTPException:
+    return HTTPException(status_code=401, detail="invalid SAML response")
+
+
+def saml_assertion_replay() -> HTTPException:
+    return HTTPException(status_code=401, detail="SAML assertion replay")

litellm/proxy/auth_v2/models.py

@@ -6,7 +6,7 @@ from urllib.parse import urlparse
 
 from pydantic import BaseModel, ConfigDict, Field
 
-from .rbac import Role
+from litellm.proxy.auth_v2.authorization import Role
 
 _LOOPBACK_HOSTS = {"localhost", "127.0.0.1", "::1"}
 

litellm/proxy/auth_v2/network.py

@@ -5,8 +5,8 @@ from typing import List, Optional, Tuple
 
 from fastapi import Request
 
-from .config import TrustedProxyConfig
-from .models import NetworkContext
+from litellm.proxy.auth_v2.config import TrustedProxyConfig
+from litellm.proxy.auth_v2.models import NetworkContext
 
 
 def _is_valid_ip(value: str) -> bool:

litellm/proxy/auth_v2/oidc/__init__.py

@@ -1,4 +0,0 @@
-from .config import OIDCProviderConfig
-from .router import build_oidc_router
-
-__all__ = ["OIDCProviderConfig", "build_oidc_router"]

litellm/proxy/auth_v2/oidc/config.py

@@ -1,32 +0,0 @@
-from typing import List, Optional
-
-from pydantic import AnyHttpUrl, BaseModel, Field, SecretStr, field_validator
-
-from ..models import require_secure_url
-
-
-class OIDCProviderConfig(BaseModel):
-    issuer: str
-    audience: List[str]
-    jwks_uri: Optional[AnyHttpUrl] = None
-    algorithms: List[str] = Field(default_factory=lambda: ["RS256"])
-    require_at_jwt: bool = False
-    client_id: Optional[str] = None
-    client_secret: Optional[SecretStr] = None
-    login_scopes: List[str] = Field(
-        default_factory=lambda: ["openid", "email", "profile"]
-    )
-    allowed_roles: List[str] = Field(default_factory=list)
-    allow_platform_roles: bool = False
-
-    @field_validator("issuer")
-    @classmethod
-    def _issuer_https(cls, value: str) -> str:
-        return require_secure_url(value)
-
-    @field_validator("jwks_uri")
-    @classmethod
-    def _jwks_https(cls, value: Optional[AnyHttpUrl]) -> Optional[AnyHttpUrl]:
-        if value is not None:
-            require_secure_url(str(value))
-        return value

litellm/proxy/auth_v2/oidc/router.py

@@ -1,150 +0,0 @@
-from __future__ import annotations
-
-import re
-from typing import TYPE_CHECKING, Any, Dict, cast
-
-from authlib.integrations.starlette_client import OAuth
-from fastapi import APIRouter, HTTPException, Request
-from fastapi.responses import RedirectResponse
-from scim2_models import User as ScimUser
-
-from .config import OIDCProviderConfig
-from ..resolver import ProvisioningStore
-from ..session import safe_relay_state
-
-if TYPE_CHECKING:
-    from ..security import AuthSecurity
-
-_CLAIM_KEYS = ("email", "preferred_username", "name", "groups", "roles")
-
-
-def _provider_key(provider: OIDCProviderConfig) -> str:
-    return re.sub(r"[^a-z0-9]+", "-", provider.issuer.lower()).strip("-")
-
-
-def _user_from_userinfo(userinfo: Dict[str, Any]) -> ScimUser:
-    return ScimUser(
-        external_id=userinfo.get("sub"),
-        user_name=userinfo.get("preferred_username") or userinfo.get("email"),
-        display_name=userinfo.get("name"),
-    )
-
-
-def _mapped_claims(userinfo: Dict[str, Any]) -> Dict[str, Any]:
-    return {key: userinfo[key] for key in _CLAIM_KEYS if userinfo.get(key) is not None}
-
-
-def build_oidc_router(auth: AuthSecurity) -> APIRouter:
-    session = auth.config.session
-    providers = {_provider_key(p): p for p in auth.config.oidc_providers}
-    oauth = OAuth()
-    for provider in auth.config.oidc_providers:
-        oauth.register(
-            name=_provider_key(provider),
-            server_metadata_url=f"{provider.issuer.rstrip('/')}/.well-known/openid-configuration",
-            client_id=provider.client_id,
-            client_secret=(
-                provider.client_secret.get_secret_value()
-                if provider.client_secret
-                else None
-            ),
-            client_kwargs={
-                "scope": " ".join(provider.login_scopes),
-                "code_challenge_method": "S256",
-            },
-        )
-
-    router = APIRouter(prefix="/auth/oidc", tags=["oidc"])
-
-    @router.get("/{provider}/login")
-    async def login(provider: str, request: Request) -> RedirectResponse:
-        client = oauth.create_client(provider)
-        if client is None:
-            raise HTTPException(status_code=404, detail="unknown provider")
-        redirect_uri = str(request.url_for("oidc_callback", provider=provider))
-        relay = safe_relay_state(
-            request.query_params.get("next"), session.default_redirect_path
-        )
-        authorization = await client.create_authorization_url(redirect_uri)
-        txn_id = auth.oauth_txn_store.create_session(
-            {
-                "provider": provider,
-                "state": authorization["state"],
-                "nonce": authorization.get("nonce"),
-                "code_verifier": authorization.get("code_verifier"),
-                "redirect_uri": redirect_uri,
-                "relay": relay,
-            }
-        )
-        response = RedirectResponse(authorization["url"], status_code=303)
-        response.set_cookie(
-            session.login_cookie,
-            txn_id,
-            httponly=True,
-            samesite="lax",
-            secure=session.secure,
-            max_age=session.login_state_ttl,
-        )
-        return response
-
-    @router.get("/{provider}/callback", name="oidc_callback")
-    async def callback(provider: str, request: Request) -> RedirectResponse:
-        client = oauth.create_client(provider)
-        if client is None:
-            raise HTTPException(status_code=404, detail="unknown provider")
-        txn_id = request.cookies.get(session.login_cookie)
-        txn = auth.oauth_txn_store.pop(txn_id) if txn_id else None
-        if txn is None or txn.get("provider") != provider:
-            raise HTTPException(
-                status_code=400, detail="invalid or expired login state"
-            )
-        returned_state = request.query_params.get("state")
-        if not returned_state or returned_state != txn["state"]:
-            raise HTTPException(status_code=400, detail="state mismatch")
-        error = request.query_params.get("error")
-        if error:
-            raise HTTPException(status_code=400, detail=error)
-        code = request.query_params.get("code")
-        if not code:
-            raise HTTPException(status_code=400, detail="missing authorization code")
-
-        token = await client.fetch_access_token(
-            redirect_uri=txn["redirect_uri"],
-            code=code,
-            code_verifier=txn.get("code_verifier"),
-            state=txn["state"],
-        )
-        if token.get("id_token"):
-            userinfo = await client.parse_id_token(token, nonce=txn.get("nonce"))
-        else:
-            userinfo = await client.userinfo(token=token)
-        from ..authenticators import _apply_role_policy
-
-        info = dict(userinfo)
-        provider_config = providers[provider]
-
-        store = cast(ProvisioningStore, auth.resolver)
-        await store.upsert_user(_user_from_userinfo(info))
-
-        claims = _mapped_claims(info)
-        _apply_role_policy(claims, provider_config)
-        session_id = auth.session_store.create_session(
-            {
-                "method": "oidc",
-                "subject": info.get("sub"),
-                "issuer": info.get("iss") or provider_config.issuer,
-                "claims": claims,
-            }
-        )
-        target = safe_relay_state(txn.get("relay"), session.default_redirect_path)
-        response = RedirectResponse(target, status_code=303)
-        response.set_cookie(
-            session.cookie,
-            session_id,
-            httponly=True,
-            samesite="lax",
-            secure=session.secure,
-        )
-        return response
-
-    return router

litellm/proxy/auth_v2/resolvers/__init__.py

@@ -0,0 +1,17 @@
+from litellm.proxy.auth_v2.resolvers.base import (
+    IdentityResolver,
+    IdentityStore,
+    ProvisioningStore,
+)
+from litellm.proxy.auth_v2.resolvers.memory import InMemoryIdentityStore
+
+# DbIdentityStore is intentionally not re-exported here: it pulls in the v1
+# proxy DB machinery (auth_checks, repositories). Import it directly from
+# litellm.proxy.auth_v2.resolvers.db when wiring a database-backed store.
+
+__all__ = [
+    "IdentityResolver",
+    "ProvisioningStore",
+    "IdentityStore",
+    "InMemoryIdentityStore",
+]

litellm/proxy/auth_v2/resolvers/base.py

@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+from typing import List, Optional, Protocol, runtime_checkable
+
+from scim2_models import Group as ScimGroup
+from scim2_models import User as ScimUser
+
+from litellm.proxy.auth_v2.models import Credential, Principal
+
+
+@runtime_checkable
+class IdentityResolver(Protocol):
+    async def resolve(self, credential: Credential) -> Principal: ...
+
+
+@runtime_checkable
+class ProvisioningStore(Protocol):
+    async def upsert_user(self, user: ScimUser) -> ScimUser: ...
+    async def get_user(self, resource_id: str) -> Optional[ScimUser]: ...
+    async def deactivate_user(self, resource_id: str) -> None: ...
+    async def list_users(self, filter_expr: Optional[str]) -> List[ScimUser]: ...
+    async def upsert_group(self, group: ScimGroup) -> ScimGroup: ...
+    async def get_group(self, resource_id: str) -> Optional[ScimGroup]: ...
+    async def delete_group(self, resource_id: str) -> None: ...
+    async def list_groups(self, filter_expr: Optional[str]) -> List[ScimGroup]: ...
+
+
+@runtime_checkable
+class IdentityStore(IdentityResolver, ProvisioningStore, Protocol):
+    """An identity backend: resolves credentials and provisions SCIM users/groups.
+
+    This is the single interface every implementation satisfies (in-memory,
+    database, ...). Resolution and provisioning live behind one store so a
+    provisioned user is immediately resolvable.
+    """

litellm/proxy/auth_v2/resolvers/db.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import uuid
+from typing import TYPE_CHECKING, List, Optional
+
+from scim2_models import Group as ScimGroup
+from scim2_models import User as ScimUser
+
+from litellm.proxy._types import UserAPIKeyAuth, hash_token
+from litellm.proxy.auth.auth_checks import (
+    get_key_object,
+    get_org_object,
+    get_team_object,
+    get_user_object,
+)
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.models import (
+    AuthMethod,
+    Credential,
+    OrganizationIdentity,
+    Principal,
+    PrincipalType,
+    TeamIdentity,
+    TeamRole,
+    UserIdentity,
+)
+from litellm.proxy.auth_v2.resolvers.base import IdentityStore
+from litellm.proxy.auth_v2.resolvers.utils import (
+    db_team_to_scim,
+    db_user_to_scim,
+    map_role,
+    member_role,
+    scim_group_to_db,
+    scim_user_to_db,
+    team_role,
+)
+from litellm.repositories.team_repository import TeamRepository
+from litellm.repositories.user_repository import UserRepository
+
+if TYPE_CHECKING:
+    from litellm.caching.caching import DualCache
+    from litellm.models.user import LiteLLM_UserTable
+    from litellm.proxy.utils import PrismaClient
+
+
+class DbIdentityStore(IdentityStore):
+    """Resolves credentials against the proxy's Prisma tables and provisions
+    SCIM users/groups into ``LiteLLM_UserTable`` / ``LiteLLM_TeamTable``.
+
+    The Prisma client and key cache are injected so this stays a plain object
+    the composition root can build once the proxy DB is connected.
+    """
+
+    def __init__(self, prisma_client: "PrismaClient", cache: "DualCache") -> None:
+        self._prisma = prisma_client
+        self._cache = cache
+
+    # ------------------------------------------------------------------ #
+    # IdentityResolver
+    # ------------------------------------------------------------------ #
+    async def resolve(self, credential: Credential) -> Principal:
+        if credential.method == AuthMethod.API_KEY:
+            return await self._resolve_api_key(credential)
+        if credential.method == AuthMethod.MUTUAL_TLS:
+            return self._service_account(credential)
+        return await self._resolve_subject(credential)
+
+    async def _resolve_api_key(self, credential: Credential) -> Principal:
+        raw = credential.claims.get("_raw_api_key")
+        if not isinstance(raw, str):
+            raise errors.invalid_token()
+        try:
+            key = await get_key_object(hash_token(raw), self._prisma, self._cache)
+        except Exception as exc:
+            raise errors.invalid_token() from exc
+        if key.blocked:
+            raise errors.account_disabled()
+        return self._principal_from_key(credential, key)
+
+    async def _resolve_subject(self, credential: Credential) -> Principal:
+        email = credential.claims.get("email")
+        try:
+            user = await get_user_object(
+                user_id=credential.subject,
+                prisma_client=self._prisma,
+                user_api_key_cache=self._cache,
+                user_id_upsert=False,
+                sso_user_id=credential.subject,
+                user_email=email if isinstance(email, str) else None,
+            )
+        except Exception as exc:
+            raise errors.invalid_token() from exc
+        if user is None:
+            raise errors.invalid_token()
+        return await self._principal_from_user(credential, user)
+
+    def _service_account(self, credential: Credential) -> Principal:
+        return Principal(
+            principal_type=PrincipalType.SERVICE_ACCOUNT,
+            subject=credential.subject,
+            issuer=credential.issuer,
+            audience=list(credential.audience),
+            scopes=list(credential.scopes),
+            auth_method=credential.method,
+            credential_ref=credential.credential_ref,
+        )
+
+    def _principal_from_key(self, credential: Credential, key: UserAPIKeyAuth) -> Principal:
+        teams: List[TeamIdentity] = []
+        if key.team_id is not None:
+            role = team_role(key.team_member.role) if key.team_member else TeamRole.MEMBER
+            teams.append(TeamIdentity(id=key.team_id, name=key.team_alias, role=role))
+        organization = (
+            OrganizationIdentity(id=key.org_id, name=key.organization_alias) if key.org_id is not None else None
+        )
+        user = UserIdentity(id=key.user_id, email=key.user_email) if key.user_id is not None else None
+        mapped = map_role(key.user_role)
+        return Principal(
+            principal_type=(PrincipalType.HUMAN if key.user_id else PrincipalType.SERVICE_ACCOUNT),
+            subject=key.user_id or key.key_alias or credential.subject,
+            issuer=credential.issuer,
+            user=user,
+            organization=organization,
+            teams=teams,
+            roles=[mapped] if mapped else [],
+            scopes=list(credential.scopes),
+            auth_method=credential.method,
+            credential_ref=credential.credential_ref,
+        )
+
+    async def _principal_from_user(self, credential: Credential, user: "LiteLLM_UserTable") -> Principal:
+        teams: List[TeamIdentity] = []
+        for team_id in user.teams or []:
+            try:
+                team = await get_team_object(team_id, self._prisma, self._cache)
+            except Exception:
+                continue
+            teams.append(
+                TeamIdentity(
+                    id=team_id,
+                    name=team.team_alias,
+                    role=member_role(team.members_with_roles, user.user_id),
+                )
+            )
+
+        organization = await self._organization(user)
+        roles = [role for role in (map_role(user.user_role),) if role is not None]
+        return Principal(
+            principal_type=PrincipalType.HUMAN,
+            subject=credential.subject,
+            issuer=credential.issuer,
+            audience=list(credential.audience),
+            user=UserIdentity(
+                id=user.user_id,
+                external_id=user.sso_user_id,
+                email=user.user_email,
+                display_name=user.user_alias,
+            ),
+            organization=organization,
+            teams=teams,
+            roles=roles,
+            scopes=list(credential.scopes),
+            auth_method=credential.method,
+            credential_ref=credential.credential_ref,
+        )
+
+    async def _organization(self, user: "LiteLLM_UserTable") -> Optional[OrganizationIdentity]:
+        if user.organization_id is None:
+            return None
+        try:
+            org = await get_org_object(user.organization_id, self._prisma, self._cache)
+        except Exception:
+            org = None
+        name = org.organization_alias if org is not None else None
+        return OrganizationIdentity(id=user.organization_id, name=name)
+
+    # ------------------------------------------------------------------ #
+    # ProvisioningStore
+    # ------------------------------------------------------------------ #
+    async def upsert_user(self, user: ScimUser) -> ScimUser:
+        repo = UserRepository(self._prisma)
+        data = scim_user_to_db(user)
+        existing = await repo.table.find_unique(where={"user_id": user.id}) if user.id else None
+        if existing is None:
+            data["user_id"] = user.id or str(uuid.uuid4())
+            stored = await repo.table.create(data=data)
+        else:
+            stored = await repo.table.update(where={"user_id": user.id}, data=data)
+        return db_user_to_scim(stored)
+
+    async def get_user(self, resource_id: str) -> Optional[ScimUser]:
+        stored = await UserRepository(self._prisma).table.find_unique(where={"user_id": resource_id})
+        return db_user_to_scim(stored) if stored is not None else None
+
+    async def deactivate_user(self, resource_id: str) -> None:
+        repo = UserRepository(self._prisma)
+        stored = await repo.table.find_unique(where={"user_id": resource_id})
+        if stored is None:
+            return
+        metadata = dict(getattr(stored, "metadata", None) or {})
+        metadata["scim_active"] = False
+        await repo.table.update(where={"user_id": resource_id}, data={"metadata": metadata})
+
+    async def list_users(self, filter_expr: Optional[str]) -> List[ScimUser]:
+        rows = await UserRepository(self._prisma).table.find_many()
+        return [db_user_to_scim(row) for row in rows]
+
+    async def upsert_group(self, group: ScimGroup) -> ScimGroup:
+        repo = TeamRepository(self._prisma)
+        data = scim_group_to_db(group)
+        existing = await repo.table.find_unique(where={"team_id": group.id}) if group.id else None
+        if existing is None:
+            data["team_id"] = group.id or str(uuid.uuid4())
+            stored = await repo.table.create(data=data)
+        else:
+            stored = await repo.table.update(where={"team_id": group.id}, data=data)
+        return db_team_to_scim(stored)
+
+    async def get_group(self, resource_id: str) -> Optional[ScimGroup]:
+        stored = await TeamRepository(self._prisma).table.find_unique(where={"team_id": resource_id})
+        return db_team_to_scim(stored) if stored is not None else None
+
+    async def delete_group(self, resource_id: str) -> None:
+        await TeamRepository(self._prisma).table.delete(where={"team_id": resource_id})
+
+    async def list_groups(self, filter_expr: Optional[str]) -> List[ScimGroup]:
+        rows = await TeamRepository(self._prisma).table.find_many()
+        return [db_team_to_scim(row) for row in rows]

litellm/proxy/auth_v2/resolvers/memory.py

@@ -1,14 +1,13 @@
 from __future__ import annotations
 
-import hashlib
 import uuid
-from typing import Any, Dict, List, Optional, Protocol, runtime_checkable
+from typing import Any, Dict, List, Optional
 
 from scim2_models import Group as ScimGroup
 from scim2_models import User as ScimUser
 
-from . import errors
-from .models import (
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.models import (
     AuthMethod,
     Credential,
     Principal,
@@ -16,43 +15,15 @@ from .models import (
     TeamIdentity,
     UserIdentity,
 )
-from .rbac import Role
+from litellm.proxy.auth_v2.resolvers.base import IdentityStore
+from litellm.proxy.auth_v2.resolvers.utils import (
+    hash_api_key,
+    public_claims,
+    roles_from_claims,
+)
 
 
-@runtime_checkable
-class IdentityResolver(Protocol):
-    async def resolve(self, credential: Credential) -> Principal: ...
-
-
-@runtime_checkable
-class ProvisioningStore(Protocol):
-    async def upsert_user(self, user: ScimUser) -> ScimUser: ...
-    async def get_user(self, resource_id: str) -> Optional[ScimUser]: ...
-    async def deactivate_user(self, resource_id: str) -> None: ...
-    async def list_users(self, filter_expr: Optional[str]) -> List[ScimUser]: ...
-    async def upsert_group(self, group: ScimGroup) -> ScimGroup: ...
-    async def get_group(self, resource_id: str) -> Optional[ScimGroup]: ...
-    async def delete_group(self, resource_id: str) -> None: ...
-    async def list_groups(self, filter_expr: Optional[str]) -> List[ScimGroup]: ...
-
-
-def _hash_api_key(raw: str) -> str:
-    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
-
-
-def _roles_from_claims(claims: Dict[str, Any]) -> List[Role]:
-    raw = claims.get("roles", [])
-    if not isinstance(raw, list):
-        return []
-    valid = {role.value for role in Role}
-    return [Role(value) for value in raw if value in valid]
-
-
-def _public_claims(claims: Dict[str, Any]) -> Dict[str, Any]:
-    return {key: value for key, value in claims.items() if not key.startswith("_")}
-
-
-class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
+class InMemoryIdentityStore(IdentityStore):
     def __init__(
         self,
         api_keys: Optional[Dict[str, Principal]] = None,
@@ -117,7 +88,7 @@ class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
         raw = credential.claims.get("_raw_api_key")
         if not isinstance(raw, str):
             raise errors.invalid_token()
-        principal = self._api_keys.get(_hash_api_key(raw))
+        principal = self._api_keys.get(hash_api_key(raw))
         if principal is None:
             raise errors.invalid_token()
         return principal
@@ -139,7 +110,7 @@ class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
                 scopes=list(credential.scopes),
                 auth_method=credential.method,
                 credential_ref=credential.credential_ref,
-                claims=_public_claims(claims),
+                claims=public_claims(claims),
             )
         return Principal(
             principal_type=PrincipalType.HUMAN,
@@ -154,11 +125,11 @@ class InMemoryIdentityStore(IdentityResolver, ProvisioningStore):
                 display_name=claims.get("name"),
             ),
             teams=self._resolve_teams(claims),
-            roles=_roles_from_claims(claims),
+            roles=roles_from_claims(claims),
             scopes=list(credential.scopes),
             auth_method=credential.method,
             credential_ref=credential.credential_ref,
-            claims=_public_claims(claims),
+            claims=public_claims(claims),
         )
 
     async def upsert_user(self, user: ScimUser) -> ScimUser:

litellm/proxy/auth_v2/resolvers/utils.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+
+import hashlib
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
+
+from scim2_models import Email, Name
+from scim2_models import Group as ScimGroup
+from scim2_models import User as ScimUser
+
+from litellm.proxy.auth_v2.authorization import Role
+from litellm.proxy.auth_v2.models import TeamRole
+
+if TYPE_CHECKING:
+    from litellm.models.team import LiteLLM_TeamTable, Member
+    from litellm.models.user import LiteLLM_UserTable
+
+
+def hash_api_key(raw: str) -> str:
+    return hashlib.sha256(raw.encode("utf-8")).hexdigest()
+
+
+def roles_from_claims(claims: Dict[str, Any]) -> List[Role]:
+    raw = claims.get("roles", [])
+    if not isinstance(raw, list):
+        return []
+    valid = {role.value for role in Role}
+    return [Role(value) for value in raw if value in valid]
+
+
+def public_claims(claims: Dict[str, Any]) -> Dict[str, Any]:
+    return {key: value for key, value in claims.items() if not key.startswith("_")}
+
+
+_ROLE_MAP: Dict[str, Role] = {
+    "proxy_admin": Role.PLATFORM_ADMIN,
+    "proxy_admin_viewer": Role.PLATFORM_VIEWER,
+    "org_admin": Role.ORG_ADMIN,
+}
+
+
+def map_role(value: Optional[object]) -> Optional[Role]:
+    """Map a LiteLLM ``user_role`` (string or LitellmUserRoles) to a platform Role."""
+    if isinstance(value, str):
+        return _ROLE_MAP.get(value)
+    return None
+
+
+def team_role(role: Optional[str]) -> TeamRole:
+    return TeamRole.ADMIN if role == "admin" else TeamRole.MEMBER
+
+
+def member_role(members: "List[Member]", user_id: Optional[str]) -> TeamRole:
+    if user_id is not None:
+        for member in members:
+            if member.user_id == user_id:
+                return team_role(member.role)
+    return TeamRole.MEMBER
+
+
+def scim_user_to_db(user: ScimUser) -> Dict[str, object]:
+    email = user.emails[0].value if user.emails else None
+    metadata: Dict[str, object] = {"scim_active": user.active}
+    if user.name is not None:
+        metadata["scim_metadata"] = {
+            "givenName": user.name.given_name,
+            "familyName": user.name.family_name,
+        }
+    data: Dict[str, object] = {"metadata": metadata}
+    if email is not None:
+        data["user_email"] = email
+    if user.external_id is not None:
+        data["sso_user_id"] = user.external_id
+    if user.display_name is not None:
+        data["user_alias"] = user.display_name
+    return data
+
+
+def db_user_to_scim(user: "LiteLLM_UserTable") -> ScimUser:
+    metadata = getattr(user, "metadata", None) or {}
+    scim_name = metadata.get("scim_metadata") or {}
+    result = ScimUser(
+        external_id=user.sso_user_id or user.user_id,
+        user_name=user.user_email or user.user_id,
+        display_name=user.user_alias,
+        active=metadata.get("scim_active", True),
+    )
+    result.id = user.user_id
+    if user.user_email:
+        result.emails = [Email(value=user.user_email, primary=True)]
+    if scim_name.get("givenName") or scim_name.get("familyName"):
+        result.name = Name(
+            given_name=scim_name.get("givenName"),
+            family_name=scim_name.get("familyName"),
+        )
+    return result
+
+
+def scim_group_to_db(group: ScimGroup) -> Dict[str, object]:
+    members = [{"user_id": member.value, "role": "user"} for member in (group.members or [])]
+    return {"team_alias": group.display_name, "members_with_roles": members}
+
+
+def db_team_to_scim(team: "LiteLLM_TeamTable") -> ScimGroup:
+    result = ScimGroup(display_name=team.team_alias or team.team_id)
+    result.id = team.team_id
+    return result

litellm/proxy/auth_v2/saml/__init__.py

@@ -1,4 +0,0 @@
-from .config import SAMLConfig
-from .router import build_saml_router
-
-__all__ = ["SAMLConfig", "build_saml_router"]

litellm/proxy/auth_v2/saml/config.py

@@ -1,40 +0,0 @@
-from typing import Dict, List, Optional
-
-from pydantic import BaseModel, Field, model_validator
-
-DEFAULT_SAML_ATTRIBUTE_MAP = {
-    "email": "email",
-    "mail": "email",
-    "givenName": "given_name",
-    "surname": "family_name",
-    "sn": "family_name",
-    "displayName": "display_name",
-    "userName": "user_name",
-    "uid": "user_name",
-    "groups": "groups",
-    "roles": "roles",
-}
-
-
-class SAMLConfig(BaseModel):
-    enabled: bool = False
-    entity_id: str
-    acs_url: str
-    idp_metadata: str = ""
-    sp_key_file: Optional[str] = None
-    sp_cert_file: Optional[str] = None
-    allow_unsolicited: bool = False
-    xmlsec_binary: Optional[str] = None
-    attribute_map: Dict[str, str] = Field(
-        default_factory=lambda: dict(DEFAULT_SAML_ATTRIBUTE_MAP)
-    )
-    allowed_roles: List[str] = Field(default_factory=list)
-    allow_platform_roles: bool = False
-
-    @model_validator(mode="after")
-    def _require_idp_metadata(self) -> "SAMLConfig":
-        if self.enabled and not self.idp_metadata.strip():
-            raise ValueError(
-                "SAML enabled but idp_metadata is empty (inline XML, local path, or URL)"
-            )
-        return self

litellm/proxy/auth_v2/scim/__init__.py

@@ -1,3 +0,0 @@
-from .router import build_scim_router
-
-__all__ = ["build_scim_router"]

litellm/proxy/auth_v2/scim/router.py

@@ -1,306 +0,0 @@
-from __future__ import annotations
-
-from typing import (
-    TYPE_CHECKING,
-    Any,
-    Callable,
-    Coroutine,
-    Dict,
-    Optional,
-    Type,
-    TypeVar,
-    cast,
-)
-
-from fastapi import APIRouter, HTTPException, Query, Request, Response, Security, status
-from fastapi.responses import JSONResponse
-from fastapi.routing import APIRoute
-from pydantic import ValidationError
-from scim2_models import (
-    Bulk,
-    ChangePassword,
-    Context,
-    Error,
-    Filter,
-    Group,
-    ListResponse,
-    Patch,
-    PatchOp,
-    Resource,
-    ResourceType,
-    Schema,
-    ServiceProviderConfig,
-    Sort,
-    User,
-)
-
-from ..resolver import ProvisioningStore
-
-if TYPE_CHECKING:
-    from ..security import AuthSecurity
-
-R = TypeVar("R", bound=Resource)
-
-
-def _error(status_code: int, detail: str) -> JSONResponse:
-    return JSONResponse(
-        status_code=status_code,
-        content=Error(status=str(status_code), detail=detail).model_dump(),
-    )
-
-
-class _ScimRoute(APIRoute):
-    """Render authentication failures with the SCIM Error schema (RFC 7644)."""
-
-    def get_route_handler(  # type: ignore[override]
-        self,
-    ) -> Callable[[Request], Coroutine[Any, Any, Response]]:
-        handler = super().get_route_handler()
-
-        async def scim_handler(request: Request) -> Response:
-            try:
-                return await handler(request)
-            except HTTPException as exc:
-                if exc.status_code not in (
-                    status.HTTP_401_UNAUTHORIZED,
-                    status.HTTP_403_FORBIDDEN,
-                ):
-                    raise
-                response = _error(exc.status_code, str(exc.detail))
-                if exc.headers:
-                    response.headers.update(exc.headers)
-                return response
-
-        return scim_handler
-
-
-async def _parse(request: Request, model: Type[R]) -> R:
-    body = await request.json()
-    return model.model_validate(body, scim_ctx=Context.RESOURCE_CREATION_REQUEST)
-
-
-def _set_path(data: Dict[str, Any], path: str, value: Any) -> None:
-    keys = path.split(".")
-    node = data
-    for key in keys[:-1]:
-        child = node.get(key)
-        if not isinstance(child, dict):
-            child = {}
-            node[key] = child
-        node = child
-    node[keys[-1]] = value
-
-
-def _remove_path(data: Dict[str, Any], path: str) -> None:
-    keys = path.split(".")
-    node = data
-    for key in keys[:-1]:
-        child = node.get(key)
-        if not isinstance(child, dict):
-            return
-        node = child
-    node.pop(keys[-1], None)
-
-
-def _targets_read_only_id(op: Any) -> bool:
-    if op.path is not None:
-        return op.path.split(".")[0].strip().lower() == "id"
-    return isinstance(op.value, dict) and any(str(k).lower() == "id" for k in op.value)
-
-
-def _apply_patch(resource: R, patch: PatchOp) -> R:
-    data: Dict[str, Any] = resource.model_dump()
-    for op in patch.operations:
-        action = op.op.value if hasattr(op.op, "value") else str(op.op)
-        if op.path is not None and ("[" in op.path or "]" in op.path):
-            raise ValueError(f"unsupported SCIM patch path filter: {op.path}")
-        if _targets_read_only_id(op):
-            raise ValueError("the SCIM id attribute is read-only")
-        if action == "remove":
-            if op.path:
-                _remove_path(data, op.path)
-            continue
-        if op.path is None and isinstance(op.value, dict):
-            data.update(op.value)
-        elif op.path is not None:
-            _set_path(data, op.path, op.value)
-    return type(resource).model_validate(data)
-
-
-def _dump(resource: Resource, ctx: Context) -> Dict[str, Any]:
-    return resource.model_dump(scim_ctx=ctx)
-
-
-def _build_protected_router(auth: AuthSecurity) -> APIRouter:
-    store = cast(ProvisioningStore, auth.resolver)
-    protected = APIRouter(
-        route_class=_ScimRoute,
-        dependencies=[Security(auth.principal, scopes=["scim:write"])],
-    )
-
-    @protected.post("/Users", status_code=status.HTTP_201_CREATED)
-    async def create_user(request: Request) -> Response:
-        try:
-            user = await _parse(request, User)
-        except ValidationError as exc:
-            return _error(status.HTTP_400_BAD_REQUEST, str(exc))
-        stored = await store.upsert_user(user)
-        return JSONResponse(
-            status_code=status.HTTP_201_CREATED,
-            content=_dump(stored, Context.RESOURCE_CREATION_RESPONSE),
-        )
-
-    @protected.get("/Users/{resource_id}")
-    async def get_user(resource_id: str) -> Response:
-        user = await store.get_user(resource_id)
-        if user is None:
-            return _error(status.HTTP_404_NOT_FOUND, f"User {resource_id} not found")
-        return JSONResponse(content=_dump(user, Context.RESOURCE_QUERY_RESPONSE))
-
-    @protected.patch("/Users/{resource_id}")
-    async def patch_user(resource_id: str, request: Request) -> Response:
-        user = await store.get_user(resource_id)
-        if user is None:
-            return _error(status.HTTP_404_NOT_FOUND, f"User {resource_id} not found")
-        try:
-            patch = PatchOp[User].model_validate(await request.json())
-            patched = _apply_patch(user, patch)
-        except (ValidationError, ValueError) as exc:
-            return _error(status.HTTP_400_BAD_REQUEST, str(exc))
-        updated = await store.upsert_user(patched)
-        return JSONResponse(content=_dump(updated, Context.RESOURCE_PATCH_RESPONSE))
-
-    @protected.delete("/Users/{resource_id}", status_code=status.HTTP_204_NO_CONTENT)
-    async def deactivate_user(resource_id: str) -> Response:
-        if await store.get_user(resource_id) is None:
-            return _error(status.HTTP_404_NOT_FOUND, f"User {resource_id} not found")
-        await store.deactivate_user(resource_id)
-        return Response(status_code=status.HTTP_204_NO_CONTENT)
-
-    @protected.get("/Users")
-    async def list_users(
-        filter_expr: Optional[str] = Query(default=None, alias="filter"),
-    ) -> Response:
-        users = await store.list_users(filter_expr)
-        listing: ListResponse[User] = ListResponse[User](
-            total_results=len(users),
-            start_index=1,
-            items_per_page=len(users),
-            resources=users or None,
-        )
-        return JSONResponse(content=_dump(listing, Context.RESOURCE_QUERY_RESPONSE))
-
-    @protected.post("/Groups", status_code=status.HTTP_201_CREATED)
-    async def create_group(request: Request) -> Response:
-        try:
-            group = await _parse(request, Group)
-        except ValidationError as exc:
-            return _error(status.HTTP_400_BAD_REQUEST, str(exc))
-        stored = await store.upsert_group(group)
-        return JSONResponse(
-            status_code=status.HTTP_201_CREATED,
-            content=_dump(stored, Context.RESOURCE_CREATION_RESPONSE),
-        )
-
-    @protected.get("/Groups/{resource_id}")
-    async def get_group(resource_id: str) -> Response:
-        group = await store.get_group(resource_id)
-        if group is None:
-            return _error(status.HTTP_404_NOT_FOUND, f"Group {resource_id} not found")
-        return JSONResponse(content=_dump(group, Context.RESOURCE_QUERY_RESPONSE))
-
-    @protected.patch("/Groups/{resource_id}")
-    async def patch_group(resource_id: str, request: Request) -> Response:
-        group = await store.get_group(resource_id)
-        if group is None:
-            return _error(status.HTTP_404_NOT_FOUND, f"Group {resource_id} not found")
-        try:
-            patch = PatchOp[Group].model_validate(await request.json())
-            patched = _apply_patch(group, patch)
-        except (ValidationError, ValueError) as exc:
-            return _error(status.HTTP_400_BAD_REQUEST, str(exc))
-        updated = await store.upsert_group(patched)
-        return JSONResponse(content=_dump(updated, Context.RESOURCE_PATCH_RESPONSE))
-
-    @protected.delete("/Groups/{resource_id}", status_code=status.HTTP_204_NO_CONTENT)
-    async def delete_group(resource_id: str) -> Response:
-        if await store.get_group(resource_id) is None:
-            return _error(status.HTTP_404_NOT_FOUND, f"Group {resource_id} not found")
-        await store.delete_group(resource_id)
-        return Response(status_code=status.HTTP_204_NO_CONTENT)
-
-    @protected.get("/Groups")
-    async def list_groups(
-        filter_expr: Optional[str] = Query(default=None, alias="filter"),
-    ) -> Response:
-        groups = await store.list_groups(filter_expr)
-        listing: ListResponse[Group] = ListResponse[Group](
-            total_results=len(groups),
-            start_index=1,
-            items_per_page=len(groups),
-            resources=groups or None,
-        )
-        return JSONResponse(content=_dump(listing, Context.RESOURCE_QUERY_RESPONSE))
-
-    return protected
-
-
-def _build_discovery_router() -> APIRouter:
-    router = APIRouter()
-
-    @router.get("/ServiceProviderConfig")
-    async def service_provider_config() -> Response:
-        config = ServiceProviderConfig(
-            patch=Patch(supported=True),
-            bulk=Bulk(supported=False, max_operations=0, max_payload_size=0),
-            filter=Filter(supported=False, max_results=0),
-            change_password=ChangePassword(supported=False),
-            sort=Sort(supported=False),
-            etag=None,
-            authentication_schemes=[],
-        )
-        return JSONResponse(content=config.model_dump())
-
-    @router.get("/ResourceTypes")
-    async def resource_types() -> Response:
-        types = [
-            ResourceType(
-                id="User",
-                name="User",
-                endpoint="/Users",
-                schema="urn:ietf:params:scim:schemas:core:2.0:User",
-            ),
-            ResourceType(
-                id="Group",
-                name="Group",
-                endpoint="/Groups",
-                schema="urn:ietf:params:scim:schemas:core:2.0:Group",
-            ),
-        ]
-        listing: ListResponse[ResourceType] = ListResponse[ResourceType](
-            total_results=len(types),
-            start_index=1,
-            items_per_page=len(types),
-            resources=types,
-        )
-        return JSONResponse(content=listing.model_dump())
-
-    @router.get("/Schemas")
-    async def schemas() -> Response:
-        resources = [User.to_schema(), Group.to_schema()]
-        listing: ListResponse[Schema] = ListResponse[Schema](
-            total_results=len(resources),
-            start_index=1,
-            items_per_page=len(resources),
-            resources=resources,
-        )
-        return JSONResponse(content=listing.model_dump())
-
-    return router
-
-
-def build_scim_router(auth: AuthSecurity) -> APIRouter:
-    router = APIRouter(prefix="/scim/v2", tags=["scim"])
-    router.include_router(_build_protected_router(auth))
-    router.include_router(_build_discovery_router())
-    return router

litellm/proxy/auth_v2/security.py

@@ -3,18 +3,25 @@ from typing import Annotated, Callable, List, Optional
 from fastapi import Request, Security
 from fastapi.security import SecurityScopes
 
-from . import errors
-from .authenticators import (
+from litellm.proxy.auth_v2 import errors
+from litellm.proxy.auth_v2.authenticators import (
     Authenticator,
     BasicAuthVerifier,
     build_authenticators,
 )
-from .config import AuthConfig
-from .models import Principal
-from .network import resolve_network_context
-from .rbac import RBACEngine, Role, has_required_scopes
-from .resolver import IdentityResolver
-from .session import SessionAuthenticator, SessionStore
+from litellm.proxy.auth_v2.config import AuthConfig
+from litellm.proxy.auth_v2.models import Principal
+from litellm.proxy.auth_v2.network import resolve_network_context
+from litellm.proxy.auth_v2.authorization import (
+    Authorizer,
+    RBACEngine,
+    Role,
+    has_required_scopes,
+)
+from litellm.proxy.auth_v2.authenticators.session import SessionAuthenticator
+from litellm.proxy.auth_v2.resolvers import IdentityResolver
+from litellm.proxy.auth_v2.sessions import StateBackend, StateStore
+from litellm.proxy.auth_v2.sessions.schemas import OAuthTransaction, SessionState
 
 
 def _combined_challenge(authenticators: List[Authenticator]) -> str:
@@ -30,9 +37,9 @@ class AuthSecurity:
     """Enforcement layer consumed purely through FastAPI ``Security()``.
 
     Construct once at the composition root and pass the bound methods
-    (``principal``, ``require_roles``, ``require_permission``) to ``Security()``;
-    routers receive the instance explicitly via ``build_*_router(auth)``. There is
-    no app mutation and no ``app.state``.
+    (``principal``, ``require_roles``, ``require_permission``) to ``Security()``.
+    Routers reach this instance at request time via ``request.app.state.auth_v2``
+    (see ``routers/dependencies.py``), so assign it there when wiring the app.
 
     Deployment note for trusted-proxy IP resolution: uvicorn's ``--proxy-headers``
     (on by default) overwrites ``request.client`` from ``X-Forwarded-For`` before
@@ -46,18 +53,20 @@ class AuthSecurity:
         self,
         config: AuthConfig,
         resolver: IdentityResolver,
-        rbac: Optional[RBACEngine] = None,
+        authorizer: Optional[Authorizer] = None,
         authenticators: Optional[List[Authenticator]] = None,
         basic_verifier: Optional[BasicAuthVerifier] = None,
+        state_backend: Optional[StateBackend] = None,
     ) -> None:
         self.config = config
         self.resolver = resolver
-        self.rbac = rbac or RBACEngine(config.casbin_policy_path)
-        self.session_store = SessionStore(
-            config.session.ttl_seconds, config.session.max_size
+        self.authorizer = authorizer or RBACEngine(config.casbin_policy_path)
+        self._state = state_backend or StateBackend(None)
+        self.session_store: StateStore[SessionState] = self._state.store(
+            "sessions", default_ttl=config.session.ttl_seconds
         )
-        self.oauth_txn_store = SessionStore(
-            config.session.login_state_ttl, config.session.max_size
+        self.oauth_txn_store: StateStore[OAuthTransaction] = self._state.store(
+            "oauth_txn", default_ttl=config.session.login_state_ttl
         )
         chain = (
             list(authenticators)
@@ -67,9 +76,7 @@ class AuthSecurity:
         chain.append(SessionAuthenticator(config.session.cookie, self.session_store))
         self.authenticators = chain
 
-    async def principal(
-        self, security_scopes: SecurityScopes, request: Request
-    ) -> Principal:
+    async def principal(self, security_scopes: SecurityScopes, request: Request) -> Principal:
         """Resolve the caller to a Principal, enforcing scheme OR and required scopes."""
         credential = None
         for authenticator in self.authenticators:
@@ -80,9 +87,7 @@ class AuthSecurity:
             raise errors.unauthenticated(_combined_challenge(self.authenticators))
 
         resolved = await self.resolver.resolve(credential)
-        principal = resolved.model_copy(
-            update={"network": resolve_network_context(request, self.config.network)}
-        )
+        principal = resolved.model_copy(update={"network": resolve_network_context(request, self.config.network)})
         if not has_required_scopes(security_scopes, principal):
             raise errors.insufficient_scope()
         return principal
@@ -93,7 +98,7 @@ class AuthSecurity:
         async def dependency(
             principal: Annotated[Principal, Security(self.principal)],
         ) -> Principal:
-            if not self.rbac.has_any_role(principal, allowed):
+            if not self.authorizer.has_any_role(principal, allowed):
                 raise errors.forbidden_role()
             return principal
 
@@ -105,7 +110,7 @@ class AuthSecurity:
         async def dependency(
             principal: Annotated[Principal, Security(self.principal)],
         ) -> Principal:
-            if not self.rbac.enforce(principal, obj, act):
+            if not self.authorizer.enforce(principal, obj, act):
                 raise errors.forbidden_permission()
             return principal
 

litellm/proxy/auth_v2/session.py

@@ -1,99 +0,0 @@
-from __future__ import annotations
-
-import secrets
-import time
-from typing import Any, Dict, Optional, Tuple
-
-from fastapi import Request
-from pydantic import BaseModel
-
-from .models import AuthMethod, Credential, CredentialRef, SecuritySchemeType
-
-
-class SessionConfig(BaseModel):
-    cookie: str = "litellm_session"
-    secure: bool = True
-    ttl_seconds: int = 3600
-    max_size: int = 10000
-    default_redirect_path: str = "/"
-    login_cookie: str = "litellm_oidc_txn"
-    login_state_ttl: int = 300
-
-
-def safe_relay_state(target: Optional[str], default: str) -> str:
-    if (
-        target
-        and target.startswith("/")
-        and not target.startswith("//")
-        and "://" not in target
-        and "\\" not in target
-    ):
-        return target
-    return default
-
-
-class SessionStore:
-    def __init__(self, ttl_seconds: int = 3600, max_size: int = 10000) -> None:
-        self._sessions: Dict[str, Tuple[float, Dict[str, Any]]] = {}
-        self._ttl = ttl_seconds
-        self._max_size = max_size
-
-    def create_session(self, identity: Dict[str, Any]) -> str:
-        now = time.time()
-        self._evict(now)
-        session_id = secrets.token_urlsafe(32)
-        self._sessions[session_id] = (now + self._ttl, identity)
-        return session_id
-
-    def get(self, session_id: str) -> Optional[Dict[str, Any]]:
-        entry = self._sessions.get(session_id)
-        if entry is None:
-            return None
-        expires_at, identity = entry
-        if expires_at < time.time():
-            self._sessions.pop(session_id, None)
-            return None
-        return identity
-
-    def pop(self, session_id: str) -> Optional[Dict[str, Any]]:
-        entry = self._sessions.pop(session_id, None)
-        if entry is None:
-            return None
-        expires_at, identity = entry
-        if expires_at < time.time():
-            return None
-        return identity
-
-    def _evict(self, now: float) -> None:
-        for key in [k for k, (exp, _) in self._sessions.items() if exp < now]:
-            self._sessions.pop(key, None)
-        overflow = len(self._sessions) - self._max_size + 1
-        if overflow > 0:
-            oldest = sorted(self._sessions, key=lambda k: self._sessions[k][0])
-            for key in oldest[:overflow]:
-                self._sessions.pop(key, None)
-
-
-class SessionAuthenticator:
-    def __init__(self, cookie_name: str, store: SessionStore) -> None:
-        self._cookie_name = cookie_name
-        self._store = store
-
-    async def authenticate(self, request: Request) -> Optional[Credential]:
-        session_id = request.cookies.get(self._cookie_name)
-        if not session_id:
-            return None
-        identity = self._store.get(session_id)
-        if identity is None:
-            return None
-        return Credential(
-            scheme=SecuritySchemeType.API_KEY,
-            method=AuthMethod(identity["method"]),
-            subject=identity["subject"],
-            issuer=identity.get("issuer"),
-            claims=identity.get("claims", {}),
-            credential_ref=CredentialRef(token_id=session_id),
-        )
-
-    def challenge(self) -> str:
-        return ""

litellm/proxy/auth_v2/sessions/__init__.py

@@ -0,0 +1,15 @@
+from litellm.proxy.auth_v2.sessions.factory import StateBackend
+from litellm.proxy.auth_v2.sessions.base import StateStore, StateValue
+from litellm.proxy.auth_v2.sessions.memory import InMemoryStateStore
+from litellm.proxy.auth_v2.sessions.redis import RedisStateStore
+from litellm.proxy.auth_v2.sessions.schemas import OAuthTransaction, SessionState
+
+__all__ = [
+    "StateBackend",
+    "StateStore",
+    "StateValue",
+    "InMemoryStateStore",
+    "RedisStateStore",
+    "SessionState",
+    "OAuthTransaction",
+]

litellm/proxy/auth_v2/sessions/base.py

@@ -0,0 +1,31 @@
+from __future__ import annotations
+
+from typing import Any, Mapping, Optional, Protocol, TypeVar, runtime_checkable
+
+StateValue = TypeVar("StateValue", bound=Mapping[str, Any])
+
+
+@runtime_checkable
+class StateStore(Protocol[StateValue]):
+    """Async key/value store with per-key TTL, generic over its value schema.
+
+    Backs short-lived auth state. Each store is parameterized by the typed
+    payload it holds (see ``schemas``) and namespaced by the backend that hands
+    it out, so several stores can share one Redis instance without colliding.
+    """
+
+    async def get(self, key: str) -> Optional[StateValue]: ...
+
+    async def set(self, key: str, value: StateValue, ttl_seconds: Optional[int] = None) -> None: ...
+
+    async def pop(self, key: str) -> Optional[StateValue]: ...
+
+    async def delete(self, key: str) -> None: ...
+
+    async def add_if_absent(self, key: str, ttl_seconds: Optional[int] = None) -> bool:
+        """Set a marker only if the key is absent; return True iff newly set.
+
+        Atomic and value-free. Use for one-time guards like SAML assertion
+        replay detection, where only key presence matters.
+        """
+        ...

litellm/proxy/auth_v2/sessions/factory.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import logging
+import os
+from typing import TYPE_CHECKING, Optional, cast
+
+from litellm._redis import get_redis_async_client
+
+from litellm.proxy.auth_v2.sessions.base import StateStore, StateValue
+from litellm.proxy.auth_v2.sessions.memory import InMemoryStateStore
+from litellm.proxy.auth_v2.sessions.redis import RedisStateStore
+
+if TYPE_CHECKING:
+    from redis.asyncio import Redis
+
+logger = logging.getLogger("litellm.proxy.auth_v2.sessions")
+
+_REDIS_ENV_SIGNALS = (
+    "REDIS_URL",
+    "REDIS_HOST",
+    "REDIS_CLUSTER_NODES",
+    "REDIS_SENTINEL_NODES",
+)
+
+
+async def _reachable(client: "Redis") -> bool:
+    try:
+        return bool(await client.ping())
+    except Exception:
+        return False
+
+
+def _default_redis_client() -> Optional["Redis"]:
+    if not any(os.getenv(signal) for signal in _REDIS_ENV_SIGNALS):
+        return None
+    try:
+        return cast("Redis", get_redis_async_client())
+    except Exception:
+        logger.warning("auth_v2 state layer could not build a Redis client", exc_info=True)
+        return None
+
+
+class StateBackend:
+    """Hands out namespaced state stores backed by Redis when reachable, else memory.
+
+    The Redis-vs-memory choice is made once, at ``connect`` time, and held for the
+    backend's lifetime. We deliberately do not fail over per operation: silently
+    moving a live session from Redis to a local dict would strand it on one worker
+    and lose it the moment another worker serves the next request.
+
+    Inject the client for tests or to share the proxy's existing connection; the
+    default builder only fires when Redis is configured via the environment.
+    """
+
+    def __init__(self, redis_client: Optional["Redis"]) -> None:
+        self._redis = redis_client
+
+    @classmethod
+    async def connect(cls, redis_client: Optional["Redis"] = None) -> "StateBackend":
+        client = redis_client if redis_client is not None else _default_redis_client()
+        if client is not None and await _reachable(client):
+            logger.info("auth_v2 state layer using Redis backend")
+            return cls(client)
+        logger.info("auth_v2 state layer using in-memory backend")
+        return cls(None)
+
+    @property
+    def using_redis(self) -> bool:
+        return self._redis is not None
+
+    def store(self, namespace: str, *, default_ttl: int) -> StateStore[StateValue]:
+        """Return a typed store for ``namespace``.
+
+        The value schema is taken from the call site's annotation, e.g.
+        ``sessions: StateStore[SessionState] = backend.store("sessions", default_ttl=3600)``.
+        """
+        if self._redis is not None:
+            return RedisStateStore(self._redis, namespace, default_ttl)
+        return InMemoryStateStore(namespace, default_ttl)

litellm/proxy/auth_v2/sessions/memory.py

@@ -0,0 +1,66 @@
+from __future__ import annotations
+
+import time
+from typing import Dict, Generic, Optional, Tuple
+
+from litellm.proxy.auth_v2.sessions.base import StateValue
+
+
+class InMemoryStateStore(Generic[StateValue]):
+    """Process-local fallback when Redis is unavailable. Single-process only."""
+
+    def __init__(self, namespace: str, default_ttl: int, max_size: int = 10000) -> None:
+        self._namespace = namespace
+        self._default_ttl = default_ttl
+        self._max_size = max_size
+        self._entries: Dict[str, Tuple[float, Optional[StateValue]]] = {}
+
+    def _key(self, key: str) -> str:
+        return f"{self._namespace}:{key}"
+
+    def _expiry(self, ttl_seconds: Optional[int]) -> float:
+        return time.time() + (self._default_ttl if ttl_seconds is None else ttl_seconds)
+
+    def _live(self, key: str, now: float) -> Optional[Tuple[float, Optional[StateValue]]]:
+        entry = self._entries.get(key)
+        if entry is None:
+            return None
+        if entry[0] < now:
+            self._entries.pop(key, None)
+            return None
+        return entry
+
+    async def get(self, key: str) -> Optional[StateValue]:
+        entry = self._live(self._key(key), time.time())
+        return entry[1] if entry is not None else None
+
+    async def set(self, key: str, value: StateValue, ttl_seconds: Optional[int] = None) -> None:
+        self._evict(time.time())
+        self._entries[self._key(key)] = (self._expiry(ttl_seconds), value)
+
+    async def pop(self, key: str) -> Optional[StateValue]:
+        entry = self._entries.pop(self._key(key), None)
+        if entry is None or entry[0] < time.time():
+            return None
+        return entry[1]
+
+    async def delete(self, key: str) -> None:
+        self._entries.pop(self._key(key), None)
+
+    async def add_if_absent(self, key: str, ttl_seconds: Optional[int] = None) -> bool:
+        now = time.time()
+        self._evict(now)
+        namespaced = self._key(key)
+        if self._live(namespaced, now) is not None:
+            return False
+        self._entries[namespaced] = (self._expiry(ttl_seconds), None)
+        return True
+
+    def _evict(self, now: float) -> None:
+        for key in [k for k, (exp, _) in self._entries.items() if exp < now]:
+            self._entries.pop(key, None)
+        overflow = len(self._entries) - self._max_size + 1
+        if overflow > 0:
+            oldest = sorted(self._entries, key=lambda k: self._entries[k][0])
+            for key in oldest[:overflow]:
+                self._entries.pop(key, None)

litellm/proxy/auth_v2/sessions/redis.py

@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import json
+from typing import TYPE_CHECKING, Generic, Optional, cast
+
+from litellm.proxy.auth_v2.sessions.base import StateValue
+
+if TYPE_CHECKING:
+    from redis.asyncio import Redis
+
+
+class RedisStateStore(Generic[StateValue]):
+    """Redis-backed store. Shared across workers; Redis enforces the TTL."""
+
+    def __init__(self, client: "Redis", namespace: str, default_ttl: int) -> None:
+        self._client = client
+        self._namespace = namespace
+        self._default_ttl = default_ttl
+
+    def _key(self, key: str) -> str:
+        return f"{self._namespace}:{key}"
+
+    def _ttl(self, ttl_seconds: Optional[int]) -> int:
+        return self._default_ttl if ttl_seconds is None else ttl_seconds
+
+    async def get(self, key: str) -> Optional[StateValue]:
+        raw = await self._client.get(self._key(key))
+        return cast(StateValue, json.loads(raw)) if raw is not None else None
+
+    async def set(self, key: str, value: StateValue, ttl_seconds: Optional[int] = None) -> None:
+        await self._client.set(self._key(key), json.dumps(value), ex=self._ttl(ttl_seconds))
+
+    async def pop(self, key: str) -> Optional[StateValue]:
+        raw = await self._client.getdel(self._key(key))
+        return cast(StateValue, json.loads(raw)) if raw is not None else None
+
+    async def delete(self, key: str) -> None:
+        await self._client.delete(self._key(key))
+
+    async def add_if_absent(self, key: str, ttl_seconds: Optional[int] = None) -> bool:
+        added = await self._client.set(self._key(key), "1", nx=True, ex=self._ttl(ttl_seconds))
+        return bool(added)

litellm/proxy/auth_v2/sessions/schemas.py

@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+from typing import Any, Dict, Optional, TypedDict
+
+
+class SessionState(TypedDict):
+    """A logged-in session, keyed by the session cookie's opaque id."""
+
+    method: str
+    subject: str
+    issuer: Optional[str]
+    claims: Dict[str, Any]
+
+
+class OAuthTransaction(TypedDict):
+    """In-flight OIDC authorization-code login, keyed by the login cookie's id."""
+
+    provider: str
+    state: str
+    nonce: Optional[str]
+    code_verifier: Optional[str]
+    redirect_uri: str
+    relay: str