"""
Has all /sso/* routes

/sso/key/generate - handles user signing in with SSO and redirects to /sso/callback
/sso/callback - returns JWT Redirect Response that redirects to LiteLLM UI

/sso/debug/login - handles user signing in with SSO and redirects to /sso/debug/callback
/sso/debug/callback - returns the OpenID object returned by the SSO provider
"""

import asyncio
import base64
import hashlib
import inspect
import json
import os
import re
import secrets
from collections.abc import Mapping, Sequence
from copy import deepcopy
from html import escape
from types import MappingProxyType
from typing import (
    TYPE_CHECKING,
    Annotated,
    Any,
    Final,
    Literal,
    NoReturn,
    Optional,
    Protocol,
    Union,
    cast,
    overload,
)
from urllib.parse import parse_qs, urlencode, urlparse

if TYPE_CHECKING:
    import httpx

import jwt
from fastapi import APIRouter, Depends, Header, HTTPException, Request, Response, status
from fastapi.responses import RedirectResponse
from pydantic import BaseModel, BeforeValidator, ConfigDict, TypeAdapter, ValidationError

import litellm
from litellm._logging import verbose_proxy_logger
from litellm._uuid import uuid
from litellm.caching.dual_cache import DualCache
from litellm.constants import (
    CLI_SSO_CLAIM_MAP,
    CLI_SSO_CLAIM_MAX_SCALAR_LENGTH,
    CLI_SSO_SESSION_CACHE_KEY_PREFIX,
    CLI_SSO_SESSION_TTL_SECONDS,
    LITELLM_CLI_SOURCE_IDENTIFIER,
    LITELLM_UI_SESSION_DURATION,
    MAX_SPENDLOG_ROWS_TO_QUERY,
    MICROSOFT_USER_DISPLAY_NAME_ATTRIBUTE,
    MICROSOFT_USER_EMAIL_ATTRIBUTE,
    MICROSOFT_USER_FIRST_NAME_ATTRIBUTE,
    MICROSOFT_USER_ID_ATTRIBUTE,
    MICROSOFT_USER_LAST_NAME_ATTRIBUTE,
)
from litellm.litellm_core_utils.dot_notation_indexing import get_nested_value
from litellm.llms.custom_httpx.http_handler import (
    AsyncHTTPHandler,
    get_async_httpx_client,
    httpxSpecialProvider,
)
from litellm.proxy._experimental.mcp_server.outbound_credentials.sso_assertion_store import (
    SSOIdentityAssertion,
    assertion_from_sso_login,
    retain_sso_identity_assertion_for_ema,
)
from litellm.proxy._types import (
    CommonProxyErrors,
    LiteLLM_UserTable,
    LitellmUserRoles,
    Member,
    NewTeamRequest,
    NewUserRequest,
    NewUserResponse,
    ProxyErrorTypes,
    ProxyException,
    SSOUserDefinedValues,
    TeamMemberAddRequest,
    UserAPIKeyAuth,
)
from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken, get_user_object
from litellm.proxy.auth.auth_utils import (
    _get_request_ip_address,
    has_user_setup_sso,
)
from litellm.proxy.auth.handle_jwt import JWTHandler
from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
from litellm.proxy.common_utils.admin_ui_utils import (
    admin_ui_disabled,
    show_missing_vars_in_env,
)
from litellm.proxy.common_utils.html_forms.jwt_display_template import (
    jwt_display_template,
)
from litellm.proxy.common_utils.html_forms.ui_login import build_ui_login_form
from litellm.proxy.common_utils.user_api_key_cache import UserApiKeyCache
from litellm.proxy.management_endpoints.internal_user_endpoints import new_user
from litellm.proxy.management_endpoints.sso import CustomMicrosoftSSO
from litellm.proxy.management_endpoints.sso.saml_sso import SAMLAuthHandler
from litellm.proxy.management_endpoints.sso_helper_utils import (
    check_is_admin_only_access,
    has_admin_ui_access,
)
from litellm.proxy.management_endpoints.team_endpoints import new_team, team_member_add
from litellm.proxy.management_endpoints.types import (
    CustomOpenID,
    get_litellm_user_role,
    is_valid_litellm_user_role,
)
from litellm.proxy.utils import (
    PrismaClient,
    ProxyLogging,
    get_custom_url,
    get_server_root_path,
)
from litellm.repositories.prisma_protocols import TableActions
from litellm.repositories.table_repositories import SSOConfigRepository
from litellm.repositories.team_repository import TeamRepository
from litellm.repositories.user_repository import UserRepository
from litellm.secret_managers.main import get_secret_bool, get_secret_str, str_to_bool
from litellm.types.proxy.management_endpoints.ui_sso import *  # noqa: F403
from litellm.types.proxy.management_endpoints.ui_sso import (
    DefaultTeamSSOParams,
    MicrosoftGraphAPIUserGroupDirectoryObject,
    MicrosoftGraphAPIUserGroupResponse,
    MicrosoftServicePrincipalTeam,
    RoleMappings,
    TeamMappings,
)
from litellm.types.proxy.ui_sso import ParsedOpenIDResult

if TYPE_CHECKING:
    from fastapi_sso.sso.base import OpenID
else:
    from typing import Any as OpenID

router: Final = APIRouter()

# OAuth bearer credential fields that must not appear in SSO debug responses
# (received_response is included in restricted-group error messages).
# Metadata fields (token_type, expires_in, scope) are intentionally kept so
# response convertors see the same fields in the PKCE path as in the non-PKCE path.
_OAUTH_TOKEN_FIELDS: Final = frozenset({"access_token", "id_token", "refresh_token"})
_CLI_SSO_FLOW_CACHE_KEY_PREFIX: Final = f"{CLI_SSO_SESSION_CACHE_KEY_PREFIX}:flow"
_CLI_SSO_START_RATE_LIMIT_CACHE_KEY_PREFIX: Final = f"{_CLI_SSO_FLOW_CACHE_KEY_PREFIX}:start_rate_limit"
_CLI_SSO_START_RATE_LIMIT_WINDOW_SECONDS: Final = 60
_CLI_SSO_START_RATE_LIMIT_MAX_ATTEMPTS: Final = 30
_CLI_SSO_USER_CODE_ALPHABET: Final = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789"
_CLI_SSO_LOGIN_ID_RE: Final = re.compile(r"^cli-[A-Za-z0-9_-]{12,124}$")
_CLI_SSO_USER_CODE_RE = re.compile(rf"^[{_CLI_SSO_USER_CODE_ALPHABET}]{{4}}-[{_CLI_SSO_USER_CODE_ALPHABET}]{{4}}$")
_CLI_SSO_SCALAR_TYPES: Final = (str, int, float, bool)
_CLI_SSO_DEST_KEY_RE: Final = re.compile(r"^[A-Za-z0-9_.-]+$")
_CLI_SSO_SECRET_KEY_FRAGMENTS: Final = frozenset(
    {
        "access_token",
        "api_key",
        "client_secret",
        "id_token",
        "password",
        "private_key",
        "refresh_token",
        "secret",
    }
)


class _UserMetadataRow(Protocol):
    @property
    def metadata(self) -> Mapping[str, object] | None: ...


def _user_meta_db(repo: UserRepository) -> "TableActions[_UserMetadataRow]":
    return cast(  # cast-ok: prisma types Json columns as str; the client hands back the deserialized value
        "TableActions[_UserMetadataRow]", repo.table
    )


class _SsoConfigRow(Protocol):
    @property
    def sso_settings(self) -> Mapping[str, object] | None: ...


def _sso_config_db(repo: SSOConfigRepository) -> "TableActions[_SsoConfigRow]":
    return cast(  # cast-ok: prisma types Json columns as str; the client hands back the deserialized value
        "TableActions[_SsoConfigRow]", repo.table
    )


class _TeamDetailRow(Protocol):
    def model_dump(self) -> Mapping[str, object]: ...


def _team_detail_db(repo: TeamRepository) -> "TableActions[_TeamDetailRow]":
    return repo.table


_MODEL_ALIASES_ADAPTER: Final = TypeAdapter(dict[str, str])
_SSO_TOKEN_CLAIMS_ADAPTER: Final = TypeAdapter(Mapping[str, object])


def _decode_model_aliases(value: object) -> object:
    """``/team/new`` stores team model aliases as a JSON-encoded string in the Json column."""
    if not isinstance(value, str):
        return value
    try:
        return _MODEL_ALIASES_ADAPTER.validate_json(value)
    except ValidationError:
        return None


class _TeamModelAliasTable(BaseModel):
    model_config = ConfigDict(protected_namespaces=())

    model_aliases: Annotated[Mapping[str, str] | None, BeforeValidator(_decode_model_aliases)] = None


class _TeamRowGrants(BaseModel):
    team_id: str
    team_alias: str | None = None
    models: tuple[str, ...] = ()
    litellm_model_table: _TeamModelAliasTable | None = None


class CliSsoTeamDetail(BaseModel):
    """The per-team snapshot cached in the CLI SSO flow and echoed to the CLI on poll."""

    team_id: str | None = None
    team_alias: str | None = None
    team_models: tuple[str, ...]
    team_model_aliases: Mapping[str, str] | None = None


_CLI_SSO_TEAM_DETAILS_ADAPTER: Final = TypeAdapter(tuple[CliSsoTeamDetail, ...])
_TEAMLESS_CLI_SSO_TEAM_DETAIL: Final = CliSsoTeamDetail(team_models=())


class _CustomSsoCall(Protocol):
    async def __call__(self, sso_response: object) -> SSOUserDefinedValues | None: ...


class _ServicePrincipalAssignment(Protocol):
    def get(self, key: str) -> str: ...


class _ServicePrincipalPage(Protocol):
    @overload
    def get(
        self,
        key: Literal["value"],
        default: Sequence["_ServicePrincipalAssignment"],
    ) -> Sequence["_ServicePrincipalAssignment"]: ...

    @overload
    def get(self, key: Literal["@odata.nextLink"]) -> str | None: ...


def _as_object(value: object) -> object:
    return value


def _hash_cli_sso_secret(secret: str) -> str:
    return hashlib.sha256(secret.encode("utf-8")).hexdigest()


def _normalize_cli_sso_user_code(user_code: str) -> str:
    return "".join(ch for ch in user_code.upper() if ch.isalnum())


def _generate_cli_sso_user_code() -> str:
    user_code: Final = "".join(secrets.choice(_CLI_SSO_USER_CODE_ALPHABET) for _ in range(8))
    return f"{user_code[:4]}-{user_code[4:]}"


def _get_cli_sso_flow_cache_key(login_id: str) -> str:
    return f"{_CLI_SSO_FLOW_CACHE_KEY_PREFIX}:{login_id}"


def _is_valid_cli_sso_login_id(login_id: str | None) -> bool:
    return isinstance(login_id, str) and bool(_CLI_SSO_LOGIN_ID_RE.fullmatch(login_id))


def _is_valid_cli_sso_user_code(user_code: str | None) -> bool:
    return isinstance(user_code, str) and bool(_CLI_SSO_USER_CODE_RE.fullmatch(user_code))


def _cli_sso_verification_uri_complete_enabled() -> bool:
    from litellm.proxy.proxy_server import general_settings

    return bool(general_settings.get("allow_cli_sso_verification_uri_complete", False))


def _cli_sso_start_response_body(
    *,
    login_id: str,
    poll_secret: str,
    user_code: str,
    verification_uri_complete: str | None,
) -> dict[str, str | int]:
    if verification_uri_complete is None:
        return {
            "login_id": login_id,
            "poll_secret": poll_secret,
            "user_code": user_code,
            "expires_in": CLI_SSO_SESSION_TTL_SECONDS,
        }
    return {
        "login_id": login_id,
        "poll_secret": poll_secret,
        "user_code": user_code,
        "verification_uri_complete": verification_uri_complete,
        "expires_in": CLI_SSO_SESSION_TTL_SECONDS,
    }


def _get_cli_sso_start_rate_limit_cache_key(request: Request, use_x_forwarded_for: bool | None = False) -> str:
    client_ip: Final = _get_request_ip_address(request=request, use_x_forwarded_for=use_x_forwarded_for) or "unknown"
    client_ip_hash: Final = _hash_cli_sso_secret(client_ip)
    return f"{_CLI_SSO_START_RATE_LIMIT_CACHE_KEY_PREFIX}:{client_ip_hash}"


def _check_cli_sso_start_rate_limit(
    request: Request,
    cache: DualCache,
    use_x_forwarded_for: bool | None = False,
) -> None:
    rate_limit_cache_key: Final = _get_cli_sso_start_rate_limit_cache_key(
        request=request, use_x_forwarded_for=use_x_forwarded_for
    )
    current_attempts: Final = cache.increment_cache(
        key=rate_limit_cache_key,
        value=1,
        ttl=_CLI_SSO_START_RATE_LIMIT_WINDOW_SECONDS,
    )
    if current_attempts > _CLI_SSO_START_RATE_LIMIT_MAX_ATTEMPTS:
        raise HTTPException(
            status_code=429,
            detail="Too many CLI login attempts. Try again later.",
        )


def _get_cli_sso_flow_or_raise(login_id: str | None, cache: DualCache) -> dict:
    if isinstance(login_id, str) and login_id.startswith("sk-"):
        raise HTTPException(
            status_code=400,
            detail=(
                "Your litellm CLI is out of date and uses a login flow this proxy no longer supports. "
                "Upgrade it with `pip install -U 'litellm[proxy]'` and run `litellm-proxy login` again."
            ),
        )
    if not _is_valid_cli_sso_login_id(login_id):
        raise HTTPException(status_code=400, detail="Invalid CLI login session id")

    cache_key: Final = _get_cli_sso_flow_cache_key(cast(str, login_id))
    redis_cache: Final = cache.redis_cache
    if redis_cache is not None:
        flow = redis_cache.get_cache(key=cache_key)
    else:
        flow = cache.get_cache(key=cache_key)
    if isinstance(flow, str):
        try:
            flow = _as_object(json.loads(flow))
        except ValueError:
            flow = None
    if not isinstance(flow, dict) or "poll_secret_hash" not in flow:
        verbose_proxy_logger.warning(
            "CLI SSO login session not found in cache for login_id=%s. If the proxy runs multiple replicas, "
            "a shared Redis cache is required for CLI login to work.",
            login_id,
        )
        raise HTTPException(
            status_code=400,
            detail=(
                "CLI login session not found or expired. Run `litellm-proxy login` again. "
                "If this happens immediately after starting a login, the proxy is likely running multiple "
                "replicas without a shared cache; configure a Redis cache "
                "so every replica can see the login session."
            ),
        )
    return flow


def _set_cli_sso_flow(login_id: str, cache: DualCache, flow: dict) -> None:
    cache_key: Final = _get_cli_sso_flow_cache_key(login_id)
    redis_cache: Final = cache.redis_cache
    if redis_cache is not None:
        redis_cache.set_cache(key=cache_key, value=json.dumps(flow), ttl=CLI_SSO_SESSION_TTL_SECONDS)
    else:
        cache.set_cache(key=cache_key, value=flow, ttl=CLI_SSO_SESSION_TTL_SECONDS)


def _verify_cli_sso_poll_secret(flow: dict, poll_secret: str | None) -> bool:
    expected_poll_secret_hash: Final = flow.get("poll_secret_hash")
    if not isinstance(expected_poll_secret_hash, str) or not isinstance(poll_secret, str):
        return False
    supplied_poll_secret_hash: Final = _hash_cli_sso_secret(poll_secret)
    return secrets.compare_digest(supplied_poll_secret_hash, expected_poll_secret_hash)


def _parse_cli_sso_claim_map() -> list[tuple[str, str]]:
    """
    Parse CLI_SSO_CLAIM_MAP / LITELLM_CLI_SSO_CLAIM_MAP.

    Format: comma-separated ``source_claim->metadata_key`` pairs, e.g.
    ``employment_type->acme_employment_type,org_info.department->department``.
    Destination keys may use an optional ``metadata.`` prefix; values are stored
    on the LiteLLM user's ``metadata`` JSON column.
    """
    claim_map_raw: Final = CLI_SSO_CLAIM_MAP.strip()
    if not claim_map_raw:
        return []

    parsed: Final[list[tuple[str, str]]] = []
    for entry in claim_map_raw.split(","):
        entry = entry.strip()
        if not entry or "->" not in entry:
            continue
        source_claim, dest_key = entry.split("->", 1)
        source_claim = source_claim.strip()
        dest_key = dest_key.strip()
        dest_key = dest_key.removeprefix("metadata.")
        if source_claim and dest_key:
            parsed.append((source_claim, dest_key))
    return parsed


def _is_safe_cli_sso_metadata_dest_key(dest_key: str) -> bool:
    if not dest_key or not _CLI_SSO_DEST_KEY_RE.fullmatch(dest_key):
        return False
    lowered: Final = dest_key.lower()
    return not any(fragment in lowered for fragment in _CLI_SSO_SECRET_KEY_FRAGMENTS)


def _is_safe_cli_sso_scalar_claim_value(value: object) -> bool:
    if not isinstance(value, _CLI_SSO_SCALAR_TYPES):
        return False
    if isinstance(value, str):
        if len(value) > CLI_SSO_CLAIM_MAX_SCALAR_LENGTH:
            return False
        if value.startswith("eyJ") and value.count(".") >= 2:
            return False
    return True


def _sso_result_to_dict(result: CustomOpenID | OpenID | dict[str, object]) -> dict[str, object]:
    if isinstance(result, dict):
        return result
    if hasattr(result, "model_dump"):
        dumped: Final = result.model_dump()
        if isinstance(dumped, dict):
            return dumped
    return {}


def _get_nested_claim_value(data: Mapping[str, object], claim_path: str) -> object:
    """Resolve a dot-notation claim path against an SSO result dict.

    Unlike ``get_nested_value``, this does not strip a leading ``metadata.``
    prefix, since OIDC claims may legitimately use ``metadata`` as a top-level
    key.
    """
    if not claim_path:
        return None
    if claim_path in data:
        return data[claim_path]
    placeholder: Final = "\x00"
    parts = claim_path.replace("\\.", placeholder).split(".")
    parts = [p.replace(placeholder, ".") for p in parts]
    current: object = data
    for part in parts:
        if isinstance(current, dict) and part in current:
            current = current[part]
        else:
            return None
    return current


def _extract_sso_claim_value(result: CustomOpenID | OpenID | dict[str, object], claim_path: str) -> object:
    extra_fields: Final = getattr(result, "extra_fields", None)
    if isinstance(extra_fields, dict):
        if claim_path in extra_fields:
            return extra_fields[claim_path]
        nested: Final = _get_nested_claim_value(extra_fields, claim_path)
        if nested is not None:
            return nested

    if isinstance(result, dict):
        return _get_nested_claim_value(result, claim_path)

    result_dict: Final = _sso_result_to_dict(result)
    return _get_nested_claim_value(result_dict, claim_path)


def _set_nested_metadata_value(metadata: dict[str, object], key_path: str, value: object) -> None:
    placeholder: Final = "\x00"
    parts = key_path.replace("\\.", placeholder).split(".")
    parts = [p.replace(placeholder, ".") for p in parts]
    current: dict[str, object] = metadata
    for part in parts[:-1]:
        existing = current.get(part)
        if not isinstance(existing, dict):
            existing = {}
            current[part] = existing
        current = existing
    current[parts[-1]] = value


def _flatten_cli_sso_metadata_for_poll(
    metadata: Mapping[str, object],
) -> dict[str, str | int | float | bool]:
    """Expose scalar attribution metadata as a flat dict for CLI poll responses."""
    flattened: Final[dict[str, str | int | float | bool]] = {}
    stack: Final[list[tuple[str, object]]] = [("", metadata)]
    while stack:
        prefix, value = stack.pop()
        if isinstance(value, dict):
            nested_items: Mapping[str, object] = value
            for key, nested in nested_items.items():
                nested_prefix = f"{prefix}.{key}" if prefix else key
                stack.append((nested_prefix, nested))
        elif isinstance(value, (str, int, float, bool)) and _is_safe_cli_sso_scalar_claim_value(value):
            flattened[prefix] = value
    return flattened


def build_cli_sso_attribution_metadata(
    result: CustomOpenID | OpenID | dict[str, object],
) -> dict[str, object]:
    """
    Build allowlisted, non-secret scalar attribution metadata from an SSO result.

    Sources are configured via CLI_SSO_CLAIM_MAP / LITELLM_CLI_SSO_CLAIM_MAP and
    may include claims captured by GENERIC_USER_EXTRA_ATTRIBUTES on CustomOpenID.
    """
    claim_map: Final = _parse_cli_sso_claim_map()
    if not claim_map:
        return {}

    metadata: Final[dict[str, object]] = {}
    for source_claim, dest_key in claim_map:
        if not _is_safe_cli_sso_metadata_dest_key(dest_key):
            verbose_proxy_logger.debug("Skipping unsafe CLI SSO metadata destination key: %s", dest_key)
            continue

        raw_value = _extract_sso_claim_value(result=result, claim_path=source_claim)
        if not _is_safe_cli_sso_scalar_claim_value(raw_value):
            continue

        _set_nested_metadata_value(metadata=metadata, key_path=dest_key, value=raw_value)

    return metadata


def _merge_cli_sso_attribution_metadata(
    existing_metadata: dict[str, object], attribution_metadata: dict[str, object]
) -> dict[str, object]:
    """Merge attribution metadata into existing user metadata in-place.

    Preserves original value types (in particular, string claim values that
    happen to look numeric are NOT coerced to ``int``/``float``). Nested dicts
    are merged iteratively so attribution claims do not clobber unrelated keys
    under the same parent.
    """
    pending: Final[list[tuple[dict[str, object], dict[str, object]]]] = [(existing_metadata, attribution_metadata)]
    while pending:
        target, source = pending.pop()
        for key, value in source.items():
            if value is None:
                continue
            existing_value = target.get(key)
            if isinstance(value, dict) and isinstance(existing_value, dict):
                pending.append((existing_value, value))
            else:
                target[key] = value
    return existing_metadata


async def _persist_cli_sso_user_metadata(
    prisma_client: PrismaClient,
    user_id: str,
    attribution_metadata: dict[str, object],
) -> None:
    if not attribution_metadata:
        return

    try:
        user_row: Final = await _user_meta_db(UserRepository(prisma_client)).find_unique(where={"user_id": user_id})
        existing_metadata: dict[str, object] = {}
        if user_row is not None:
            row_metadata: Final = user_row.metadata
            if isinstance(row_metadata, dict):
                existing_metadata = deepcopy(row_metadata)

        merged_metadata: Final = _merge_cli_sso_attribution_metadata(
            existing_metadata=existing_metadata,
            attribution_metadata=attribution_metadata,
        )
        await _user_meta_db(UserRepository(prisma_client)).update_many(
            where={"user_id": user_id},
            data={"metadata": merged_metadata},
        )
        verbose_proxy_logger.info(
            "Persisted CLI SSO attribution metadata for user %s: %s",
            user_id,
            list(_flatten_cli_sso_metadata_for_poll(attribution_metadata).keys()),
        )
    except Exception as e:
        verbose_proxy_logger.error("Failed to persist CLI SSO attribution metadata for user %s: %s", user_id, e)


def _cli_poll_attribution_metadata_from_session(
    session_data: Mapping[str, object],
) -> dict[str, str | int | float | bool]:
    stored: Final = session_data.get("attribution_metadata")
    if isinstance(stored, dict):
        return _flatten_cli_sso_metadata_for_poll(stored)
    return {}


def _render_cli_sso_verification_page(
    verify_url: str,
    browser_complete_token: str,
    prefill_user_code: str | None = None,
) -> str:
    escaped_verify_url: Final = escape(verify_url, quote=True)
    escaped_browser_complete_token: Final = escape(browser_complete_token, quote=True)
    user_code_value_attr: Final = f' value="{escape(prefill_user_code, quote=True)}"' if prefill_user_code else ""
    instructions: Final = (
        "Confirm the verification code below to finish this login."
        if prefill_user_code
        else "Enter the verification code shown in your terminal to finish this login."
    )
    return f"""
    <!doctype html>
    <html>
      <head>
        <title>LiteLLM CLI Login</title>
        <style>
          body {{
            font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", sans-serif;
            margin: 0;
            min-height: 100vh;
            display: flex;
            align-items: center;
            justify-content: center;
            background: #f8fafc;
            color: #0f172a;
          }}
          main {{
            width: min(420px, calc(100vw - 32px));
            background: #ffffff;
            border: 1px solid #e2e8f0;
            border-radius: 8px;
            padding: 28px;
            box-shadow: 0 12px 32px rgba(15, 23, 42, 0.08);
          }}
          h1 {{ font-size: 22px; margin: 0 0 12px; }}
          p {{ line-height: 1.5; margin: 0 0 18px; color: #334155; }}
          label {{ display: block; font-weight: 600; margin-bottom: 8px; }}
          input {{
            box-sizing: border-box;
            width: 100%;
            padding: 12px;
            border: 1px solid #cbd5e1;
            border-radius: 6px;
            font-size: 20px;
            letter-spacing: 0.08em;
            text-transform: uppercase;
          }}
          button {{
            margin-top: 16px;
            width: 100%;
            padding: 12px;
            border: 0;
            border-radius: 6px;
            background: #0f172a;
            color: #ffffff;
            font-weight: 600;
            cursor: pointer;
          }}
        </style>
      </head>
      <body>
        <main>
          <h1>Complete CLI Login</h1>
          <p>{instructions}</p>
          <form method="post" action="{escaped_verify_url}">
            <input type="hidden" name="browser_complete_token" value="{escaped_browser_complete_token}" />
            <label for="user_code">Verification code</label>
            <input id="user_code" name="user_code" autocomplete="one-time-code"{user_code_value_attr} required autofocus />
            <button type="submit">Continue</button>
          </form>
        </main>
      </body>
    </html>
    """


@router.post("/sso/cli/start", tags=["experimental"], include_in_schema=False)
async def cli_sso_start(request: Request):
    from litellm.proxy.proxy_server import cli_sso_session_cache, general_settings

    _check_cli_sso_start_rate_limit(
        request=request,
        cache=cli_sso_session_cache,
        use_x_forwarded_for=bool((general_settings or {}).get("use_x_forwarded_for", False)),
    )

    login_id: Final = f"cli-{secrets.token_urlsafe(24)}"
    poll_secret: Final = secrets.token_urlsafe(32)
    user_code: Final = _generate_cli_sso_user_code()

    flow: Final = {
        "poll_secret_hash": _hash_cli_sso_secret(poll_secret),
        "user_code_hash": _hash_cli_sso_secret(_normalize_cli_sso_user_code(user_code)),
        "sso_complete": False,
        "user_code_verified": False,
        "session_data": None,
    }
    _set_cli_sso_flow(login_id=login_id, cache=cli_sso_session_cache, flow=flow)

    verification_uri_complete: Final[str | None] = (
        (
            get_custom_url(request_base_url=str(request.base_url), route="sso/key/generate")
            + "?"
            + urlencode(
                {
                    "source": LITELLM_CLI_SOURCE_IDENTIFIER,
                    "key": login_id,
                    "user_code": user_code,
                }
            )
        )
        if _cli_sso_verification_uri_complete_enabled()
        else None
    )
    return _cli_sso_start_response_body(
        login_id=login_id,
        poll_secret=poll_secret,
        user_code=user_code,
        verification_uri_complete=verification_uri_complete,
    )


@router.post("/sso/cli/complete/{login_id}", tags=["experimental"], include_in_schema=False)
async def cli_sso_complete(request: Request, login_id: str):
    from fastapi.responses import HTMLResponse

    from litellm.proxy.common_utils.html_forms.cli_sso_success import (
        render_cli_sso_success_page,
    )
    from litellm.proxy.proxy_server import cli_sso_session_cache

    flow: Final = _get_cli_sso_flow_or_raise(login_id=login_id, cache=cli_sso_session_cache)
    if not flow.get("sso_complete") or not flow.get("session_data"):
        raise HTTPException(status_code=400, detail="CLI login is not ready")

    body: Final = (await request.body()).decode("utf-8")
    form_values: Final = parse_qs(body)
    supplied_user_code: Final = (form_values.get("user_code") or [""])[0]
    supplied_browser_complete_token: Final = (form_values.get("browser_complete_token") or [""])[0]
    supplied_user_code_hash: Final = _hash_cli_sso_secret(_normalize_cli_sso_user_code(supplied_user_code))
    supplied_browser_complete_token_hash: Final = _hash_cli_sso_secret(supplied_browser_complete_token)

    expected_user_code_hash: Final = flow.get("user_code_hash")
    if not isinstance(expected_user_code_hash, str) or not secrets.compare_digest(
        supplied_user_code_hash, expected_user_code_hash
    ):
        raise HTTPException(status_code=400, detail="Invalid verification code")

    expected_browser_complete_token_hash: Final = flow.get("browser_complete_token_hash")
    if not isinstance(expected_browser_complete_token_hash, str) or not secrets.compare_digest(
        supplied_browser_complete_token_hash, expected_browser_complete_token_hash
    ):
        raise HTTPException(status_code=400, detail="Invalid verification code")

    flow["user_code_verified"] = True
    _set_cli_sso_flow(login_id=login_id, cache=cli_sso_session_cache, flow=flow)

    html_content: Final = render_cli_sso_success_page()
    return HTMLResponse(content=html_content, status_code=200)


def normalize_email(email: str | None) -> str | None:
    """
    Normalize email address to lowercase for consistent storage and comparison.

    Email addresses should be treated as case-insensitive for SSO purposes,
    even though RFC 5321 technically allows case-sensitive local parts.
    This prevents issues where SSO providers return emails with different casing
    than what's stored in the database.

    Args:
        email: Email address to normalize, can be None

    Returns:
        Lowercased email address, or None if input is None
    """
    if email is None:
        return None
    return email.lower() if isinstance(email, str) else email


# Ordered highest to lowest privilege
LITELLM_USER_ROLE_HIERARCHY: Final = (
    LitellmUserRoles.PROXY_ADMIN,
    LitellmUserRoles.PROXY_ADMIN_VIEW_ONLY,
    LitellmUserRoles.INTERNAL_USER,
    LitellmUserRoles.INTERNAL_USER_VIEW_ONLY,
)


def determine_role_from_groups(
    user_groups: list[str],
    role_mappings: "RoleMappings",
) -> LitellmUserRoles | None:
    """
    Determine the highest privilege role for a user based on their groups.

    Role hierarchy (highest to lowest):
    - proxy_admin
    - proxy_admin_viewer
    - internal_user
    - internal_user_viewer

    Args:
        user_groups: List of group names from the SSO token
        role_mappings: RoleMappings configuration object

    Returns:
        The highest privilege role found, or default_role if no matches, or None
    """
    if not role_mappings.roles:
        # No role mappings configured, return default_role
        return role_mappings.default_role

    # Convert user_groups to a set for efficient lookup
    user_groups_set: Final = set(user_groups) if isinstance(user_groups, list) else set()

    # Find the highest privilege role the user belongs to
    for role in LITELLM_USER_ROLE_HIERARCHY:
        if role in role_mappings.roles:
            role_groups = role_mappings.roles[role]
            if isinstance(role_groups, list) and user_groups_set.intersection(set(role_groups)):
                verbose_proxy_logger.debug(
                    "User groups %s matched role '%s' via groups: %s", user_groups, role.value, role_groups
                )
                return role

    # No matching groups found, return default_role
    verbose_proxy_logger.debug(
        "User groups %s did not match any role mappings, using default_role: %s",
        user_groups,
        role_mappings.default_role,
    )
    return role_mappings.default_role


def process_sso_jwt_access_token(
    access_token_str: str | None,
    sso_jwt_handler: JWTHandler | None,
    result: OpenID | dict | None,
    role_mappings: Optional["RoleMappings"] = None,
) -> dict | None:
    """
    Process SSO JWT access token and extract team IDs and user role if available.

    This function decodes the JWT access token and extracts team IDs and user
    role, then sets them on the result object. Role extraction from the access
    token is needed because some SSO providers (e.g., Keycloak) do not include
    role claims in the UserInfo endpoint response.

    Args:
        access_token_str: The JWT access token string
        sso_jwt_handler: SSO-specific JWT handler for team ID extraction
        result: The SSO result object to update with team IDs and role
        role_mappings: Optional role mappings configuration for group-based role determination

    Returns:
        The decoded access token payload dict, or None if decoding failed or
        inputs were missing. Callers can pass this to _sync_user_role_from_jwt_role_map
        so it has access to custom role claims (e.g. custom_roles) that are
        encoded inside the JWT but stripped from received_response.
    """
    if access_token_str and result:
        import jwt

        try:
            access_token_payload: Final = jwt.decode(access_token_str, options={"verify_signature": False})
        except jwt.exceptions.DecodeError:
            verbose_proxy_logger.debug(
                "Access token is not a valid JWT (possibly an opaque token), skipping JWT-based extraction"
            )
            return None

        # Extract team IDs from access token if sso_jwt_handler is available
        if sso_jwt_handler:
            if isinstance(result, dict):
                result_team_ids: list[str] | None = result.get("team_ids", [])
                if not result_team_ids:
                    team_ids = sso_jwt_handler.get_team_ids_from_jwt(access_token_payload)
                    result["team_ids"] = team_ids
            else:
                result_team_ids = getattr(result, "team_ids", []) if result else []
                if not result_team_ids:
                    team_ids = sso_jwt_handler.get_team_ids_from_jwt(access_token_payload)
                    setattr(result, "team_ids", team_ids)

        # Extract user role from access token if not already set from UserInfo
        existing_role = result.get("user_role") if isinstance(result, dict) else getattr(result, "user_role", None)
        if existing_role is None:
            user_role: LitellmUserRoles | None = None

            # Try role_mappings first (group-based role determination)
            if role_mappings is not None and role_mappings.roles:
                group_claim: Final = role_mappings.group_claim
                user_groups_raw: Final[object] = get_nested_value(access_token_payload, group_claim)

                user_groups: list[str] = []
                if isinstance(user_groups_raw, list):
                    raw_groups: Final[Sequence[object]] = user_groups_raw
                    user_groups = [str(g) for g in raw_groups]
                elif isinstance(user_groups_raw, str):
                    user_groups = [g.strip() for g in user_groups_raw.split(",") if g.strip()]
                elif user_groups_raw is not None:
                    user_groups = [str(user_groups_raw)]

                if user_groups:
                    user_role = determine_role_from_groups(user_groups, role_mappings)
                    verbose_proxy_logger.debug(
                        "Determined role '%s' from access token groups '%s' using role_mappings", user_role, user_groups
                    )
                elif role_mappings.default_role:
                    user_role = role_mappings.default_role

            # Fallback: try GENERIC_USER_ROLE_ATTRIBUTE on the access token payload
            if user_role is None:
                generic_user_role_attribute_name: Final = os.getenv("GENERIC_USER_ROLE_ATTRIBUTE", "role")
                user_role_from_token: Final = get_nested_value(access_token_payload, generic_user_role_attribute_name)
                if user_role_from_token is not None:
                    user_role = get_litellm_user_role(user_role_from_token)
                    verbose_proxy_logger.debug(
                        "Extracted role '%s' from access token field '%s'", user_role, generic_user_role_attribute_name
                    )

            if user_role is not None:
                if isinstance(result, dict):
                    result["user_role"] = user_role
                else:
                    setattr(result, "user_role", user_role)
                verbose_proxy_logger.debug("Set user_role='%s' from JWT access token", user_role)

        return access_token_payload

    return None


def _decode_sso_token_claims(token: str | None) -> Mapping[str, object]:
    if not token:
        return MappingProxyType({})
    try:
        return MappingProxyType(
            _SSO_TOKEN_CLAIMS_ADAPTER.validate_python(jwt.decode(token, options={"verify_signature": False}))
        )
    except (jwt.exceptions.InvalidTokenError, ValidationError):
        verbose_proxy_logger.debug("SSO token is not a decodable JWT, skipping token claims")
        return MappingProxyType({})


def _merge_sso_token_claims(
    userinfo: Mapping[str, object],
    id_token: str | None,
    access_token: str | None,
) -> Mapping[str, object]:
    sources: Final = (userinfo, _decode_sso_token_claims(id_token), _decode_sso_token_claims(access_token))
    claim_names: Final = frozenset(key for source in sources for key in source)
    return MappingProxyType(
        {key: next((source[key] for source in sources if source.get(key) is not None), None) for key in claim_names}
    )


async def _raise_if_sso_exceeds_free_user_limit(premium_user: bool, prisma_client: PrismaClient | None) -> None:
    """Free tier allows SSO for up to 5 billable users; beyond that requires an Enterprise license."""
    if premium_user is True:
        return
    if prisma_client is None:
        raise ProxyException(
            message=CommonProxyErrors.db_not_connected_error.value,
            type=ProxyErrorTypes.auth_error,
            param="premium_user",
            code=status.HTTP_403_FORBIDDEN,
        )
    billable_users: Final = await UserRepository(prisma_client).count_billable_users()
    if billable_users and billable_users > 5:
        raise ProxyException(
            message="You must be a LiteLLM Enterprise user to use SSO for more than 5 users. If you have a license please set `LITELLM_LICENSE` in your env. If you want to obtain a license meet with us here: https://enterprise.litellm.ai/demo You are seeing this error message because You configured SSO (one of `MICROSOFT_CLIENT_ID`, `GOOGLE_CLIENT_ID`, `GENERIC_CLIENT_ID`, or SAML) in your env. Please unset it",
            type=ProxyErrorTypes.auth_error,
            param="premium_user",
            code=status.HTTP_403_FORBIDDEN,
        )


@router.get("/sso/key/generate", tags=["experimental"], include_in_schema=False)
async def google_login(
    request: Request,
    source: str | None = None,
    key: str | None = None,
    existing_key: str | None = None,
    return_to: str | None = None,
    user_code: str | None = None,
):
    """
    Create Proxy API Keys using Google Workspace SSO. Requires setting PROXY_BASE_URL in .env
    PROXY_BASE_URL should be the your deployed proxy endpoint, e.g. PROXY_BASE_URL="https://litellm-production-7002.up.railway.app/"
    Example:
    """
    from litellm.proxy.proxy_server import (
        cli_sso_session_cache,
        general_settings,
        premium_user,
        prisma_client,
        user_api_key_cache,
        user_custom_ui_sso_sign_in_handler,
    )

    microsoft_client_id: Final = os.getenv("MICROSOFT_CLIENT_ID", None)
    google_client_id: Final = os.getenv("GOOGLE_CLIENT_ID", None)
    generic_client_id: Final = os.getenv("GENERIC_CLIENT_ID", None)

    ####### Check if UI is disabled #######
    _disable_ui_flag: Final = os.getenv("DISABLE_ADMIN_UI")
    if _disable_ui_flag is not None:
        is_disabled: Final = str_to_bool(value=_disable_ui_flag)
        if is_disabled:
            return admin_ui_disabled()

    ####### Check if user is a Enterprise / Premium User #######
    if (
        microsoft_client_id is not None
        or google_client_id is not None
        or generic_client_id is not None
        or SAMLAuthHandler.is_saml_configured()
    ):
        await _raise_if_sso_exceeds_free_user_limit(premium_user, prisma_client)

    ####### Detect DB + MASTER KEY in .env #######
    missing_env_vars: Final = show_missing_vars_in_env()
    if missing_env_vars is not None:
        return missing_env_vars

    # get url from request - always use regular callback, but set state for CLI
    redirect_url: Final = SSOAuthenticationHandler.get_redirect_url_for_sso(
        request=request,
        sso_callback_route="sso/callback",
    )

    if source == LITELLM_CLI_SOURCE_IDENTIFIER:
        _get_cli_sso_flow_or_raise(login_id=key, cache=cli_sso_session_cache)

    # Store CLI login handle in state for OAuth flow
    cli_state: Final[str | None] = SSOAuthenticationHandler._get_cli_state(
        source=source,
        key=key,
        user_code=(user_code if _cli_sso_verification_uri_complete_enabled() else None),
    )

    # check if user defined a custom auth sso sign in handler, if yes, use it
    if user_custom_ui_sso_sign_in_handler is not None:
        try:
            from litellm_enterprise.proxy.auth.custom_sso_handler import (
                EnterpriseCustomSSOHandler,
            )

            return await EnterpriseCustomSSOHandler.handle_custom_ui_sso_sign_in(
                request=request,
            )
        except ImportError:
            raise ValueError(
                "Enterprise features are not available. Custom UI SSO sign-in requires LiteLLM Enterprise."
            )

    if (
        microsoft_client_id is None
        and google_client_id is None
        and generic_client_id is None
        and SAMLAuthHandler.is_saml_configured()
    ):
        verbose_proxy_logger.info("Redirecting to SAML SSO login")
        return await SAMLAuthHandler.build_login_redirect(
            request=request,
            cache=user_api_key_cache,
            relay_state=return_to,
        )

    # Check if we should use SSO handler
    if (
        SSOAuthenticationHandler.should_use_sso_handler(
            microsoft_client_id=microsoft_client_id,
            google_client_id=google_client_id,
            generic_client_id=generic_client_id,
        )
        is True
    ):
        verbose_proxy_logger.info("Redirecting to SSO login for %s", redirect_url)
        sso_redirect: Final = await SSOAuthenticationHandler.get_sso_login_redirect(
            redirect_url=redirect_url,
            microsoft_client_id=microsoft_client_id,
            google_client_id=google_client_id,
            generic_client_id=generic_client_id,
            state=cli_state,
            request=request,
        )
        if sso_redirect is not None:
            _persist_return_to_cookie(sso_redirect, return_to)
        return sso_redirect

    from fastapi.responses import HTMLResponse

    hide_default_credentials_hint: Final = (
        os.getenv("LITELLM_HIDE_DEFAULT_CREDENTIALS_HINT", "false").lower() == "true"
        or general_settings.get("hide_default_credentials_hint", False) is True
    )
    form_response: Final = HTMLResponse(
        content=build_ui_login_form(
            show_deprecation_banner=True,
            hide_default_credentials_hint=hide_default_credentials_hint,
        ),
        status_code=200,
    )
    # Preserve return_to across the username/password sign-in too, via the SAME shared, never-raising
    # helper the SSO branch uses, so /login can resume the connect flow instead of dead-ending at the
    # dashboard. One implementation → the two sign-in branches cannot diverge (and the login form always
    # renders, since the helper never raises on a bad return_to).
    _persist_return_to_cookie(form_response, return_to)
    return form_response


def generic_response_convertor(
    response,
    jwt_handler: JWTHandler,
    sso_jwt_handler: JWTHandler | None = None,
    role_mappings: Optional["RoleMappings"] = None,
    team_mappings: Optional["TeamMappings"] = None,
) -> CustomOpenID:
    generic_user_id_attribute_name: Final = os.getenv("GENERIC_USER_ID_ATTRIBUTE", "preferred_username")
    generic_user_display_name_attribute_name: Final = os.getenv("GENERIC_USER_DISPLAY_NAME_ATTRIBUTE", "sub")
    generic_user_email_attribute_name: Final = os.getenv("GENERIC_USER_EMAIL_ATTRIBUTE", "email")

    generic_user_first_name_attribute_name: Final = os.getenv("GENERIC_USER_FIRST_NAME_ATTRIBUTE", "first_name")
    generic_user_last_name_attribute_name: Final = os.getenv("GENERIC_USER_LAST_NAME_ATTRIBUTE", "last_name")

    generic_provider_attribute_name: Final = os.getenv("GENERIC_USER_PROVIDER_ATTRIBUTE", "provider")

    generic_user_role_attribute_name: Final = os.getenv("GENERIC_USER_ROLE_ATTRIBUTE", "role")

    generic_user_extra_attributes: Final = os.getenv("GENERIC_USER_EXTRA_ATTRIBUTES", None)

    verbose_proxy_logger.debug(
        " generic_user_id_attribute_name: %s\n generic_user_email_attribute_name: %s",
        generic_user_id_attribute_name,
        generic_user_email_attribute_name,
    )

    all_teams: Final = []
    if sso_jwt_handler is not None:
        team_ids = sso_jwt_handler.get_all_jwt_team_ids(cast(dict, response))
        all_teams.extend(team_ids)

    if team_mappings is not None and team_mappings.team_ids_jwt_field is not None:
        team_ids_from_db_mapping: Final[list[str] | None] = get_nested_value(
            data=cast(dict, response),
            key_path=team_mappings.team_ids_jwt_field,
            default=[],
        )
        if team_ids_from_db_mapping:
            all_teams.extend(team_ids_from_db_mapping)
            verbose_proxy_logger.debug(
                "Loaded team_ids from DB team_mappings.team_ids_jwt_field='%s': %s",
                team_mappings.team_ids_jwt_field,
                team_ids_from_db_mapping,
            )
    else:
        team_ids = jwt_handler.get_all_jwt_team_ids(cast(dict, response))
        all_teams.extend(team_ids)

    # Determine user role based on role_mappings if available
    # Only apply role_mappings for GENERIC SSO provider
    user_role: LitellmUserRoles | None = None

    if role_mappings is not None and role_mappings.provider.lower() in [
        "generic",
        "okta",
    ]:
        # Use role_mappings to determine role from groups
        group_claim: Final = role_mappings.group_claim
        user_groups_raw: Final[object] = get_nested_value(response, group_claim)

        # Handle different formats: could be a list, string (comma-separated), or single value
        user_groups: list[str] = []
        if isinstance(user_groups_raw, list):
            raw_groups: Final[Sequence[object]] = user_groups_raw
            user_groups = [str(g) for g in raw_groups]
        elif isinstance(user_groups_raw, str):
            # Handle comma-separated string
            user_groups = [g.strip() for g in user_groups_raw.split(",") if g.strip()]
        elif user_groups_raw is not None:
            # Single value
            user_groups = [str(user_groups_raw)]

        if user_groups:
            user_role = determine_role_from_groups(user_groups, role_mappings)
            verbose_proxy_logger.debug(
                "Determined role '%s' from groups '%s' using role_mappings",
                user_role.value if user_role else None,
                user_groups,
            )
        else:
            # No groups found, use default_role
            user_role = role_mappings.default_role
            verbose_proxy_logger.debug(
                "No groups found in '%s', using default_role: %s", group_claim, role_mappings.default_role
            )

    # Fallback to existing logic if role_mappings not used
    if user_role is None:
        user_role_from_sso: Final = get_nested_value(response, generic_user_role_attribute_name)
        if user_role_from_sso is not None:
            role: Final = get_litellm_user_role(user_role_from_sso)
            if role is not None:
                user_role = role
                verbose_proxy_logger.debug(
                    "Found valid LitellmUserRoles '%s' from SSO attribute '%s'",
                    role.value,
                    generic_user_role_attribute_name,
                )

    # Build extra_fields dict from GENERIC_USER_EXTRA_ATTRIBUTES if specified
    extra_fields: dict[str, object] | None = None
    if generic_user_extra_attributes:
        extra_fields = {}
        for attr_name in generic_user_extra_attributes.split(","):
            attr_name = attr_name.strip()
            extra_fields[attr_name] = get_nested_value(response, attr_name)

    return CustomOpenID(
        id=get_nested_value(response, generic_user_id_attribute_name),
        display_name=get_nested_value(response, generic_user_display_name_attribute_name),
        email=normalize_email(get_nested_value(response, generic_user_email_attribute_name)),
        first_name=get_nested_value(response, generic_user_first_name_attribute_name),
        last_name=get_nested_value(response, generic_user_last_name_attribute_name),
        provider=get_nested_value(response, generic_provider_attribute_name),
        team_ids=all_teams,
        user_role=user_role,
        extra_fields=extra_fields,
    )


def _setup_generic_sso_env_vars(
    generic_client_id: str, redirect_url: str
) -> tuple[str, list[str], str, str, str, bool]:
    """Setup and validate Generic SSO environment variables."""
    generic_client_secret: Final = os.getenv("GENERIC_CLIENT_SECRET", None)
    generic_scope: Final = os.getenv("GENERIC_SCOPE", "openid email profile").split(" ")
    generic_authorization_endpoint: Final = os.getenv("GENERIC_AUTHORIZATION_ENDPOINT", None)
    generic_token_endpoint: Final = os.getenv("GENERIC_TOKEN_ENDPOINT", None)
    generic_userinfo_endpoint: Final = os.getenv("GENERIC_USERINFO_ENDPOINT", None)
    generic_include_client_id: Final = os.getenv("GENERIC_INCLUDE_CLIENT_ID", "false").lower() == "true"

    # Validate required environment variables
    if generic_client_secret is None:
        raise ProxyException(
            message="GENERIC_CLIENT_SECRET not set. Set it in .env file",
            type=ProxyErrorTypes.auth_error,
            param="GENERIC_CLIENT_SECRET",
            code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    if generic_authorization_endpoint is None:
        raise ProxyException(
            message="GENERIC_AUTHORIZATION_ENDPOINT not set. Set it in .env file",
            type=ProxyErrorTypes.auth_error,
            param="GENERIC_AUTHORIZATION_ENDPOINT",
            code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    if generic_token_endpoint is None:
        raise ProxyException(
            message="GENERIC_TOKEN_ENDPOINT not set. Set it in .env file",
            type=ProxyErrorTypes.auth_error,
            param="GENERIC_TOKEN_ENDPOINT",
            code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    if generic_userinfo_endpoint is None:
        raise ProxyException(
            message="GENERIC_USERINFO_ENDPOINT not set. Set it in .env file",
            type=ProxyErrorTypes.auth_error,
            param="GENERIC_USERINFO_ENDPOINT",
            code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )

    verbose_proxy_logger.debug(
        "authorization_endpoint: %s\ntoken_endpoint: %s\nuserinfo_endpoint: %s",
        generic_authorization_endpoint,
        generic_token_endpoint,
        generic_userinfo_endpoint,
    )
    verbose_proxy_logger.debug("GENERIC_REDIRECT_URI: %s\nGENERIC_CLIENT_ID: %s\n", redirect_url, generic_client_id)

    return (
        generic_client_secret,
        generic_scope,
        generic_authorization_endpoint,
        generic_token_endpoint,
        generic_userinfo_endpoint,
        generic_include_client_id,
    )


async def _setup_team_mappings() -> Optional["TeamMappings"]:
    """Setup team mappings from SSO database settings."""
    from litellm.types.proxy.management_endpoints.ui_sso import TeamMappings

    team_mappings: TeamMappings | None = None
    try:
        from litellm.proxy.utils import get_prisma_client_or_throw

        prisma_client: Final = get_prisma_client_or_throw("Prisma client is None, connect a database to your proxy")

        sso_db_record: Final = await _sso_config_db(SSOConfigRepository(prisma_client)).find_unique(
            where={"id": "sso_config"}
        )

        if sso_db_record and sso_db_record.sso_settings:
            sso_settings_dict: Final = dict(sso_db_record.sso_settings)
            team_mappings_data: Final = sso_settings_dict.get("team_mappings")

            if team_mappings_data:
                if isinstance(team_mappings_data, dict):
                    team_mappings = TeamMappings(**team_mappings_data)
                elif isinstance(team_mappings_data, TeamMappings):
                    team_mappings = team_mappings_data

                if team_mappings and team_mappings.team_ids_jwt_field:
                    verbose_proxy_logger.debug(
                        "Loaded team_mappings with team_ids_jwt_field: '%s'", team_mappings.team_ids_jwt_field
                    )
    except Exception as e:
        verbose_proxy_logger.debug(
            "Could not load team_mappings from database: %s. Continuing with config-based team mapping.", e
        )

    return team_mappings


async def _setup_role_mappings() -> Optional["RoleMappings"]:
    """Setup role mappings from SSO database settings."""
    role_mappings: RoleMappings | None = None
    try:
        from litellm.proxy.utils import get_prisma_client_or_throw

        prisma_client: Final = get_prisma_client_or_throw("Prisma client is None, connect a database to your proxy")

        sso_db_record: Final = await _sso_config_db(SSOConfigRepository(prisma_client)).find_unique(
            where={"id": "sso_config"}
        )

        if sso_db_record and sso_db_record.sso_settings:
            sso_settings_dict: Final = dict(sso_db_record.sso_settings)
            role_mappings_data = sso_settings_dict.get("role_mappings")

            if role_mappings_data:
                if isinstance(role_mappings_data, dict):
                    role_mappings = RoleMappings(**role_mappings_data)
                elif isinstance(role_mappings_data, RoleMappings):
                    role_mappings = role_mappings_data

                if role_mappings:
                    verbose_proxy_logger.debug("Loaded role_mappings for provider '%s'", role_mappings.provider)
    except Exception as e:
        verbose_proxy_logger.debug(
            "Could not load role_mappings from database: %s. Continuing with existing role logic.", e
        )

    generic_role_mappings: Final = os.getenv("GENERIC_ROLE_MAPPINGS_ROLES", None)
    generic_role_mappings_group_claim: Final = os.getenv("GENERIC_ROLE_MAPPINGS_GROUP_CLAIM", None)
    generic_role_mappings_default_role: Final = os.getenv("GENERIC_ROLE_MAPPINGS_DEFAULT_ROLE", None)
    if generic_role_mappings is not None:
        verbose_proxy_logger.debug("Found role_mappings for generic provider in environment variables")
        import ast

        try:
            generic_user_role_mappings_data: dict[LitellmUserRoles, list[str]] = ast.literal_eval(generic_role_mappings)
            if isinstance(generic_user_role_mappings_data, dict):
                role_mappings_data = {
                    "provider": "generic",
                    "group_claim": generic_role_mappings_group_claim,
                    "default_role": generic_role_mappings_default_role,
                    "roles": generic_user_role_mappings_data,
                }

                role_mappings = RoleMappings(**role_mappings_data)
                verbose_proxy_logger.debug(
                    "Loaded role_mappings from environments for provider '%s'.", role_mappings.provider
                )
                return role_mappings
        except TypeError as e:
            verbose_proxy_logger.warning(
                "Error decoding role mappings from environment variables: %s. Continuing with existing role logic.", e
            )
    return role_mappings


def _parse_generic_sso_headers() -> dict[str, str]:
    """Parse comma-separated GENERIC_SSO_HEADERS env var into a dict."""
    raw: Final = os.getenv("GENERIC_SSO_HEADERS", None)
    if raw is None:
        return {}
    result: Final[dict[str, str]] = {}
    for header in raw.split(","):
        header = header.strip()
        if header:
            key, value = header.split("=")
            result[key] = value
    return result


def _handle_generic_sso_error(
    e: Exception,
    generic_authorization_endpoint: str | None,
    generic_token_endpoint: str | None,
    additional_headers: dict,
) -> NoReturn:
    """Handle errors from generic SSO verify_and_process. Always re-raises."""
    error_message: Final = str(e)

    # Surface a helpful PKCE misconfiguration hint only when:
    # 1. The error mentions PKCE/code verifier, AND
    # 2. PKCE is not currently configured (GENERIC_CLIENT_USE_PKCE != true)
    pkce_configured: Final = os.getenv("GENERIC_CLIENT_USE_PKCE", "false").lower() == "true"
    if not pkce_configured and ("PKCE" in error_message or "code verifier" in error_message.lower()):
        is_okta: Final = (generic_authorization_endpoint and "okta" in generic_authorization_endpoint.lower()) or (
            generic_token_endpoint and "okta" in generic_token_endpoint.lower()
        )
        provider_name: Final = "Okta" if is_okta else "Your OAuth provider"

        detailed_message = (
            f"SSO authentication failed: {provider_name} requires PKCE (Proof Key for Code Exchange) "
            f"but it's not enabled in your LiteLLM configuration.\n\n"
            f"SOLUTION: Add this environment variable and restart your proxy:\n"
            f"  GENERIC_CLIENT_USE_PKCE=true\n\n"
        )
        if is_okta:
            detailed_message += (
                "For AWS ECS: Add the environment variable to your task definition.\n"
                "For Docker: Add -e GENERIC_CLIENT_USE_PKCE=true to your docker run command.\n"
                "For .env file: Add GENERIC_CLIENT_USE_PKCE=true to your .env file.\n\n"
            )
        detailed_message += f"Original error: {error_message}"

        raise ProxyException(
            message=detailed_message,
            type=ProxyErrorTypes.auth_error,
            param="GENERIC_CLIENT_USE_PKCE",
            code=status.HTTP_401_UNAUTHORIZED,
        )

    if isinstance(e, ProxyException):
        verbose_proxy_logger.error(
            "SSO authentication failed: %s. Passed in headers: %s",
            e,
            additional_headers,
        )
    else:
        verbose_proxy_logger.exception(
            "Error verifying and processing generic SSO: %s. Passed in headers: %s",
            e,
            additional_headers,
        )
    raise e


async def get_generic_sso_response(
    request: Request,
    jwt_handler: JWTHandler,
    sso_jwt_handler: JWTHandler | None,  # sso specific jwt handler - used for restricted sso group access control
    generic_client_id: str,
    redirect_url: str,
) -> tuple[
    OpenID | dict, dict | None, dict | None, SSOIdentityAssertion | None
]:  # (result, received_response, access_token_payload, sso_assertion)
    # make generic sso provider
    from fastapi_sso.sso.base import DiscoveryDocument
    from fastapi_sso.sso.generic import create_provider

    received_response: dict | None = None
    sso_assertion: SSOIdentityAssertion | None = None

    # Setup environment variables
    (
        generic_client_secret,
        generic_scope,
        generic_authorization_endpoint,
        generic_token_endpoint,
        generic_userinfo_endpoint,
        generic_include_client_id,
    ) = _setup_generic_sso_env_vars(generic_client_id, redirect_url)

    discovery: Final = DiscoveryDocument(
        authorization_endpoint=generic_authorization_endpoint,
        token_endpoint=generic_token_endpoint,
        userinfo_endpoint=generic_userinfo_endpoint,
    )

    role_mappings: Final = await _setup_role_mappings()
    team_mappings: Final = await _setup_team_mappings()
    generic_include_token_claims: Final = os.getenv("GENERIC_INCLUDE_TOKEN_CLAIMS", "false").lower() == "true"

    def response_convertor(response: Mapping[str, object], httpx_session: object):
        nonlocal received_response  # return for user debugging
        response_id_token: Final = response.get("id_token")
        response_access_token: Final = response.get("access_token")
        id_token: Final = (
            response_id_token if isinstance(response_id_token, str) and response_id_token else generic_sso.id_token
        )
        access_token: Final = (
            response_access_token
            if isinstance(response_access_token, str) and response_access_token
            else generic_sso.access_token
        )
        claims: Final = (
            _merge_sso_token_claims(
                userinfo=response,
                id_token=id_token,
                access_token=access_token,
            )
            if generic_include_token_claims
            else response
        )
        received_response = {  # mutable-ok: preserve the existing dict return contract
            key: value for key, value in claims.items() if key not in _OAUTH_TOKEN_FIELDS
        }
        return generic_response_convertor(
            response=claims,
            jwt_handler=jwt_handler,
            sso_jwt_handler=sso_jwt_handler,
            role_mappings=role_mappings,
            team_mappings=team_mappings,
        )

    SSOProvider: Final = create_provider(
        name="oidc",
        discovery_document=discovery,
        response_convertor=response_convertor,
    )
    generic_sso: Final = SSOProvider(
        client_id=generic_client_id,
        client_secret=generic_client_secret,
        redirect_uri=redirect_url,
        allow_insecure_http=True,
        scope=generic_scope,
    )
    verbose_proxy_logger.debug("calling generic_sso.verify_and_process")
    additional_generic_sso_headers_dict: Final = _parse_generic_sso_headers()

    code_verifier: str | None = None  # assigned inside try; initialized for type tracking
    access_token_payload: dict | None = None  # decoded JWT access token claims

    try:
        token_exchange_params: Final = await SSOAuthenticationHandler.prepare_token_exchange_parameters(
            request=request,
            generic_include_client_id=generic_include_client_id,
        )

        # Extract code_verifier (and the cache key for deferred deletion) before calling fastapi-sso
        code_verifier = token_exchange_params.pop("code_verifier", None)
        pkce_cache_key: Final = token_exchange_params.pop("_pkce_cache_key", None)

        # Get authorization code from query params (only used in the PKCE path below;
        # the non-PKCE path delegates to verify_and_process which handles OAuth error
        # callbacks — user-denied, CSRF mismatch — internally).
        authorization_code: Final = request.query_params.get("code")

        if code_verifier:
            # State-to-session-cookie binding.  The non-PKCE branch below
            # delegates to fastapi-sso's ``verify_and_process``, which
            # performs its own session-cookie check.  The PKCE branch
            # bypasses that helper, so we validate the URL ``state``
            # against the ``litellm_oauth_state`` cookie set on the
            # redirect response — without this an attacker can pre-mint
            # a state + cached PKCE verifier and hijack a victim's auth
            # code (Login-CSRF / token theft).
            url_state: Final = request.query_params.get("state")
            cookie_state: Final = request.cookies.get("litellm_oauth_state")
            if not url_state or not cookie_state or not secrets.compare_digest(url_state, cookie_state):
                raise ProxyException(
                    message=("Invalid OAuth state parameter — does not match the browser-bound state cookie."),
                    type=ProxyErrorTypes.auth_error,
                    param="state",
                    code=status.HTTP_400_BAD_REQUEST,
                )
            if not authorization_code:
                raise ProxyException(
                    message="Missing authorization code in callback",
                    type=ProxyErrorTypes.auth_error,
                    param="code",
                    code=status.HTTP_400_BAD_REQUEST,
                )
            if not generic_client_id:
                raise ProxyException(
                    message="GENERIC_CLIENT_ID must be set when PKCE is enabled",
                    type=ProxyErrorTypes.auth_error,
                    param="GENERIC_CLIENT_ID",
                    code=status.HTTP_401_UNAUTHORIZED,
                )
            if not generic_token_endpoint:
                raise ProxyException(
                    message="GENERIC_TOKEN_ENDPOINT must be set when PKCE is enabled",
                    type=ProxyErrorTypes.auth_error,
                    param="GENERIC_TOKEN_ENDPOINT",
                    code=status.HTTP_401_UNAUTHORIZED,
                )
            # All guards above raise, so authorization_code is a non-empty str here.
            # Use an explicit type guard rather than assert (assert is a no-op with -O).
            if not isinstance(authorization_code, str):
                raise ProxyException(
                    message="Missing authorization code in callback",
                    type=ProxyErrorTypes.auth_error,
                    param="code",
                    code=status.HTTP_400_BAD_REQUEST,
                )
            combined_response: Final = await SSOAuthenticationHandler._pkce_token_exchange(
                authorization_code=authorization_code,
                code_verifier=code_verifier,
                client_id=generic_client_id,
                client_secret=generic_client_secret,
                token_endpoint=generic_token_endpoint,
                userinfo_endpoint=generic_userinfo_endpoint,
                include_client_id=generic_include_client_id,
                redirect_url=redirect_url,
                additional_headers=additional_generic_sso_headers_dict,
            )
            # Pass the full response so custom response_convertor implementations
            # can access all fields (including id_token for claim extraction).
            result = response_convertor(combined_response, generic_sso)
            sso_assertion = assertion_from_sso_login(
                combined_response.get("id_token"), combined_response.get("refresh_token")
            )
            # In the PKCE path verify_and_process is skipped, so generic_sso.access_token
            # is never set. Read the token directly from the exchange response instead so
            # process_sso_jwt_access_token can extract JWT-embedded roles/teams.
            access_token_str: str | None = combined_response.get("access_token")
        else:
            result = await generic_sso.verify_and_process(
                request,
                params=token_exchange_params,
                headers=additional_generic_sso_headers_dict,
            )
            access_token_str = generic_sso.access_token
            sso_assertion = assertion_from_sso_login(generic_sso.id_token, generic_sso.refresh_token)

        access_token_payload = process_sso_jwt_access_token(
            access_token_str, sso_jwt_handler, result, role_mappings=role_mappings
        )
        # Delete the single-use PKCE verifier only after all downstream processing
        # (response_convertor and process_sso_jwt_access_token) has completed
        # successfully.  Deleting earlier would consume the verifier on a transient
        # failure, forcing the user to restart the entire OAuth flow from scratch.
        if pkce_cache_key:
            await SSOAuthenticationHandler._delete_pkce_verifier(pkce_cache_key)

    except Exception as e:
        _handle_generic_sso_error(
            e,
            generic_authorization_endpoint,
            generic_token_endpoint,
            additional_generic_sso_headers_dict,
        )
    verbose_proxy_logger.debug("generic result: %s", result)
    return result or {}, received_response, access_token_payload, sso_assertion


async def create_team_member_add_task(team_id, user_info):
    """Create a task for adding a member to a team."""
    try:
        member: Final = Member(user_id=user_info.user_id, role="user")
        team_member_add_request: Final = TeamMemberAddRequest(
            member=member,
            team_id=team_id,
        )
        return await team_member_add(
            data=team_member_add_request,
            user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN),
        )
    except Exception as e:
        verbose_proxy_logger.debug("[Non-Blocking] Error trying to add sso user to db: %s", e)


async def add_missing_team_member(user_info: NewUserResponse | LiteLLM_UserTable, sso_teams: list[str]):
    """
    - Get missing teams (diff b/w user_info.team_ids and sso_teams)
    - Add missing user to missing teams
    """
    # Handle None as empty list for new users
    user_teams: Final = user_info.teams if user_info.teams is not None else []
    missing_teams: Final = set(sso_teams) - set(user_teams)
    missing_teams_list: Final = list(missing_teams)
    tasks = []
    tasks = [create_team_member_add_task(team_id, user_info) for team_id in missing_teams_list]

    try:
        await asyncio.gather(*tasks)
    except Exception as e:
        verbose_proxy_logger.debug("[Non-Blocking] Error trying to add sso user to db: %s", e)


def get_disabled_non_admin_personal_key_creation():
    key_generation_settings: Final = litellm.key_generation_settings
    if key_generation_settings is None:
        return False
    personal_key_generation: Final = key_generation_settings.get("personal_key_generation") or {}
    allowed_user_roles: Final = personal_key_generation.get("allowed_user_roles") or []
    return bool("proxy_admin" in allowed_user_roles)


async def get_existing_user_info_from_db(
    user_id: str | None,
    user_email: str | None,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
) -> LiteLLM_UserTable | None:
    try:
        user_info = await get_user_object(
            user_id=user_id,
            user_email=user_email,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            user_id_upsert=False,
            parent_otel_span=None,
            proxy_logging_obj=proxy_logging_obj,
            sso_user_id=user_id,
        )
    except Exception as e:
        verbose_proxy_logger.debug("Error getting user object: %s", e)
        user_info = None

    return user_info


async def get_user_info_from_db(
    result: CustomOpenID | OpenID | dict,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    proxy_logging_obj: ProxyLogging,
    user_email: str | None,
    user_defined_values: SSOUserDefinedValues | None,
    alternate_user_id: str | None = None,
) -> LiteLLM_UserTable | NewUserResponse | None:
    try:
        potential_user_ids: Final = []
        if alternate_user_id is not None:
            potential_user_ids.append(alternate_user_id)
        if not isinstance(result, dict):
            _id = getattr(result, "id", None)
            if _id is not None and isinstance(_id, str):
                potential_user_ids.append(_id)
        else:
            _id = result.get("id", None)
            if _id is not None and isinstance(_id, str):
                potential_user_ids.append(_id)

        user_email = normalize_email(
            getattr(result, "email", None) if not isinstance(result, dict) else result.get("email", None)
        )

        user_info: LiteLLM_UserTable | NewUserResponse | None = None

        for user_id in potential_user_ids:
            user_info = await get_existing_user_info_from_db(
                user_id=user_id,
                user_email=user_email,
                prisma_client=prisma_client,
                user_api_key_cache=user_api_key_cache,
                proxy_logging_obj=proxy_logging_obj,
            )
            if user_info is not None:
                break

        verbose_proxy_logger.debug(
            "user_info: %s; litellm.default_internal_user_params: %s", user_info, litellm.default_internal_user_params
        )

        # Upsert SSO User to LiteLLM DB
        user_info = await SSOAuthenticationHandler.upsert_sso_user(
            result=result,
            user_info=user_info,
            user_email=user_email,
            user_defined_values=user_defined_values,
            prisma_client=prisma_client,
        )

        await SSOAuthenticationHandler.add_user_to_teams_from_sso_response(
            result=result,
            user_info=user_info,
        )

        return user_info
    except Exception as e:
        verbose_proxy_logger.exception("[Non-Blocking] Error trying to add sso user to db: %s", e)

    return None


def _should_use_role_from_sso_response(sso_role: str | None) -> bool:
    """returns true if SSO upsert should use the 'role' defined on the SSO response"""
    if sso_role is None:
        return False

    if not is_valid_litellm_user_role(sso_role):
        verbose_proxy_logger.debug(
            "SSO role '%s' is not a valid LiteLLM user role. Ignoring role from SSO response. See LitellmUserRoles enum for valid roles.",
            sso_role,
        )
        return False
    return True


def _build_sso_user_update_data(
    result: Union["CustomOpenID", OpenID, dict] | None,
    user_email: str | None,
    user_id: str | None,
) -> dict[str, object]:
    """
    Build the update data dictionary for SSO user upsert.

    Args:
        result: The SSO response containing user information
        user_email: The user's email from SSO
        user_id: The user's ID for logging purposes

    Returns:
        dict: Update data containing user_email and optionally user_role if valid
    """
    update_data: Final[dict[str, object]] = {"user_email": normalize_email(user_email)}

    # Get SSO role from result and include if valid
    sso_role: Final = getattr(result, "user_role", None)
    if sso_role is not None:
        # Convert enum to string if needed
        sso_role_str: Final = sso_role.value if isinstance(sso_role, LitellmUserRoles) else sso_role

        # Only include if it's a valid LiteLLM role
        if _should_use_role_from_sso_response(sso_role_str):
            update_data["user_role"] = sso_role_str
            verbose_proxy_logger.info("Updating user %s role from SSO: %s", user_id, sso_role_str)

    return update_data


async def _sync_user_role_from_jwt_role_map(
    jwt_handler: JWTHandler | None,
    received_response: dict | None,
    user_info: LiteLLM_UserTable | NewUserResponse | None,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    user_defined_values: SSOUserDefinedValues | None,
) -> None:
    """
    Apply jwt_litellm_role_map during SSO login.

    When jwt_litellm_role_map is configured with sync_user_role_and_teams=True,
    this ensures SSO users get the same role mapping as API/JWT users. Without
    this, the SSO path falls back to INTERNAL_USER_VIEW_ONLY for roles that
    don't directly match LitellmUserRoles enum values.
    """
    if jwt_handler is None or received_response is None:
        return
    if not jwt_handler.litellm_jwtauth.sync_user_role_and_teams:
        return
    if not jwt_handler.litellm_jwtauth.jwt_litellm_role_map:
        return

    mapped_role: Final = jwt_handler.map_jwt_role_to_litellm_role(received_response)
    if mapped_role is None:
        return

    verbose_proxy_logger.info("SSO jwt_litellm_role_map matched role: %s", mapped_role.value)

    # Update user_defined_values so downstream code uses the mapped role
    if user_defined_values is not None:
        user_defined_values["user_role"] = mapped_role.value

    # Update existing DB record if role differs
    if user_info is not None and user_info.user_role != mapped_role.value:
        await _user_meta_db(UserRepository(prisma_client)).update(
            where={"user_id": user_info.user_id},
            data={"user_role": mapped_role.value},
        )
        user_info.user_role = mapped_role.value
        await user_api_key_cache.async_set_cache(
            key=user_info.user_id,
            value=user_info,
            model_type=LiteLLM_UserTable,
        )


def apply_user_info_values_to_sso_user_defined_values(
    user_info: LiteLLM_UserTable | NewUserResponse | None,
    user_defined_values: SSOUserDefinedValues | None,
) -> SSOUserDefinedValues | None:
    if user_defined_values is None:
        return None
    if user_info is not None and user_info.user_id is not None:
        user_defined_values["user_id"] = user_info.user_id

    # SSO role takes precedence - only use DB role if SSO didn't provide one
    # This ensures SSO is the authoritative source for user roles
    sso_role: Final = user_defined_values.get("user_role")
    db_role: Final = user_info.user_role if user_info else None

    if _should_use_role_from_sso_response(sso_role):
        # SSO provided a valid role, keep it and log that we're using it
        verbose_proxy_logger.info("Using SSO role: %s (DB role was: %s)", sso_role, db_role)
    else:
        # SSO didn't provide a valid role, fall back to DB role or default
        if user_info is None or user_info.user_role is None:
            user_defined_values["user_role"] = LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value
            verbose_proxy_logger.debug("No SSO or DB role found, using default: INTERNAL_USER_VIEW_ONLY")
        else:
            user_defined_values["user_role"] = user_info.user_role
            verbose_proxy_logger.debug("Using DB role: %s", user_info.user_role)

    # Preserve the user's existing models from the database
    if user_info is not None and hasattr(user_info, "models") and user_info.models:
        user_defined_values["models"] = user_info.models

    return user_defined_values


async def check_and_update_if_proxy_admin_id(user_role: str, user_id: str, prisma_client: PrismaClient | None):
    """
    - Check if user role in DB is admin
    - If not, update user role in DB to admin role
    """
    proxy_admin_id: Final = os.getenv("PROXY_ADMIN_ID")
    if proxy_admin_id is not None and proxy_admin_id == user_id:
        if user_role and user_role == LitellmUserRoles.PROXY_ADMIN.value:
            return user_role

        if prisma_client:
            await _user_meta_db(UserRepository(prisma_client)).update(
                where={"user_id": user_id},
                data={"user_role": LitellmUserRoles.PROXY_ADMIN.value},
            )

        user_role = LitellmUserRoles.PROXY_ADMIN.value

    return user_role


@router.get("/sso/callback", tags=["experimental"], include_in_schema=False)
async def auth_callback(request: Request, state: str | None = None):
    """Verify login"""
    verbose_proxy_logger.info("Starting SSO callback with state: %s", state)

    oauth_error: Final = request.query_params.get("error")
    if oauth_error:
        oauth_error_description: Final = request.query_params.get("error_description")
        verbose_proxy_logger.warning(
            "SSO callback received OAuth error: %s, description: %s", oauth_error, oauth_error_description
        )
        raise HTTPException(
            status_code=401,
            detail=f"OAuth error: {oauth_error}"
            + (f", error_description: {oauth_error_description}" if oauth_error_description else ""),
        )

    # Check if this is a CLI login (state starts with our CLI prefix)
    from litellm.constants import LITELLM_CLI_SESSION_TOKEN_PREFIX
    from litellm.proxy._types import LiteLLM_JWTAuth
    from litellm.proxy.auth.handle_jwt import JWTHandler
    from litellm.proxy.proxy_server import (
        general_settings,
        jwt_handler,
        master_key,
        prisma_client,
        user_api_key_cache,
    )

    if prisma_client is None:
        raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)

    sso_jwt_handler: JWTHandler | None = None
    ui_access_mode: Final = general_settings.get("ui_access_mode", None)
    if ui_access_mode is not None and isinstance(ui_access_mode, dict):
        sso_jwt_handler = JWTHandler()
        sso_jwt_handler.update_environment(
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            litellm_jwtauth=LiteLLM_JWTAuth(
                team_ids_jwt_field=general_settings.get("ui_access_mode", {}).get("sso_group_jwt_field", None),
            ),
            leeway=0,
        )

    microsoft_client_id: Final = os.getenv("MICROSOFT_CLIENT_ID", None)
    google_client_id: Final = os.getenv("GOOGLE_CLIENT_ID", None)
    generic_client_id: Final = os.getenv("GENERIC_CLIENT_ID", None)
    received_response: dict | None = None
    access_token_payload: dict | None = None
    sso_assertion: SSOIdentityAssertion | None = None
    # get url from request
    if master_key is None:
        raise ProxyException(
            message="Master Key not set for Proxy. Please set Master Key to use Admin UI. Set `LITELLM_MASTER_KEY` in .env or set general_settings:master_key in config.yaml.  https://docs.litellm.ai/docs/proxy/virtual_keys. If set, use `--detailed_debug` to debug issue.",
            type=ProxyErrorTypes.auth_error,
            param="master_key",
            code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )
    redirect_url = SSOAuthenticationHandler.get_redirect_url_for_sso(request=request, sso_callback_route="sso/callback")

    verbose_proxy_logger.info("Redirecting to %s", redirect_url)
    result = None
    if google_client_id is not None:
        result = await GoogleSSOHandler.get_google_callback_response(
            request=request,
            google_client_id=google_client_id,
            redirect_url=redirect_url,
        )
    elif microsoft_client_id is not None:
        result = await MicrosoftSSOHandler.get_microsoft_callback_response(
            request=request,
            microsoft_client_id=microsoft_client_id,
            redirect_url=redirect_url,
        )

    elif generic_client_id is not None:
        (
            result,
            received_response,
            access_token_payload,
            sso_assertion,
        ) = await get_generic_sso_response(
            request=request,
            jwt_handler=jwt_handler,
            generic_client_id=generic_client_id,
            redirect_url=redirect_url,
            sso_jwt_handler=sso_jwt_handler,
        )

    if result is None:
        raise HTTPException(
            status_code=401,
            detail="Result not returned by SSO provider.",
        )

    if state and state.startswith(f"{LITELLM_CLI_SESSION_TOKEN_PREFIX}:"):
        # State format: {PREFIX}:{login_id}[:{user_code}]
        state_parts: Final = state.split(":", 2)
        key_id: Final = state_parts[1] if len(state_parts) > 1 else None
        prefill_user_code: Final = state_parts[2] if len(state_parts) > 2 else None

        verbose_proxy_logger.info("CLI SSO callback detected")
        return await cli_sso_callback(
            request=request,
            key=key_id,
            prefill_user_code=prefill_user_code,
            result=result,
            received_response=received_response,
            sso_assertion=sso_assertion,
        )

    # Control-plane cross-origin: read return_to from cookie.
    # Starlette's cookie_parser already handles RFC 2109 unquoting.
    cp_return_to: Final[str | None] = request.cookies.get("litellm_cp_return_to")

    return await SSOAuthenticationHandler.get_redirect_response_from_openid(
        result=result,
        request=request,
        received_response=received_response,
        generic_client_id=generic_client_id,
        ui_access_mode=ui_access_mode,
        access_token_payload=access_token_payload,
        jwt_handler=jwt_handler,
        return_to=cp_return_to,
        sso_assertion=sso_assertion,
    )


@router.get("/sso/saml/login", tags=["experimental"], include_in_schema=False)
async def saml_login(request: Request, return_to: str | None = None):
    """SP-initiated SAML login. Redirects the user to the configured IdP."""
    from litellm.proxy.proxy_server import user_api_key_cache

    _disable_ui_flag: Final = os.getenv("DISABLE_ADMIN_UI")
    if _disable_ui_flag is not None and str_to_bool(value=_disable_ui_flag):
        return admin_ui_disabled()

    return await SAMLAuthHandler.build_login_redirect(request=request, cache=user_api_key_cache, relay_state=return_to)


@router.get("/sso/saml/metadata", tags=["experimental"], include_in_schema=False)
async def saml_metadata(request: Request):
    """Service Provider metadata XML, for registering this proxy at the IdP."""
    from litellm.proxy.proxy_server import user_api_key_cache

    metadata: Final = await SAMLAuthHandler.build_sp_metadata(request=request, cache=user_api_key_cache)
    return Response(content=metadata, media_type="application/xml")


@router.post("/sso/saml/callback", tags=["experimental"], include_in_schema=False)
async def saml_callback(request: Request):
    """Assertion Consumer Service. Validates the IdP assertion and issues a UI session."""
    from litellm.proxy.proxy_server import (
        general_settings,
        jwt_handler,
        master_key,
        premium_user,
        prisma_client,
        user_api_key_cache,
    )

    _disable_ui_flag: Final = os.getenv("DISABLE_ADMIN_UI")
    if _disable_ui_flag is not None and str_to_bool(value=_disable_ui_flag):
        return admin_ui_disabled()

    if prisma_client is None:
        raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)
    if master_key is None:
        raise ProxyException(
            message="Master Key not set for Proxy. Set `LITELLM_MASTER_KEY` in .env or general_settings:master_key in config.yaml.",
            type=ProxyErrorTypes.auth_error,
            param="master_key",
            code=status.HTTP_500_INTERNAL_SERVER_ERROR,
        )

    post_data: Final = await SAMLAuthHandler.read_acs_post_data(request)
    if "SAMLResponse" not in post_data:
        raise HTTPException(status_code=400, detail="Missing SAMLResponse in callback request.")

    result: Final = await SAMLAuthHandler.handle_acs(request=request, cache=user_api_key_cache, post_data=post_data)

    await _raise_if_sso_exceeds_free_user_limit(premium_user, prisma_client)

    ui_access_mode: Final = general_settings.get("ui_access_mode", None)
    relay_state: Final = post_data.get("RelayState")
    cp_return_to: Final[str | None] = (
        relay_state
        if isinstance(relay_state, str) and SSOAuthenticationHandler._validate_return_to(relay_state)
        else None
    )

    return await SSOAuthenticationHandler.get_redirect_response_from_openid(
        result=result,
        request=request,
        received_response=None,
        generic_client_id=None,
        ui_access_mode=ui_access_mode,
        access_token_payload=None,
        jwt_handler=jwt_handler,
        return_to=cp_return_to,
    )


async def _build_cli_sso_user_defined_values(
    result: OpenID | dict,
    parsed_openid_result: ParsedOpenIDResult,
) -> SSOUserDefinedValues | None:
    from litellm.proxy.proxy_server import user_custom_sso

    custom_sso_handler: Final[_CustomSsoCall | None] = user_custom_sso
    user_id: Final = parsed_openid_result.get("user_id")
    if custom_sso_handler is not None:
        if inspect.iscoroutinefunction(custom_sso_handler):
            return await custom_sso_handler(result)
        raise ValueError("user_custom_sso must be a coroutine function")
    if user_id is None:
        return None
    return SSOUserDefinedValues(
        models=[],
        user_id=user_id,
        user_email=parsed_openid_result.get("user_email"),
        max_budget=litellm.max_internal_user_budget,
        user_role=parsed_openid_result.get("user_role"),
        budget_duration=litellm.internal_user_budget_duration,
    )


def _cli_sso_team_detail(team_row: Mapping[str, object]) -> CliSsoTeamDetail:
    team: Final = _TeamRowGrants.model_validate(team_row)
    alias_table: Final = team.litellm_model_table
    return CliSsoTeamDetail(
        team_id=team.team_id,
        team_alias=team.team_alias,
        team_models=team.models,
        team_model_aliases=alias_table.model_aliases if alias_table is not None else None,
    )


async def fetch_cli_sso_team_details(
    prisma_client: PrismaClient,
    teams: Sequence[str],
) -> tuple[CliSsoTeamDetail, ...] | None:
    """``None`` means the lookup itself failed, which is not the same as the user having no teams."""
    if not teams:
        return ()
    try:
        prisma_teams: Final = await _team_detail_db(TeamRepository(prisma_client)).find_many(
            where={"team_id": {"in": teams}},
            include={"litellm_model_table": True},
        )
    except Exception as e:
        verbose_proxy_logger.error("Error fetching team details for CLI SSO session: %s", e)
        return None
    return tuple(_cli_sso_team_detail(team_row.model_dump()) for team_row in prisma_teams)


def _cli_sso_session_teams(team_details: Sequence[CliSsoTeamDetail]) -> list[str]:
    """The teams a login may bind to: only those whose row still exists.

    A team deleted out from under a membership, which is what deleting an organization
    leaves behind, can never resolve its grants, so offering it would refuse every
    future login for that user with nothing they could do to recover.
    """
    return [detail.team_id for detail in team_details if detail.team_id is not None]


def selected_cli_sso_team_detail(team_details: object, team_id: str | None) -> CliSsoTeamDetail | None:
    """``None`` means the team's grants are unknown. An empty grant is a real value meaning unrestricted,
    so an unknown one must not be minted as empty."""
    if team_id is None:
        return _TEAMLESS_CLI_SSO_TEAM_DETAIL
    try:
        details: Final = _CLI_SSO_TEAM_DETAILS_ADAPTER.validate_python(team_details)
    except ValidationError:
        return None
    return next((detail for detail in details if detail.team_id == team_id), None)


async def _complete_cli_sso_callback_session(
    *,
    request: Request,
    key: str,
    flow: dict,
    result: OpenID | dict,
    parsed_openid_result: ParsedOpenIDResult,
    user_defined_values: SSOUserDefinedValues | None,
    prisma_client: PrismaClient,
    user_api_key_cache: UserApiKeyCache,
    cli_sso_session_cache: DualCache,
    proxy_logging_obj: ProxyLogging,
    prefill_user_code: str | None = None,
    sso_assertion: SSOIdentityAssertion | None = None,
):
    from fastapi.responses import HTMLResponse

    user_id: Final = parsed_openid_result.get("user_id")
    user_email: Final = parsed_openid_result.get("user_email")
    user_info: Final = await get_user_info_from_db(
        result=result,
        prisma_client=prisma_client,
        user_api_key_cache=user_api_key_cache,
        proxy_logging_obj=proxy_logging_obj,
        user_email=user_email,
        user_defined_values=user_defined_values,
        alternate_user_id=user_id,
    )
    if user_info is None:
        raise HTTPException(status_code=500, detail="Failed to retrieve user information from SSO")
    if not user_info.user_id:
        raise HTTPException(status_code=500, detail="Failed to retrieve user information from SSO")

    await retain_sso_identity_assertion_for_ema(user_id=user_info.user_id, assertion=sso_assertion)

    teams: list[str] = []
    if hasattr(user_info, "teams") and user_info.teams:
        teams = user_info.teams if isinstance(user_info.teams, list) else []

    team_details: Final = await fetch_cli_sso_team_details(prisma_client=prisma_client, teams=teams)
    if team_details is None:
        raise HTTPException(
            status_code=500,
            detail="Could not resolve team model grants for this login. Please try again",
        )
    resolved_teams: Final = _cli_sso_session_teams(team_details)
    attribution_metadata: Final = build_cli_sso_attribution_metadata(result=result)
    if attribution_metadata:
        await _persist_cli_sso_user_metadata(
            prisma_client=prisma_client,
            user_id=cast(str, user_info.user_id),
            attribution_metadata=attribution_metadata,
        )

    flow["session_data"] = {
        "user_id": cast(str, user_info.user_id),
        "user_role": user_info.user_role,
        "models": user_info.models if hasattr(user_info, "models") else [],
        "user_email": user_email,
        "teams": resolved_teams,
        "team_details": [detail.model_dump() for detail in team_details],
        "attribution_metadata": attribution_metadata,
    }
    flow["sso_complete"] = True
    browser_complete_token: Final = secrets.token_urlsafe(32)
    flow["browser_complete_token_hash"] = _hash_cli_sso_secret(browser_complete_token)
    _set_cli_sso_flow(login_id=key, cache=cli_sso_session_cache, flow=flow)

    verbose_proxy_logger.info(
        "Stored CLI SSO session for user: %s, teams: %s, num_teams: %s",
        user_info.user_id,
        resolved_teams,
        len(resolved_teams),
    )
    verify_url: Final = get_custom_url(
        request_base_url=str(request.base_url),
        route=f"sso/cli/complete/{key}",
    )
    return HTMLResponse(
        content=_render_cli_sso_verification_page(
            verify_url=verify_url,
            browser_complete_token=browser_complete_token,
            prefill_user_code=prefill_user_code,
        ),
        status_code=200,
    )


async def cli_sso_callback(
    request: Request,
    key: str | None = None,
    result: OpenID | dict | None = None,
    received_response: dict | None = None,
    prefill_user_code: str | None = None,
    sso_assertion: SSOIdentityAssertion | None = None,
):
    """CLI SSO callback - stores session info for JWT generation on polling"""
    verbose_proxy_logger.info("CLI SSO callback")

    from litellm.proxy.proxy_server import (
        cli_sso_session_cache,
        general_settings,
        prisma_client,
        proxy_logging_obj,
        user_api_key_cache,
    )

    flow: Final = _get_cli_sso_flow_or_raise(login_id=key, cache=cli_sso_session_cache)

    if prisma_client is None:
        raise HTTPException(status_code=500, detail=CommonProxyErrors.db_not_connected_error.value)

    if result is None:
        raise HTTPException(
            status_code=500,
            detail="SSO authentication failed - no result returned from provider",
        )

    # After None check, cast to non-None type for type checker
    result_non_none: Final[OpenID | dict] = cast(OpenID | dict, result)

    try:
        parsed_openid_result: Final = SSOAuthenticationHandler._get_user_email_and_id_from_result(
            result=result_non_none,
            generic_client_id=os.getenv("GENERIC_CLIENT_ID", None),
        )
        verbose_proxy_logger.debug("parsed_openid_result: %s", parsed_openid_result)
        user_defined_values: Final = await _build_cli_sso_user_defined_values(
            result=result_non_none,
            parsed_openid_result=parsed_openid_result,
        )

        SSOAuthenticationHandler.verify_user_in_restricted_sso_group(
            general_settings=general_settings,
            result=result_non_none,
            received_response=received_response,
        )

        return await _complete_cli_sso_callback_session(
            request=request,
            key=cast(str, key),
            flow=flow,
            result=result_non_none,
            parsed_openid_result=parsed_openid_result,
            user_defined_values=user_defined_values,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            cli_sso_session_cache=cli_sso_session_cache,
            proxy_logging_obj=proxy_logging_obj,
            prefill_user_code=prefill_user_code,
            sso_assertion=sso_assertion,
        )
    except ProxyException:
        raise
    except HTTPException:
        raise
    except Exception as e:
        verbose_proxy_logger.error("Error with CLI SSO callback: %s", e)
        raise HTTPException(status_code=500, detail=f"Failed to process CLI SSO: {e}")


@router.get("/sso/cli/poll/{key_id}", tags=["experimental"], include_in_schema=False)
async def cli_poll_key(
    key_id: str,
    team_id: str | None = None,
    x_litellm_cli_poll_secret: str | None = Header(default=None),
):
    """
    CLI polling endpoint - retrieves session from cache and generates JWT.

    Flow:
    1. First poll (no team_id): Returns teams list without generating JWT
    2. Second poll (with team_id): Generates JWT with selected team and deletes session

    Args:
        key_id: The CLI login session ID
        team_id: Optional team ID to assign to the JWT. If provided, must be one of user's teams.
    """
    from litellm.proxy.auth.auth_checks import ExperimentalUIJWTToken
    from litellm.proxy.proxy_server import cli_sso_session_cache

    try:
        flow: Final = _get_cli_sso_flow_or_raise(login_id=key_id, cache=cli_sso_session_cache)
        if not _verify_cli_sso_poll_secret(flow=flow, poll_secret=x_litellm_cli_poll_secret):
            raise HTTPException(status_code=403, detail="Invalid CLI polling secret")

        if not flow.get("sso_complete") or not flow.get("user_code_verified"):
            return {"status": "pending"}

        session_data: Final = flow.get("session_data")

        if isinstance(session_data, dict):
            user_teams: Final = session_data.get("teams", [])
            user_team_details: Final = session_data.get("team_details")
            user_id: Final = session_data["user_id"]

            verbose_proxy_logger.info(
                "CLI poll: user=%s, team_id=%s, user_teams=%s, num_teams=%s",
                user_id,
                team_id,
                user_teams,
                len(user_teams),
            )

            # If no team_id provided and user has teams, return teams list for selection
            # Don't generate JWT yet - let CLI select a team first. For newer
            # clients we return rich team details (id + alias); older clients
            # can continue to rely on the simple "teams" list.
            if team_id is None and len(user_teams) > 1:
                verbose_proxy_logger.info("Returning teams list for user %s to select from: %s", user_id, user_teams)
                # Best-effort construction of team_details if it wasn't
                # already cached for some reason.
                team_details_response: list[dict[str, object]] | None = None
                if isinstance(user_team_details, list) and user_team_details:
                    team_details_response = user_team_details
                elif user_teams:
                    team_details_response = [{"team_id": t, "team_alias": None} for t in user_teams]
                poll_response: dict[str, object] = {
                    "status": "ready",
                    "user_id": user_id,
                    "teams": user_teams,
                    "team_details": team_details_response,
                    "requires_team_selection": True,
                }
                attribution_metadata = _cli_poll_attribution_metadata_from_session(session_data)
                if attribution_metadata:
                    poll_response["attribution_metadata"] = attribution_metadata
                return poll_response

            # Validate team_id if provided
            if team_id is not None:
                if team_id not in user_teams:
                    raise HTTPException(
                        status_code=403,
                        detail=f"User does not belong to team: {team_id}. Available teams: {user_teams}",
                    )
            else:
                # If no team_id provided and user has 0 or 1 team, use first team (or None)
                team_id = user_teams[0] if len(user_teams) > 0 else None

            selected_team: Final = selected_cli_sso_team_detail(
                team_details=user_team_details,
                team_id=team_id,
            )
            if selected_team is None:
                raise HTTPException(
                    status_code=500,
                    detail=f"Could not resolve the model grants for team: {team_id}. Please run `lite login` again",
                )

            user_info: Final = LiteLLM_UserTable(
                user_id=user_id,
                user_role=session_data["user_role"],
                models=session_data.get("models", []),
            )

            jwt_token: Final = ExperimentalUIJWTToken.get_cli_jwt_auth_token(
                user_info=user_info,
                team_id=team_id,
                team_alias=selected_team.team_alias,
                team_models=selected_team.team_models,
                team_model_aliases=selected_team.team_model_aliases,
                max_budget=None,
            )

            # Delete cache entry (single-use)
            cli_sso_session_cache.delete_cache(key=_get_cli_sso_flow_cache_key(key_id))

            verbose_proxy_logger.info("CLI JWT generated for user: %s, team: %s", user_id, team_id)
            poll_response = {
                "status": "ready",
                "key": jwt_token,
                "user_id": user_id,
                "team_id": team_id,
                "teams": user_teams,
                # Echo back any team details we have so clients can
                # present nicer information if needed.
                "team_details": user_team_details,
            }
            attribution_metadata = _cli_poll_attribution_metadata_from_session(session_data)
            if attribution_metadata:
                poll_response["attribution_metadata"] = attribution_metadata
            return poll_response
        else:
            return {"status": "pending"}

    except HTTPException:
        raise
    except Exception as e:
        verbose_proxy_logger.error("Error polling for CLI JWT: %s", e)
        raise HTTPException(status_code=500, detail=f"Error checking session status: {e}")


async def insert_sso_user(
    result_openid: OpenID | dict | None,
    user_defined_values: SSOUserDefinedValues | None = None,
) -> NewUserResponse:
    """
    Helper function to create a New User in LiteLLM DB after a successful SSO login

    Args:
        result_openid (OpenID): User information in OpenID format if the login was successful.
        user_defined_values (Optional[SSOUserDefinedValues], optional): LiteLLM SSOValues / fields that were read

    Returns:
        Tuple[str, str]: User ID and User Role
    """
    verbose_proxy_logger.debug("Inserting SSO user into DB. User values: %s", user_defined_values)
    if result_openid is None:
        raise ValueError("result_openid is None")
    if isinstance(result_openid, dict):
        result_openid = OpenID(**result_openid)

    if user_defined_values is None:
        raise ValueError("user_defined_values is None")

    # Apply default_internal_user_params
    if litellm.default_internal_user_params:
        # Preserve the SSO-extracted role if it's a valid LiteLLM role,
        # regardless of how it was determined (role_mappings, Microsoft app_roles,
        # GENERIC_USER_ROLE_ATTRIBUTE, custom SSO handler, etc.)
        sso_role: Final = user_defined_values.get("user_role")
        if _should_use_role_from_sso_response(sso_role):
            # Preserve the SSO-extracted role, but apply other defaults
            preserved_role: Final = sso_role
            user_defined_values.update(litellm.default_internal_user_params)
            user_defined_values["user_role"] = preserved_role  # Restore preserved role
            verbose_proxy_logger.debug("Preserved SSO-extracted role '%s'", preserved_role)
        else:
            # SSO didn't provide a valid role, apply all defaults including role
            user_defined_values.update(litellm.default_internal_user_params)

    # Set budget for internal users
    if user_defined_values.get("user_role") == LitellmUserRoles.INTERNAL_USER.value:
        if user_defined_values.get("max_budget") is None:
            user_defined_values["max_budget"] = litellm.max_internal_user_budget
        if user_defined_values.get("budget_duration") is None:
            user_defined_values["budget_duration"] = litellm.internal_user_budget_duration

    if user_defined_values["user_role"] is None:
        user_defined_values["user_role"] = LitellmUserRoles.INTERNAL_USER_VIEW_ONLY

    new_user_request: Final = NewUserRequest(
        user_id=user_defined_values["user_id"],
        user_email=normalize_email(user_defined_values["user_email"]),
        user_role=user_defined_values["user_role"],
        max_budget=user_defined_values["max_budget"],
        budget_duration=user_defined_values["budget_duration"],
        sso_user_id=user_defined_values["user_id"],
        auto_create_key=False,
    )

    if result_openid and hasattr(result_openid, "provider"):
        new_user_request.metadata = {"auth_provider": getattr(result_openid, "provider")}

    response: Final = await new_user(
        data=new_user_request,
        user_api_key_dict=UserAPIKeyAuth(user_role=LitellmUserRoles.PROXY_ADMIN),
    )

    return response


@router.get(
    "/sso/get/ui_settings",
    tags=["experimental"],
    include_in_schema=False,
    dependencies=[Depends(user_api_key_auth)],
)
async def get_ui_settings(request: Request):
    from litellm.proxy.proxy_server import general_settings, proxy_state

    _proxy_base_url: Final = os.getenv("PROXY_BASE_URL", None)
    _logout_url: Final = os.getenv("PROXY_LOGOUT_URL", None)
    _api_doc_base_url: Final = os.getenv("LITELLM_UI_API_DOC_BASE_URL", None)
    _is_sso_enabled: Final = has_user_setup_sso()
    disable_expensive_db_queries: Final = (
        proxy_state.get_proxy_state_variable("spend_logs_row_count") > MAX_SPENDLOG_ROWS_TO_QUERY
    )
    default_team_disabled = general_settings.get("default_team_disabled", False)
    if "PROXY_DEFAULT_TEAM_DISABLED" in os.environ:
        if os.environ["PROXY_DEFAULT_TEAM_DISABLED"].lower() == "true":
            default_team_disabled = True

    return {
        "PROXY_BASE_URL": _proxy_base_url,
        "PROXY_LOGOUT_URL": _logout_url,
        "LITELLM_UI_API_DOC_BASE_URL": _api_doc_base_url,
        "DEFAULT_TEAM_DISABLED": default_team_disabled,
        "SSO_ENABLED": _is_sso_enabled,
        "NUM_SPEND_LOGS_ROWS": proxy_state.get_proxy_state_variable("spend_logs_row_count"),
        "DISABLE_EXPENSIVE_DB_QUERIES": disable_expensive_db_queries,
    }


@router.get(
    "/sso/readiness",
    tags=["experimental"],
    dependencies=[Depends(user_api_key_auth)],
)
async def sso_readiness():
    """
    Health endpoint for checking SSO readiness.
    Checks if the configured SSO provider has all required environment variables set in memory.
    """
    microsoft_client_id: Final = os.getenv("MICROSOFT_CLIENT_ID", None)
    google_client_id: Final = os.getenv("GOOGLE_CLIENT_ID", None)
    generic_client_id: Final = os.getenv("GENERIC_CLIENT_ID", None)

    # Determine which SSO provider is configured
    configured_provider = None
    if google_client_id is not None:
        configured_provider = "google"
    elif microsoft_client_id is not None:
        configured_provider = "microsoft"
    elif generic_client_id is not None:
        configured_provider = "generic"

    # If no SSO is configured, return healthy (SSO is optional)
    if configured_provider is None:
        return {
            "status": "healthy",
            "sso_configured": False,
            "message": "No SSO provider configured",
        }

    # Check required environment variables for the configured provider
    missing_vars: Final = []

    if configured_provider == "google":
        google_client_secret: Final = os.getenv("GOOGLE_CLIENT_SECRET", None)
        if google_client_secret is None:
            missing_vars.append("GOOGLE_CLIENT_SECRET")

    elif configured_provider == "microsoft":
        microsoft_client_secret: Final = os.getenv("MICROSOFT_CLIENT_SECRET", None)
        microsoft_tenant: Final = os.getenv("MICROSOFT_TENANT", None)
        if microsoft_client_secret is None:
            missing_vars.append("MICROSOFT_CLIENT_SECRET")
        if microsoft_tenant is None:
            missing_vars.append("MICROSOFT_TENANT")

    elif configured_provider == "generic":
        generic_client_secret: Final = os.getenv("GENERIC_CLIENT_SECRET", None)
        generic_authorization_endpoint: Final = os.getenv("GENERIC_AUTHORIZATION_ENDPOINT", None)
        generic_token_endpoint: Final = os.getenv("GENERIC_TOKEN_ENDPOINT", None)
        generic_userinfo_endpoint: Final = os.getenv("GENERIC_USERINFO_ENDPOINT", None)
        if generic_client_secret is None:
            missing_vars.append("GENERIC_CLIENT_SECRET")
        if generic_authorization_endpoint is None:
            missing_vars.append("GENERIC_AUTHORIZATION_ENDPOINT")
        if generic_token_endpoint is None:
            missing_vars.append("GENERIC_TOKEN_ENDPOINT")
        if generic_userinfo_endpoint is None:
            missing_vars.append("GENERIC_USERINFO_ENDPOINT")

    # If all required variables are present, return healthy
    if len(missing_vars) == 0:
        return {
            "status": "healthy",
            "sso_configured": True,
            "provider": configured_provider,
            "message": f"{configured_provider.capitalize()} SSO is properly configured",
        }

    # If some variables are missing, return unhealthy
    raise HTTPException(
        status_code=503,
        detail={
            "status": "unhealthy",
            "sso_configured": True,
            "provider": configured_provider,
            "missing_environment_variables": missing_vars,
            "message": f"{configured_provider.capitalize()} SSO is configured but missing required environment variables: {', '.join(missing_vars)}",
        },
    )


def _is_same_origin_return_path(return_to: str) -> bool:
    """True for a strictly relative return path that stays on the gateway's own origin by
    construction, and is therefore safe to honor without a configured ``control_plane_url``.
    Used by the MCP gateway DCR authorize round-trip so a browser sent through login lands
    back on the authorize request.

    Requires a single leading ``/`` (not protocol-relative ``//``), no backslash (browsers
    fold ``\\`` to ``/``, so ``/\\evil.com`` would escape the origin), and no control or
    whitespace characters. Rejecting control chars keeps a ``\\r\\n``/tab-bearing value out
    of the redirect ``Location`` and the ``litellm_cp_return_to`` cookie entirely, rather
    than relying on downstream header encoding to neutralize it."""
    if not return_to.startswith("/") or return_to.startswith("//") or "\\" in return_to:
        return False
    return not any(ord(ch) < 0x20 or ch in (" ", "\x7f") for ch in return_to)


async def _sso_return_to_redirect(
    return_to: str | None,
    jwt_token: str,
    redis_usage_cache,
    user_api_key_cache,
) -> RedirectResponse | None:
    """Resolve the post-SSO redirect for a ``return_to``, or None to fall through to the dashboard.

    Two arms, both clearing the one-shot ``litellm_cp_return_to`` cookie:
    - **Same-origin relative path** (the MCP gateway DCR authorize round-trip): set the session cookie
      exactly like the dashboard path, then send the browser back where it came from.
    - **Control-plane cross-origin** (``control_plane_url``): stash the JWT behind a single-use opaque
      code (60s TTL) so the token never lands in browser history/logs; the control plane redeems it via
      ``POST /v3/login/exchange``.

    Extracted from ``get_redirect_response_from_openid`` to keep that method inside the complexity
    budget; behavior is identical to the inline arms it replaces (including letting
    ``_validate_return_to`` raise for a mismatched absolute return_to, as before)."""
    if return_to is None:
        return None

    if _is_same_origin_return_path(return_to):
        redirect_response = RedirectResponse(url=return_to, status_code=303)
        redirect_response.set_cookie(key="token", value=jwt_token)
        redirect_response.delete_cookie("litellm_cp_return_to")
        return redirect_response

    if SSOAuthenticationHandler._validate_return_to(return_to):
        code: Final = secrets.token_urlsafe(32)
        cache_key: Final = f"login_code:{code}"
        cache_value: Final = {"token": jwt_token, "redirect_url": return_to}
        if redis_usage_cache is not None:
            await redis_usage_cache.async_set_cache(key=cache_key, value=cache_value, ttl=60)
        else:
            await user_api_key_cache.async_set_cache(key=cache_key, value=cache_value, ttl=60)

        separator: Final = "&" if "?" in return_to else "?"
        redirect_url: Final = return_to + separator + urlencode({"login": "success", "code": code})
        verbose_proxy_logger.info("Cross-origin SSO: redirecting to control plane with login code")
        redirect_response = RedirectResponse(url=redirect_url, status_code=303)
        redirect_response.delete_cookie("litellm_cp_return_to")
        return redirect_response

    return None


def _persist_return_to_cookie(response: Response, return_to: str | None) -> None:
    """Best-effort: persist a SAFE ``return_to`` on ``response`` as the one-shot ``litellm_cp_return_to``
    cookie so ANY sign-in path — SSO / Okta / generic OR the username/password form — can resume there
    afterwards. THIS is the single source of truth, called by every sign-in branch so they cannot
    diverge (a per-branch reimplementation is exactly how the two drifted before). Honors a strictly
    relative same-origin path, and (when ``control_plane_url`` is configured) a return_to matching that
    origin. It NEVER raises: a mismatched or invalid ``return_to`` is simply not stored, so it can never
    block sign-in — the login entrypoint must always render."""
    if return_to is None:
        return
    try:
        safe: Final = _is_same_origin_return_path(return_to) or SSOAuthenticationHandler._validate_return_to(return_to)
    except HTTPException:
        return  # a non-matching absolute return_to is ignored, never blocks sign-in
    if safe:
        response.set_cookie(
            key="litellm_cp_return_to",
            value=return_to,
            max_age=600,
            httponly=True,
            samesite="lax",
        )


class SSOAuthenticationHandler:
    """
    Handler for SSO Authentication across all SSO providers
    """

    @staticmethod
    def _validate_return_to(return_to: str) -> bool:
        """
        Validate that return_to matches the configured control_plane_url origin.

        Returns True if return_to is valid and should be used.
        Returns False if control_plane_url is not configured (return_to is ignored).
        Raises HTTPException(400) if return_to origin does not match control_plane_url origin.
        """
        from litellm.proxy.proxy_server import general_settings

        control_plane_url: Final = general_settings.get("control_plane_url")
        if control_plane_url is None:
            return False

        def _origin(url: str) -> tuple:
            parsed: Final = urlparse(url)
            scheme: Final = (parsed.scheme or "").lower()
            hostname: Final = (parsed.hostname or "").lower()
            default_port: Final = 443 if scheme == "https" else 80
            port: Final = parsed.port if parsed.port is not None else default_port
            return (scheme, hostname, port)

        if _origin(return_to) != _origin(control_plane_url):
            raise HTTPException(
                status_code=400,
                detail="return_to does not match the configured control_plane_url",
            )

        return True

    @staticmethod
    async def get_sso_login_redirect(
        redirect_url: str,
        google_client_id: str | None = None,
        microsoft_client_id: str | None = None,
        generic_client_id: str | None = None,
        state: str | None = None,
        request: Request | None = None,
    ) -> RedirectResponse | None:
        """
        Step 1. Call Get Login Redirect for the SSO provider. Send the redirect response to `redirect_url`

        Args:
            redirect_url (str): The URL to redirect the user to after login
            google_client_id (Optional[str], optional): The Google Client ID. Defaults to None.
            microsoft_client_id (Optional[str], optional): The Microsoft Client ID. Defaults to None.
            generic_client_id (Optional[str], optional): The Generic Client ID. Defaults to None.
            request: Optional FastAPI request, used to drive the ``Secure``
                attribute on the ``litellm_oauth_state`` CSRF cookie.

        Returns:
            RedirectResponse: The redirect response from the SSO provider.
        """
        # Google SSO Auth
        if google_client_id is not None:
            from fastapi_sso.sso.google import GoogleSSO

            google_client_secret: Final = os.getenv("GOOGLE_CLIENT_SECRET", None)
            if google_client_secret is None:
                raise ProxyException(
                    message="GOOGLE_CLIENT_SECRET not set. Set it in .env file",
                    type=ProxyErrorTypes.auth_error,
                    param="GOOGLE_CLIENT_SECRET",
                    code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                )
            google_sso: Final = GoogleSSO(
                client_id=google_client_id,
                client_secret=google_client_secret,
                redirect_uri=redirect_url,
            )
            verbose_proxy_logger.info(
                "In /google-login/key/generate, \nGOOGLE_REDIRECT_URI: %s\nGOOGLE_CLIENT_ID: %s",
                redirect_url,
                google_client_id,
            )
            with google_sso:
                return await google_sso.get_login_redirect(state=state)
        # Microsoft SSO Auth
        elif microsoft_client_id is not None:
            microsoft_client_secret: Final = os.getenv("MICROSOFT_CLIENT_SECRET", None)
            microsoft_tenant: Final = os.getenv("MICROSOFT_TENANT", None)
            if microsoft_client_secret is None:
                raise ProxyException(
                    message="MICROSOFT_CLIENT_SECRET not set. Set it in .env file",
                    type=ProxyErrorTypes.auth_error,
                    param="MICROSOFT_CLIENT_SECRET",
                    code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                )
            microsoft_sso: Final = CustomMicrosoftSSO(
                client_id=microsoft_client_id,
                client_secret=microsoft_client_secret,
                tenant=microsoft_tenant,
                redirect_uri=redirect_url,
                allow_insecure_http=True,
            )
            with microsoft_sso:
                return await microsoft_sso.get_login_redirect(state=state)
        elif generic_client_id is not None:
            from fastapi_sso.sso.base import DiscoveryDocument
            from fastapi_sso.sso.generic import create_provider

            generic_client_secret: Final = os.getenv("GENERIC_CLIENT_SECRET", None)
            generic_scope: Final = os.getenv("GENERIC_SCOPE", "openid email profile").split(" ")
            generic_authorization_endpoint: Final = os.getenv("GENERIC_AUTHORIZATION_ENDPOINT", None)
            generic_token_endpoint: Final = os.getenv("GENERIC_TOKEN_ENDPOINT", None)
            generic_userinfo_endpoint: Final = os.getenv("GENERIC_USERINFO_ENDPOINT", None)
            if generic_client_secret is None:
                raise ProxyException(
                    message="GENERIC_CLIENT_SECRET not set. Set it in .env file",
                    type=ProxyErrorTypes.auth_error,
                    param="GENERIC_CLIENT_SECRET",
                    code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                )
            if generic_authorization_endpoint is None:
                raise ProxyException(
                    message="GENERIC_AUTHORIZATION_ENDPOINT not set. Set it in .env file",
                    type=ProxyErrorTypes.auth_error,
                    param="GENERIC_AUTHORIZATION_ENDPOINT",
                    code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                )
            if generic_token_endpoint is None:
                raise ProxyException(
                    message="GENERIC_TOKEN_ENDPOINT not set. Set it in .env file",
                    type=ProxyErrorTypes.auth_error,
                    param="GENERIC_TOKEN_ENDPOINT",
                    code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                )
            if generic_userinfo_endpoint is None:
                raise ProxyException(
                    message="GENERIC_USERINFO_ENDPOINT not set. Set it in .env file",
                    type=ProxyErrorTypes.auth_error,
                    param="GENERIC_USERINFO_ENDPOINT",
                    code=status.HTTP_500_INTERNAL_SERVER_ERROR,
                )
            verbose_proxy_logger.debug(
                "authorization_endpoint: %s\ntoken_endpoint: %s\nuserinfo_endpoint: %s",
                generic_authorization_endpoint,
                generic_token_endpoint,
                generic_userinfo_endpoint,
            )
            verbose_proxy_logger.debug(
                "GENERIC_REDIRECT_URI: %s\nGENERIC_CLIENT_ID: %s\n", redirect_url, generic_client_id
            )
            discovery: Final = DiscoveryDocument(
                authorization_endpoint=generic_authorization_endpoint,
                token_endpoint=generic_token_endpoint,
                userinfo_endpoint=generic_userinfo_endpoint,
            )
            SSOProvider: Final = create_provider(name="oidc", discovery_document=discovery)
            generic_sso: Final = SSOProvider(
                client_id=generic_client_id,
                client_secret=generic_client_secret,
                redirect_uri=redirect_url,
                allow_insecure_http=True,
                scope=generic_scope,
            )
            return await SSOAuthenticationHandler.get_generic_sso_redirect_response(
                generic_sso=generic_sso,
                state=state,
                generic_authorization_endpoint=generic_authorization_endpoint,
                request=request,
            )
        raise ValueError(
            "Unknown SSO provider. Please setup SSO with client IDs https://docs.litellm.ai/docs/proxy/admin_ui_sso"
        )

    @staticmethod
    async def get_generic_sso_redirect_response(
        generic_sso: Any,
        state: str | None = None,
        generic_authorization_endpoint: str | None = None,
        request: Request | None = None,
    ) -> RedirectResponse | None:
        """
        Get the redirect response for Generic SSO
        """
        from urllib.parse import parse_qs, urlencode, urlparse, urlunparse

        from litellm.proxy.proxy_server import redis_usage_cache, user_api_key_cache

        with generic_sso:
            # State is bound to the caller's browser via a ``litellm_oauth_state``
            # HttpOnly cookie set on the redirect response below; the SSO
            # callback validates the URL ``state`` against that cookie before
            # completing the PKCE token exchange.  Without this binding, an
            # attacker who pre-mints a state + a cached PKCE verifier can hand
            # the link to a victim and capture the resulting access token
            # (Login CSRF / token theft).
            (
                redirect_params,
                code_verifier,
            ) = SSOAuthenticationHandler._get_generic_sso_redirect_params(
                state=state,
                generic_authorization_endpoint=generic_authorization_endpoint,
            )

            # Separate PKCE params from state params (fastapi-sso doesn't accept code_challenge)
            pkce_params: Final = {}
            state_only_params: Final = {}
            for key, value in redirect_params.items():
                if key in ("code_challenge", "code_challenge_method"):
                    pkce_params[key] = value
                else:
                    state_only_params[key] = value

            # Get the redirect response from fastapi-sso with only state param
            redirect_response: Final = await generic_sso.get_login_redirect(**state_only_params)

            # If PKCE is enabled, add PKCE parameters to the redirect URL
            if code_verifier and "state" in redirect_params:
                # Store code_verifier in cache (10 min TTL). Wrap in dict for proper
                # JSON serialization in Redis. Use Redis when available so callbacks
                # landing on another pod can retrieve it (multi-pod SSO).
                cache_key: Final = f"pkce_verifier:{redirect_params['state']}"
                if redis_usage_cache is not None:
                    await redis_usage_cache.async_set_cache(
                        key=cache_key,
                        value={"code_verifier": code_verifier},
                        ttl=600,
                    )
                else:
                    await user_api_key_cache.async_set_cache(
                        key=cache_key,
                        value={"code_verifier": code_verifier},
                        ttl=600,
                    )
                verbose_proxy_logger.debug("PKCE code_verifier stored in cache (TTL: 600s)")

                # Add PKCE parameters to the authorization URL
                if pkce_params:
                    parsed_url: Final = urlparse(str(redirect_response.headers["location"]))
                    query_params: Final = parse_qs(parsed_url.query)

                    # Add PKCE parameters
                    for key, value in pkce_params.items():
                        query_params[key] = [value]

                    # Reconstruct the URL with PKCE parameters
                    new_query: Final = urlencode(query_params, doseq=True)
                    new_url: Final = urlunparse(
                        (
                            parsed_url.scheme,
                            parsed_url.netloc,
                            parsed_url.path,
                            parsed_url.params,
                            new_query,
                            parsed_url.fragment,
                        )
                    )

                    # Update the redirect response
                    redirect_response.headers["location"] = new_url

                # Bind state to the user's browser session.  The /callback
                # handler validates the URL ``state`` against this cookie via
                # ``secrets.compare_digest`` before exchanging the PKCE
                # code_verifier.  Only set the cookie when PKCE is in use
                # (i.e. inside this ``code_verifier`` branch) so two
                # concurrent SSO sessions — one PKCE, one plain — cannot
                # overwrite each other's state cookie.
                state_value: Final = redirect_params.get("state")
                if state_value and redirect_response is not None:
                    # Production-safe default: require HTTPS for the
                    # CSRF-protection cookie unless we can prove the
                    # incoming request is HTTP (local dev).  Without
                    # ``Secure`` the cookie is sent over plain HTTP,
                    # letting a network observer read and replay the
                    # state value and bypass this protection.
                    secure_flag: Final = request is None or request.url.scheme == "https"
                    redirect_response.set_cookie(
                        key="litellm_oauth_state",
                        value=state_value,
                        max_age=600,
                        httponly=True,
                        samesite="lax",
                        secure=secure_flag,
                    )
            return redirect_response

    @staticmethod
    def _get_generic_sso_redirect_params(
        state: str | None = None,
        generic_authorization_endpoint: str | None = None,
    ) -> tuple[dict[str, str], str | None]:
        """
        Get redirect parameters for Generic SSO with proper state priority handling.
        Optionally generates PKCE parameters if GENERIC_CLIENT_USE_PKCE is enabled.

        Priority order:
        1. CLI state (if provided)
        2. GENERIC_CLIENT_STATE environment variable
        3. Generated UUID (required by Okta and most OAuth providers)


        Args:
            state: Optional state parameter (e.g., CLI state)
            generic_authorization_endpoint: Authorization endpoint URL

        Returns:
            Tuple[dict, Optional[str]]:
                - Redirect parameters for SSO login (may include PKCE params)
                - code_verifier (if PKCE is enabled, None otherwise)
        """
        redirect_params: Final = {}
        code_verifier: str | None = None

        if state:
            # CLI state takes priority
            # the litellm proxy cli sends the "state" parameter to the proxy server for auth. We should maintain the state parameter for the cli if it is provided
            redirect_params["state"] = state
        else:
            generic_client_state: Final = os.getenv("GENERIC_CLIENT_STATE", None)
            if generic_client_state:
                redirect_params["state"] = generic_client_state
            else:
                redirect_params["state"] = uuid.uuid4().hex

        # Handle PKCE (Proof Key for Code Exchange) if enabled
        # Set GENERIC_CLIENT_USE_PKCE=true to enable PKCE for enhanced OAuth security
        use_pkce: Final = os.getenv("GENERIC_CLIENT_USE_PKCE", "false").lower() == "true"

        if use_pkce:
            (
                code_verifier,
                code_challenge,
            ) = SSOAuthenticationHandler.generate_pkce_params()
            redirect_params["code_challenge"] = code_challenge
            redirect_params["code_challenge_method"] = "S256"
            verbose_proxy_logger.debug("PKCE enabled for authorization request")

        return redirect_params, code_verifier

    @staticmethod
    def should_use_sso_handler(
        google_client_id: str | None = None,
        microsoft_client_id: str | None = None,
        generic_client_id: str | None = None,
    ) -> bool:
        if google_client_id is not None or microsoft_client_id is not None or generic_client_id is not None:
            return True
        return False

    @staticmethod
    def get_redirect_url_for_sso(
        request: Request,
        sso_callback_route: str,
        existing_key: str | None = None,
    ) -> str:
        """
        Get the redirect URL for SSO

        Note: existing_key is not added to the URL to avoid changing the callback URL.
        It should be passed via the state parameter instead.
        """
        from litellm.proxy.utils import get_custom_url

        redirect_url = get_custom_url(request_base_url=str(request.base_url))
        if redirect_url.endswith("/"):
            redirect_url += sso_callback_route
        else:
            redirect_url += "/" + sso_callback_route

        return redirect_url

    @staticmethod
    async def upsert_sso_user(
        result: CustomOpenID | OpenID | dict | None,
        user_info: NewUserResponse | LiteLLM_UserTable | None,
        user_email: str | None,
        user_defined_values: SSOUserDefinedValues | None,
        prisma_client: PrismaClient,
    ):
        """
        Connects the SSO Users to the User Table in LiteLLM DB

        - If user on LiteLLM DB, update the user_email and user_role (if SSO provides valid role) with the SSO values
        - If user not on LiteLLM DB, insert the user into LiteLLM DB
        """
        try:
            if user_info is not None:
                user_id: Final = user_info.user_id
                update_data: Final = _build_sso_user_update_data(
                    result=result,
                    user_email=user_email,
                    user_id=user_id,
                )

                await _user_meta_db(UserRepository(prisma_client)).update_many(
                    where={"user_id": user_id}, data=update_data
                )
            else:
                verbose_proxy_logger.info("user not in DB, inserting user into LiteLLM DB")
                # user not in DB, insert User into LiteLLM DB
                user_info = await insert_sso_user(
                    result_openid=result,
                    user_defined_values=user_defined_values,
                )
            return user_info
        except Exception as e:
            verbose_proxy_logger.exception("Error upserting SSO user into LiteLLM DB: %s", e)
            return user_info

    @staticmethod
    async def add_user_to_teams_from_sso_response(
        result: CustomOpenID | OpenID | dict | None,
        user_info: NewUserResponse | LiteLLM_UserTable | None,
    ):
        """
        Adds the user as a team member to the teams specified in the SSO responses `team_ids` field


        The `team_ids` field is populated by litellm after processing the SSO response
        """
        if user_info is None:
            verbose_proxy_logger.debug("User not found in LiteLLM DB, skipping team member addition")
            return
        sso_teams: Final = getattr(result, "team_ids", [])
        await add_missing_team_member(user_info=user_info, sso_teams=sso_teams)

    @staticmethod
    def verify_user_in_restricted_sso_group(
        general_settings: dict,
        result: CustomOpenID | OpenID | dict | None,
        received_response: dict | None,
    ) -> Literal[True]:
        """
        when ui_access_mode.type == "restricted_sso_group":

        - result.team_ids should contain the restricted_sso_group
        - if not, raise a ProxyException
        - if so, return True
        - if result.team_ids is None, return False
        - if result.team_ids is an empty list, return False
        - if result.team_ids is a list, return True if the restricted_sso_group is in the list, otherwise return False
        """

        ui_access_mode: Final = cast(dict | str | None, general_settings.get("ui_access_mode"))

        if ui_access_mode is None:
            return True
        if isinstance(ui_access_mode, str):
            return True
        team_ids: Final = getattr(result, "team_ids", [])

        if ui_access_mode.get("type") == "restricted_sso_group":
            restricted_sso_group: Final = ui_access_mode.get("restricted_sso_group")
            if restricted_sso_group not in team_ids:
                raise ProxyException(
                    message=f"User is not in the restricted SSO group: {restricted_sso_group}. User groups: {team_ids}. Received SSO response: {received_response}",
                    type=ProxyErrorTypes.auth_error,
                    param="restricted_sso_group",
                    code=status.HTTP_403_FORBIDDEN,
                )
        return True

    @staticmethod
    async def create_litellm_team_from_sso_group(
        litellm_team_id: str,
        litellm_team_name: str | None = None,
    ):
        """
        Creates a Litellm Team from a SSO Group ID

        Your SSO provider might have groups that should be created on LiteLLM

        Use this helper to create a Litellm Team from a SSO Group ID

        Args:
            litellm_team_id (str): The ID of the Litellm Team
            litellm_team_name (Optional[str]): The name of the Litellm Team
        """
        from litellm.proxy.proxy_server import prisma_client

        if prisma_client is None:
            raise ProxyException(
                message="Prisma client not found. Set it in the proxy_server.py file",
                type=ProxyErrorTypes.auth_error,
                param="prisma_client",
                code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )
        try:
            team_obj: Final = await _team_detail_db(TeamRepository(prisma_client)).find_first(
                where={"team_id": litellm_team_id}
            )
            verbose_proxy_logger.debug("Team object: %s", team_obj)

            # only create a new team if it doesn't exist
            if team_obj:
                verbose_proxy_logger.debug("Team already exists: %s - %s", litellm_team_id, litellm_team_name)
                return

            team_request: NewTeamRequest = NewTeamRequest(
                team_id=litellm_team_id,
                team_alias=litellm_team_name,
            )
            if litellm.default_team_params:
                team_request = SSOAuthenticationHandler._cast_and_deepcopy_litellm_default_team_params(
                    default_team_params=litellm.default_team_params,
                    litellm_team_id=litellm_team_id,
                    litellm_team_name=litellm_team_name,
                    team_request=team_request,
                )

            await new_team(
                data=team_request,
                # params used for Audit Logging
                http_request=Request(scope={"type": "http", "method": "POST"}),
                user_api_key_dict=UserAPIKeyAuth(
                    token="",
                    key_alias=f"litellm.{MicrosoftSSOHandler.__name__}",
                ),
            )
        except Exception as e:
            verbose_proxy_logger.exception("Error creating Litellm Team: %s", e)

    @staticmethod
    def _cast_and_deepcopy_litellm_default_team_params(
        default_team_params: DefaultTeamSSOParams | dict,
        team_request: NewTeamRequest,
        litellm_team_id: str,
        litellm_team_name: str | None = None,
    ) -> NewTeamRequest:
        """
        Casts and deepcopies the litellm.default_team_params to a NewTeamRequest object

        - Ensures we create a new DefaultTeamSSOParams object
        - Handle the case where litellm.default_team_params is a dict or a DefaultTeamSSOParams object
        - Adds the litellm_team_id and litellm_team_name to the DefaultTeamSSOParams object
        """
        if isinstance(default_team_params, dict):
            _team_request: Final = deepcopy(default_team_params)
            _team_request["team_id"] = litellm_team_id
            _team_request["team_alias"] = litellm_team_name
            team_request = NewTeamRequest(**_team_request)
        elif isinstance(litellm.default_team_params, DefaultTeamSSOParams):
            _default_team_params: Final = deepcopy(litellm.default_team_params)
            _new_team_request: Final = team_request.model_dump()
            _new_team_request.update(_default_team_params)
            team_request = NewTeamRequest.model_validate(_new_team_request)
        return team_request

    @staticmethod
    def _get_cli_state(
        source: str | None,
        key: str | None,
        existing_key: str | None = None,
        user_code: str | None = None,
    ) -> str | None:
        """
        Checks the request 'source' if a cli state token was passed in

        This is used to authenticate through the CLI login flow.

        The state parameter format is: {PREFIX}:{login_id}[:{user_code}]
        - The state parameter is used to pass data through the OAuth flow without changing the callback URL
        - user_code is appended only for the opt-in verification_uri_complete flow so the verify page can pre-fill it
        """
        from litellm.constants import (
            LITELLM_CLI_SESSION_TOKEN_PREFIX,
        )

        if source == LITELLM_CLI_SOURCE_IDENTIFIER and key:
            if _is_valid_cli_sso_user_code(user_code):
                return f"{LITELLM_CLI_SESSION_TOKEN_PREFIX}:{key}:{user_code}"
            return f"{LITELLM_CLI_SESSION_TOKEN_PREFIX}:{key}"
        else:
            return None

    @staticmethod
    def _get_user_email_and_id_from_result(
        result: OpenID | dict | None,
        generic_client_id: str | None = None,
    ) -> ParsedOpenIDResult:
        """
        Gets the user email and id from the OpenID result after validating the email domain
        """
        user_email: str | None = normalize_email(getattr(result, "email", None))
        user_id: str | None = getattr(result, "id", None) if result is not None else None
        user_role: str | None = None

        if user_email is not None and os.getenv("ALLOWED_EMAIL_DOMAINS") is not None:
            email_domain: Final = user_email.split("@")[1]
            allowed_domains: Final = os.getenv("ALLOWED_EMAIL_DOMAINS").split(",")
            if email_domain not in allowed_domains:
                raise HTTPException(
                    status_code=401,
                    detail={
                        "message": f"The email domain={email_domain}, is not an allowed email domain={allowed_domains}. Contact your admin to change this."
                    },
                )

        # Extract user_role from result (works for all SSO providers)
        if result is not None:
            _user_role: Final = getattr(result, "user_role", None)
            if _user_role is not None:
                # Convert enum to string if needed
                user_role = _user_role.value if isinstance(_user_role, LitellmUserRoles) else _user_role
                verbose_proxy_logger.debug("Extracted user_role from SSO result: %s", user_role)

        # generic client id - override with custom attribute name if specified
        if generic_client_id is not None and result is not None:
            generic_user_role_attribute_name: Final = os.getenv("GENERIC_USER_ROLE_ATTRIBUTE", "role")
            user_id = getattr(result, "id", None)
            user_email = normalize_email(getattr(result, "email", None))
            if user_role is None:
                _role_from_attr: Final = getattr(result, generic_user_role_attribute_name, None)
                if _role_from_attr is not None:
                    # Convert enum to string if needed
                    user_role = (
                        _role_from_attr.value if isinstance(_role_from_attr, LitellmUserRoles) else _role_from_attr
                    )

        if user_id is None and result is not None:
            _first_name: Final = getattr(result, "first_name", "") or ""
            _last_name: Final = getattr(result, "last_name", "") or ""
            user_id = _first_name + _last_name

        if user_email is not None and (user_id is None or len(user_id) == 0):
            user_id = user_email

        return ParsedOpenIDResult(
            user_email=user_email,
            user_id=user_id,
            user_role=user_role,
        )

    @staticmethod
    async def get_redirect_response_from_openid(
        result: OpenID | dict | CustomOpenID,
        request: Request,
        received_response: dict | None = None,
        generic_client_id: str | None = None,
        ui_access_mode: dict | None = None,
        access_token_payload: dict | None = None,
        jwt_handler: JWTHandler | None = None,
        return_to: str | None = None,
        sso_assertion: SSOIdentityAssertion | None = None,
    ) -> RedirectResponse:

        from litellm.proxy.proxy_server import (
            general_settings,
            generate_key_helper_fn,
            master_key,
            premium_user,
            proxy_logging_obj,
            redis_usage_cache,
            user_api_key_cache,
            user_custom_sso,
        )
        from litellm.proxy.utils import get_prisma_client_or_throw
        from litellm.types.proxy.ui_sso import ReturnedUITokenObject

        prisma_client: Final = get_prisma_client_or_throw("Prisma client is None, connect a database to your proxy")

        # User is Authe'd in - generate key for the UI to access Proxy
        parsed_openid_result: Final = SSOAuthenticationHandler._get_user_email_and_id_from_result(
            result=result, generic_client_id=generic_client_id
        )
        user_email: Final = parsed_openid_result.get("user_email")
        user_id = parsed_openid_result.get("user_id")
        user_role = parsed_openid_result.get("user_role")
        verbose_proxy_logger.info("SSO callback result: %s", result)

        user_info = None
        user_id_models: Final[list] = []
        max_internal_user_budget: Final = litellm.max_internal_user_budget
        internal_user_budget_duration: Final = litellm.internal_user_budget_duration

        # User might not be already created on first generation of key
        # But if it is, we want their models preferences
        user_defined_values: SSOUserDefinedValues | None = None

        custom_sso_handler: Final[_CustomSsoCall | None] = user_custom_sso
        if custom_sso_handler is not None:
            if inspect.iscoroutinefunction(custom_sso_handler):
                user_defined_values = await custom_sso_handler(result)
            else:
                raise ValueError("user_custom_sso must be a coroutine function")
        elif user_id is not None:
            user_defined_values = SSOUserDefinedValues(
                models=user_id_models,
                user_id=user_id,
                user_email=user_email,
                max_budget=max_internal_user_budget,
                user_role=user_role,
                budget_duration=internal_user_budget_duration,
            )

        # (IF SET) Verify user is in restricted SSO group
        SSOAuthenticationHandler.verify_user_in_restricted_sso_group(
            general_settings=general_settings,
            result=result,
            received_response=received_response,
        )

        user_info = await get_user_info_from_db(
            result=result,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            proxy_logging_obj=proxy_logging_obj,
            user_email=user_email,
            user_defined_values=user_defined_values,
            alternate_user_id=user_id,
        )

        # Sync user role from JWT claims via jwt_litellm_role_map (if configured).
        # This ensures SSO users get the same role mapping as API/JWT users.
        # Use the decoded access_token_payload (not received_response) because
        # custom role claims (e.g. custom_roles) are encoded inside the JWT
        # access token, which is stripped from received_response.
        await _sync_user_role_from_jwt_role_map(
            jwt_handler=jwt_handler,
            received_response=access_token_payload or received_response,
            user_info=user_info,
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            user_defined_values=user_defined_values,
        )

        user_defined_values = apply_user_info_values_to_sso_user_defined_values(
            user_info=user_info, user_defined_values=user_defined_values
        )

        if user_defined_values is None:
            raise Exception(
                "Unable to map user identity to known values. 'user_defined_values' is None. File an issue - https://github.com/BerriAI/litellm/issues"
            )

        verbose_proxy_logger.info("user_defined_values for creating ui key: %s", user_defined_values)

        response: Final = await generate_key_helper_fn(
            request_type="key",
            duration=LITELLM_UI_SESSION_DURATION,
            key_max_budget=litellm.max_ui_session_budget,
            aliases={},
            config={},
            spend=0,
            team_id="litellm-dashboard",
            models=user_defined_values["models"],
            user_id=user_defined_values["user_id"],
            user_email=user_defined_values["user_email"],
            user_role=user_defined_values["user_role"],
            max_budget=user_defined_values["max_budget"],
            budget_duration=user_defined_values["budget_duration"],
            table_name="key",
        )

        key = response["token"]
        user_id = response["user_id"]

        user_role = user_defined_values["user_role"] or LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value
        if user_id and isinstance(user_id, str):
            user_role = await check_and_update_if_proxy_admin_id(
                user_role=user_role, user_id=user_id, prisma_client=prisma_client
            )

        verbose_proxy_logger.debug("user_role: %s; ui_access_mode: %s", user_role, ui_access_mode)
        ## CHECK IF ROLE ALLOWED TO USE PROXY ##
        is_admin_only_access: Final = check_is_admin_only_access(ui_access_mode or {})
        if is_admin_only_access:
            has_access: Final = has_admin_ui_access(user_role or "")
            if not has_access:
                raise HTTPException(
                    status_code=401,
                    detail={
                        "error": f"User not allowed to access proxy. User role={user_role}, proxy mode={ui_access_mode}"
                    },
                )

        if isinstance(user_id, str) and user_id:
            await retain_sso_identity_assertion_for_ema(user_id=user_id, assertion=sso_assertion)

        disabled_non_admin_personal_key_creation: Final = get_disabled_non_admin_personal_key_creation()
        litellm_dashboard_ui = get_custom_url(request_base_url=str(request.base_url), route="ui/")

        if get_secret_bool("EXPERIMENTAL_UI_LOGIN"):
            _user_info: LiteLLM_UserTable | None = None
            if user_defined_values is not None and user_defined_values["user_id"] is not None:
                _user_info = LiteLLM_UserTable(
                    user_id=user_defined_values["user_id"],
                    user_role=user_defined_values["user_role"] or user_role,
                    models=[],
                    max_budget=litellm.max_ui_session_budget,
                )
            if _user_info is None:
                raise HTTPException(
                    status_code=401,
                    detail={"error": "User Information is required for experimental UI login"},
                )

            key = ExperimentalUIJWTToken.get_experimental_ui_login_jwt_auth_token(_user_info)

        returned_ui_token_object: Final = ReturnedUITokenObject(
            user_id=cast(str, user_id),
            key=key,
            user_email=user_email,
            user_role=user_role or LitellmUserRoles.INTERNAL_USER_VIEW_ONLY.value,
            login_method="sso",
            premium_user=premium_user,
            auth_header_name=general_settings.get("litellm_key_header_name", "Authorization"),
            disabled_non_admin_personal_key_creation=disabled_non_admin_personal_key_creation,
            server_root_path=get_server_root_path(),
        )

        from litellm.proxy.auth.login_utils import encode_ui_session_jwt

        jwt_token: Final = encode_ui_session_jwt(returned_ui_token_object, master_key or "")

        # Post-SSO return_to handling (the same-origin DCR round-trip and the control-plane
        # cross-origin code exchange) lives in one shared helper so this method stays inside the
        # complexity budget. None falls through to the dashboard redirect below.
        return_to_redirect: Final = await _sso_return_to_redirect(
            return_to=return_to,
            jwt_token=jwt_token,
            redis_usage_cache=redis_usage_cache,
            user_api_key_cache=user_api_key_cache,
        )
        if return_to_redirect is not None:
            return return_to_redirect

        if user_id is not None and isinstance(user_id, str):
            litellm_dashboard_ui += "?login=success"
        verbose_proxy_logger.info("Redirecting to %s", litellm_dashboard_ui)
        redirect_response: Final = RedirectResponse(url=litellm_dashboard_ui, status_code=303)
        redirect_response.set_cookie(key="token", value=jwt_token)
        return redirect_response

    @staticmethod
    async def prepare_token_exchange_parameters(
        request: Request,
        generic_include_client_id: bool,
    ) -> dict:
        """
        Prepare token exchange parameters for Generic SSO.

        Args:
            request: Request object
            generic_include_client_id: Generic OAuth Client ID

        Returns:
            dict: Token exchange parameters
        """
        # Prepare token exchange parameters (may add code_verifier: str later)
        token_params: Final[dict[str, object]] = {"include_client_id": generic_include_client_id}

        # Retrieve PKCE code_verifier if PKCE was used in authorization.
        # Gate on GENERIC_CLIENT_USE_PKCE to avoid an unnecessary Redis round-trip
        # on every non-PKCE SSO callback.
        query_params: Final = dict(request.query_params)
        state: Final = query_params.get("state")

        use_pkce: Final = os.getenv("GENERIC_CLIENT_USE_PKCE", "false").lower() == "true"

        if use_pkce and not state:
            verbose_proxy_logger.warning(
                "PKCE is enabled (GENERIC_CLIENT_USE_PKCE=true) but no 'state' parameter "
                "was found in the callback. The PKCE verifier cannot be retrieved without "
                "a state value — the token exchange will proceed without code_verifier, "
                "which the provider may reject. Ensure your OAuth provider returns 'state' "
                "in the callback redirect."
            )

        if state and use_pkce:
            from litellm.proxy.proxy_server import redis_usage_cache, user_api_key_cache

            cache_key: Final = f"pkce_verifier:{state}"
            if redis_usage_cache is not None:
                cached_data = await redis_usage_cache.async_get_cache(key=cache_key)
            else:
                cached_data = await user_api_key_cache.async_get_cache(key=cache_key)

            code_verifier = None
            # Track why code_verifier is absent for accurate strict-mode diagnostics.
            _empty_value_in_dict = False  # dict format correct but value is empty/null

            if cached_data:
                # Extract code_verifier from dict (stored as dict for JSON serialization)
                if isinstance(cached_data, dict) and "code_verifier" in cached_data:
                    code_verifier = cached_data["code_verifier"]
                    if not code_verifier:
                        # Dict format is correct but value is empty or null.  This is
                        # a distinct case from an unrecognized format — the entry exists
                        # but was stored with an empty/null verifier (data integrity issue).
                        _empty_value_in_dict = True
                        verbose_proxy_logger.warning(
                            "PKCE verifier dict for state '%s' has an empty/null code_verifier "
                            "value — may indicate a storage bug. Treating as a cache miss.",
                            state,
                        )
                    else:
                        verbose_proxy_logger.debug("PKCE code_verifier retrieved from cache")
                elif isinstance(cached_data, str):
                    # Handle legacy format (plain string) for backward compatibility
                    code_verifier = cached_data
                    verbose_proxy_logger.warning(
                        "Retrieved code_verifier in legacy plain-string format. Future storage will use dict format."
                    )
                else:
                    # Defer the detailed ERROR log to the strict-mode branch below
                    # (which includes state and a diagnostic message).  Log at DEBUG
                    # here to avoid duplicate ERROR entries in the same request.
                    verbose_proxy_logger.debug(
                        "Unexpected PKCE verifier cache format (type=%s); skipping.",
                        type(cached_data).__name__,
                    )

            if code_verifier:
                # Add code_verifier to token exchange parameters.
                token_params["code_verifier"] = code_verifier
                # Return the cache key so the caller can delete it *after* a
                # successful token exchange (avoids losing the verifier on retry
                # if the exchange fails partway through).
                token_params["_pkce_cache_key"] = cache_key
            else:
                await SSOAuthenticationHandler._handle_missing_pkce_verifier(
                    state=state,
                    cache_key=cache_key,
                    cached_data=cached_data,
                    empty_value_in_dict=_empty_value_in_dict,
                    redis_usage_cache=redis_usage_cache,
                    user_api_key_cache=user_api_key_cache,
                )
        return token_params

    @staticmethod
    async def _handle_missing_pkce_verifier(
        state: str | None,
        cache_key: str,
        cached_data: object,
        empty_value_in_dict: bool,
        redis_usage_cache: object,
        user_api_key_cache: object,
    ) -> None:
        """Handle the case where PKCE verifier could not be extracted from cache.

        In strict mode (PKCE_STRICT_CACHE_MISS=true) raises ProxyException.
        Otherwise logs a warning and returns (token exchange proceeds without verifier).
        """
        active_cache: Final = redis_usage_cache if redis_usage_cache is not None else user_api_key_cache
        strict_cache_miss: Final = os.getenv("PKCE_STRICT_CACHE_MISS", "false").lower() == "true"
        if strict_cache_miss:
            if empty_value_in_dict:
                await SSOAuthenticationHandler._delete_pkce_verifier(cache_key)
                raise ProxyException(
                    message=(
                        f"PKCE verifier for state '{state}' was found in cache but "
                        f"has an empty or null code_verifier value — possible storage bug."
                    ),
                    type=ProxyErrorTypes.auth_error,
                    param="PKCE_CACHE_MISS",
                    code=status.HTTP_401_UNAUTHORIZED,
                )
            elif cached_data is not None:
                await SSOAuthenticationHandler._delete_pkce_verifier(cache_key)
                verbose_proxy_logger.error(
                    "PKCE verifier for state '%s' has an unrecognized format (type=%s); "
                    "treating as a cache miss. Investigate the cached value — it may be "
                    "a corrupt or stale entry.",
                    state,
                    type(cached_data).__name__,
                )
                raise ProxyException(
                    message=(
                        f"PKCE verifier for state '{state}' has an unrecognized format "
                        f"(type={type(cached_data).__name__}). The cached entry may be corrupt."
                    ),
                    type=ProxyErrorTypes.auth_error,
                    param="PKCE_CACHE_MISS",
                    code=status.HTTP_401_UNAUTHORIZED,
                )
            else:
                if redis_usage_cache is not None:
                    cause = (
                        "The authorization and callback were likely handled by different "
                        "instances — the verifier was stored on one pod but not found on another."
                    )
                else:
                    cause = (
                        "The verifier may have expired (TTL), been lost on a pod restart, "
                        "or the PKCE authorization step was never completed. "
                        "Configure Redis so all proxy instances share the PKCE verifier."
                    )
                verbose_proxy_logger.error(
                    "PKCE is enabled but no verifier found in cache for state '%s'. %s Cache type: %s.",
                    state,
                    cause,
                    type(active_cache).__name__,
                )
                raise ProxyException(
                    message=f"PKCE verifier not found in cache for state '{state}'. {cause}",
                    type=ProxyErrorTypes.auth_error,
                    param="PKCE_CACHE_MISS",
                    code=status.HTTP_401_UNAUTHORIZED,
                )
        else:
            if cached_data is not None:
                await SSOAuthenticationHandler._delete_pkce_verifier(cache_key)
            verbose_proxy_logger.warning(
                "PKCE is enabled but verifier not found in cache for state '%s' "
                "(cache type: %s, raw data present: %s). "
                "Continuing without code_verifier — set PKCE_STRICT_CACHE_MISS=true to fail fast instead.",
                state,
                type(active_cache).__name__,
                cached_data is not None,
            )

    @staticmethod
    async def _delete_pkce_verifier(cache_key: str) -> None:
        """Delete a single-use PKCE verifier from cache after a successful exchange.

        Failure is non-fatal: a leftover verifier is a minor security concern
        (unused key in cache) but not worth aborting an otherwise-successful login.
        """
        from litellm.proxy.proxy_server import redis_usage_cache, user_api_key_cache

        try:
            if redis_usage_cache is not None:
                await redis_usage_cache.async_delete_cache(key=cache_key)
            else:
                await user_api_key_cache.async_delete_cache(key=cache_key)
        except Exception as exc:
            verbose_proxy_logger.warning(
                "PKCE: failed to delete verifier cache key '%s' (best-effort cleanup): %s",
                cache_key,
                exc,
            )

    @staticmethod
    def generate_pkce_params() -> tuple[str, str]:
        """
        Generate PKCE (Proof Key for Code Exchange) parameters for OAuth 2.0.

        Returns:
            Tuple[str, str]: (code_verifier, code_challenge)
            - code_verifier: Random 43-128 character string (we use 43 for efficiency)
            - code_challenge: Base64-URL-encoded SHA256 hash of the code_verifier

        Reference: https://datatracker.ietf.org/doc/html/rfc7636
        """
        # Generate a cryptographically random code_verifier (43 characters)
        # Using 32 random bytes which becomes 43 characters when base64-url-encoded
        code_verifier: Final = base64.urlsafe_b64encode(secrets.token_bytes(32)).decode("utf-8").rstrip("=")

        # Generate code_challenge using S256 method (SHA256)
        code_challenge_bytes: Final = hashlib.sha256(code_verifier.encode("utf-8")).digest()
        code_challenge: Final = base64.urlsafe_b64encode(code_challenge_bytes).decode("utf-8").rstrip("=")

        return code_verifier, code_challenge

    @staticmethod
    def _validate_token_response(response: "httpx.Response") -> dict:
        """
        Parse and validate the token endpoint response.

        Ensures the response is valid JSON, a dict, and contains a non-null
        access_token string. Raises ProxyException on any validation failure.
        """
        try:
            token_response_raw: Final[object] = _as_object(response.json())
        except Exception as json_err:
            verbose_proxy_logger.error(
                "Failed to parse token response as JSON: %s. Body: %s",
                json_err,
                response.text[:500],
            )
            raise ProxyException(
                message=f"Token endpoint returned invalid JSON: {json_err}",
                type=ProxyErrorTypes.auth_error,
                param="token_exchange",
                code=status.HTTP_401_UNAUTHORIZED,
            )

        if not isinstance(token_response_raw, dict):
            verbose_proxy_logger.error(
                "Token endpoint returned non-dict JSON (type=%s). Body: %s",
                type(token_response_raw).__name__,
                response.text[:500],
            )
            raise ProxyException(
                message=(
                    f"Token endpoint returned unexpected response format "
                    f"(expected JSON object, got {type(token_response_raw).__name__})"
                ),
                type=ProxyErrorTypes.auth_error,
                param="token_exchange",
                code=status.HTTP_401_UNAUTHORIZED,
            )
        token_response: Final[dict] = token_response_raw

        access_token_val: Final = token_response.get("access_token")
        if not isinstance(access_token_val, str) or not access_token_val:
            error: Final = token_response.get("error")
            error_desc: Final = token_response.get("error_description", "")
            if error:
                detail = f"{error} - {error_desc}" if error_desc else error
            else:
                detail = (
                    "token endpoint returned HTTP 200 but no access_token "
                    f"(response keys: {sorted(token_response.keys())})"
                )
            verbose_proxy_logger.error("Token response missing or null access_token. detail=%s", detail)
            raise ProxyException(
                message=f"Token exchange failed: {detail}",
                type=ProxyErrorTypes.auth_error,
                param="token_exchange",
                code=status.HTTP_401_UNAUTHORIZED,
            )

        return token_response

    @staticmethod
    async def _pkce_token_exchange(
        authorization_code: str,
        code_verifier: str,
        client_id: str,
        client_secret: str | None,
        token_endpoint: str,
        userinfo_endpoint: str | None,
        include_client_id: bool,
        redirect_url: str | None,
        additional_headers: dict[str, str],
    ) -> dict:
        """
        Performs a direct OAuth token exchange including the PKCE code_verifier.

        fastapi-sso does not forward code_verifier, so when PKCE is enabled we
        bypass it and call the token endpoint ourselves, then fetch user info.

        Returns a combined dict of the token response and user info, suitable
        for passing to a response_convertor.
        """
        verbose_proxy_logger.debug(
            "PKCE: performing direct token exchange (code_verifier length=%d)",
            len(code_verifier),
        )

        token_data: Final[dict[str, str]] = {
            "grant_type": "authorization_code",
            "code": authorization_code,
            "code_verifier": code_verifier,
        }
        # Only include redirect_uri when set — omitting it avoids sending the
        # literal string "None" to the provider if the env var is missing.
        if redirect_url:
            token_data["redirect_uri"] = redirect_url

        request_headers: Final = {
            **additional_headers,
            "Content-Type": "application/x-www-form-urlencoded",  # must not be overridden
            "Accept": "application/json",
        }

        if not include_client_id:
            # Use Basic Auth only when a secret is available; public PKCE clients omit it.
            if client_secret:
                credentials: Final = base64.b64encode(f"{client_id}:{client_secret}".encode()).decode()
                request_headers["Authorization"] = f"Basic {credentials}"
            else:
                token_data["client_id"] = client_id
        else:
            token_data["client_id"] = client_id
            if client_secret:
                token_data["client_secret"] = client_secret

        http_client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.SSO_HANDLER)
        try:
            response: Final = await http_client.post(
                url=token_endpoint,
                data=token_data,
                headers=request_headers,
                timeout=30.0,
            )
        except Exception as exc:
            # Catch network-level errors (SSL, DNS, TCP, timeout, etc.) and
            # wrap them as a clean ProxyException rather than leaking raw
            # httpx or OS exceptions to callers.
            verbose_proxy_logger.error("PKCE token endpoint unreachable: %s", exc)
            raise ProxyException(
                message=f"Token endpoint request failed: {exc}",
                type=ProxyErrorTypes.auth_error,
                param="token_exchange",
                code=status.HTTP_401_UNAUTHORIZED,
            ) from exc
        if response.status_code != 200:
            verbose_proxy_logger.error(
                "PKCE token exchange failed. status=%s body=%s",
                response.status_code,
                response.text[:500],
            )
            raise ProxyException(
                message=f"Token exchange failed: {response.status_code} - {response.text[:500]}",
                type=ProxyErrorTypes.auth_error,
                param="token_exchange",
                code=status.HTTP_401_UNAUTHORIZED,
            )

        token_response: Final = SSOAuthenticationHandler._validate_token_response(response)

        verbose_proxy_logger.debug(
            "PKCE token exchange successful. id_token_present=%s",
            bool(token_response.get("id_token")),
        )
        # Bearer credentials (access_token, id_token, refresh_token) are always sourced
        # from token_response — not from userinfo — in the merge step below.
        userinfo: Final = await SSOAuthenticationHandler._get_pkce_userinfo(
            access_token=token_response["access_token"],
            id_token=token_response.get("id_token"),
            userinfo_endpoint=userinfo_endpoint,
            additional_headers=additional_headers,
        )

        # Merge: userinfo takes precedence for identity claims (sub, email, name, …) per
        # the OpenID Connect spec (userinfo is the authoritative source for identity).
        # Bearer credentials (access_token, id_token, refresh_token) from the token endpoint
        # take precedence over same-named fields in userinfo — non-standard providers sometimes
        # include token fields in userinfo, which must not shadow the real bearer token.
        # If a bearer field is absent from the token response, any userinfo-provided value
        # is preserved as a fallback (useful for non-standard providers that omit id_token
        # from the token response but include it in userinfo).
        #
        # Three-way merge semantics for each bearer-credential field:
        #   1. token_response has a non-null value → use it (token endpoint is authoritative)
        #   2. token_response explicitly sent null  → remove the key so callers get a clean
        #      absence signal; the null from the token endpoint overrides userinfo too
        #   3. field absent from token_response     → leave whatever userinfo provided as-is
        #      (e.g. userinfo-provided id_token from a non-standard provider)
        merged: Final = {**token_response, **userinfo}
        for field in _OAUTH_TOKEN_FIELDS:
            if token_response.get(field) is not None:
                # Case 1: non-null in token_response — restore authoritative value.
                merged[field] = token_response[field]
            elif field in token_response:
                # Case 2: key exists but value is explicitly null — remove from merged.
                merged.pop(field, None)
            # Case 3: field absent from token_response — leave userinfo value as-is.
        return merged

    @staticmethod
    async def _get_pkce_userinfo(
        access_token: str,
        id_token: str | None,
        userinfo_endpoint: str | None,
        additional_headers: dict[str, str],
    ) -> dict:
        """
        Fetches user info from the userinfo endpoint.
        Falls back to decoding the id_token if the endpoint is unavailable.
        """
        # None = request not yet attempted, failed, or returned empty/null (treated as failure
        # so the id_token fallback can be attempted instead of returning a session with no claims).
        userinfo: dict | None = None

        if userinfo_endpoint:
            try:
                client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.SSO_HANDLER)
                resp: Final = await client.get(
                    url=userinfo_endpoint,
                    headers={
                        **additional_headers,
                        "Authorization": f"Bearer {access_token}",  # must not be overridden
                    },
                )
                if resp.status_code == 200:
                    try:
                        userinfo_raw: Final[dict[str, object] | None] = resp.json()
                        if not userinfo_raw:
                            # JSON null (None) or empty dict ({}) — no identity claims.
                            # Treat as failure so id_token fallback can be attempted.
                            verbose_proxy_logger.warning(
                                "Userinfo endpoint returned an empty or null response "
                                "(type=%s); treating as failure and attempting id_token fallback. "
                                "Check your provider's userinfo endpoint configuration.",
                                type(userinfo_raw).__name__,
                            )
                            userinfo = None
                        else:
                            userinfo = userinfo_raw
                    except Exception as json_err:
                        verbose_proxy_logger.warning(
                            "Userinfo endpoint returned non-JSON response (status 200): %s",
                            json_err,
                        )
                else:
                    verbose_proxy_logger.warning(
                        "Userinfo endpoint returned %s (body: %s), falling back to id_token",
                        resp.status_code,
                        resp.text[:500],
                    )
            except Exception as e:
                verbose_proxy_logger.warning("Userinfo endpoint error: %s, falling back to id_token", e)

        # Only fall back to id_token when the userinfo request failed (None).
        # Empty dict ({}) and JSON null are both treated as failure (set to None above) since
        # they contain no identity claims — id_token fallback is attempted in that case too.
        # Explicitly check for a non-empty string to avoid attempting JWT decode on
        # a blank or non-string id_token field from a misbehaving provider.
        if userinfo is None and isinstance(id_token, str) and id_token:
            try:
                userinfo = jwt.decode(id_token, options={"verify_signature": False})
                if not userinfo:
                    # jwt.decode returned an empty dict (payload-free JWT or provider bug).
                    # Treat this the same as a missing userinfo — the session would have no
                    # identity claims, which is equivalent to a broken session.
                    verbose_proxy_logger.warning("id_token decoded to an empty payload — treating as failure.")
                    userinfo = None
            except Exception as decode_err:
                verbose_proxy_logger.error("Failed to decode id_token: %s", decode_err)
                raise ProxyException(
                    message=f"Failed to decode id_token JWT: {decode_err}",
                    type=ProxyErrorTypes.auth_error,
                    param="userinfo",
                    code=status.HTTP_401_UNAUTHORIZED,
                )

        if userinfo is None:
            id_token_attempted: Final = isinstance(id_token, str) and bool(id_token)
            if userinfo_endpoint:
                if id_token_attempted:
                    detail = (
                        "userinfo endpoint failed and id_token was present but "
                        "decoded to an empty payload — no identity claims available"
                    )
                else:
                    detail = "userinfo endpoint failed and no id_token was present in the token response"
            else:
                if id_token_attempted:
                    detail = (
                        "no userinfo endpoint is configured (GENERIC_USERINFO_ENDPOINT) "
                        "and id_token decoded to an empty payload — no identity claims available"
                    )
                else:
                    detail = (
                        "no userinfo endpoint is configured (GENERIC_USERINFO_ENDPOINT) and no id_token was present"
                    )
            raise ProxyException(
                message=f"SSO user info unavailable: {detail}.",
                type=ProxyErrorTypes.auth_error,
                param="userinfo",
                code=status.HTTP_401_UNAUTHORIZED,
            )

        return userinfo


class MicrosoftSSOHandler:
    """
    Handles Microsoft SSO callback response and returns a CustomOpenID object
    """

    DEFAULT_GRAPH_API_BASE_URL = "https://graph.microsoft.com/v1.0"

    """
    Constants
    """
    MAX_GRAPH_API_PAGES = 200

    # used for debugging to show the user groups litellm found from Graph API
    GRAPH_API_RESPONSE_KEY = "graph_api_user_groups"

    @staticmethod
    def get_graph_api_base_url() -> str:
        """
        Returns the Microsoft Graph API base URL, configurable via the
        `MICROSOFT_GRAPH_ENDPOINT` env var so non-default clouds such as Azure
        Government (GCC High) can point at `https://graph.microsoft.us/v1.0`
        """
        return get_secret_str("MICROSOFT_GRAPH_ENDPOINT") or MicrosoftSSOHandler.DEFAULT_GRAPH_API_BASE_URL

    @staticmethod
    def get_graph_api_user_groups_endpoint() -> str:
        return f"{MicrosoftSSOHandler.get_graph_api_base_url()}/me/memberOf"

    @staticmethod
    async def get_microsoft_callback_response(
        request: Request,
        microsoft_client_id: str,
        redirect_url: str,
        return_raw_sso_response: bool = False,
    ) -> CustomOpenID | OpenID | dict:
        """
        Get the Microsoft SSO callback response

        Args:
            return_raw_sso_response: If True, return the raw SSO response
        """
        microsoft_client_secret: Final = os.getenv("MICROSOFT_CLIENT_SECRET", None)
        microsoft_tenant: Final = os.getenv("MICROSOFT_TENANT", None)
        if microsoft_client_secret is None:
            raise ProxyException(
                message="MICROSOFT_CLIENT_SECRET not set. Set it in .env file",
                type=ProxyErrorTypes.auth_error,
                param="MICROSOFT_CLIENT_SECRET",
                code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )
        if microsoft_tenant is None:
            raise ProxyException(
                message="MICROSOFT_TENANT not set. Set it in .env file",
                type=ProxyErrorTypes.auth_error,
                param="MICROSOFT_TENANT",
                code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )
        microsoft_sso: Final = CustomMicrosoftSSO(
            client_id=microsoft_client_id,
            client_secret=microsoft_client_secret,
            tenant=microsoft_tenant,
            redirect_uri=redirect_url,
            allow_insecure_http=True,
        )
        original_msft_result: Final = (
            await microsoft_sso.verify_and_process(
                request=request,
                convert_response=False,
            )
            or {}
        )

        user_team_ids: Final = await MicrosoftSSOHandler.get_user_groups_from_graph_api(
            access_token=microsoft_sso.access_token
        )

        # Extract app roles from the id_token JWT
        app_roles: Final = MicrosoftSSOHandler.get_app_roles_from_id_token(id_token=microsoft_sso.id_token)
        verbose_proxy_logger.debug("Extracted app roles from id_token: %s", app_roles)

        # Combine groups and app roles
        user_role: Final = MicrosoftSSOHandler.get_user_role_from_app_roles(app_roles)

        verbose_proxy_logger.debug("Combined team_ids (groups + app roles): %s", user_team_ids)

        # if user is trying to get the raw sso response for debugging, return the raw sso response
        if return_raw_sso_response:
            original_msft_result[MicrosoftSSOHandler.GRAPH_API_RESPONSE_KEY] = user_team_ids
            original_msft_result["app_roles"] = app_roles
            return original_msft_result or {}

        result: Final = MicrosoftSSOHandler.openid_from_response(
            response=original_msft_result,
            team_ids=user_team_ids,
            user_role=user_role,
        )
        return result

    @staticmethod
    def openid_from_response(
        response: dict | None,
        team_ids: list[str],
        user_role: LitellmUserRoles | None,
    ) -> CustomOpenID:
        response = response or {}
        verbose_proxy_logger.debug("Microsoft SSO Callback Response: %s", response)
        openid_response: Final = CustomOpenID(
            email=normalize_email(response.get(MICROSOFT_USER_EMAIL_ATTRIBUTE) or response.get("mail")),
            display_name=response.get(MICROSOFT_USER_DISPLAY_NAME_ATTRIBUTE),
            provider="microsoft",
            id=response.get(MICROSOFT_USER_ID_ATTRIBUTE),
            first_name=response.get(MICROSOFT_USER_FIRST_NAME_ATTRIBUTE),
            last_name=response.get(MICROSOFT_USER_LAST_NAME_ATTRIBUTE),
            team_ids=team_ids,
            user_role=user_role,
        )
        verbose_proxy_logger.debug("Microsoft SSO OpenID Response: %s", openid_response)
        return openid_response

    @staticmethod
    def get_user_role_from_app_roles(
        app_roles: Sequence[str] | None,
    ) -> LitellmUserRoles | None:
        """
        Resolve the one role LiteLLM stores for a user from their Entra app roles.

        Entra does not guarantee `roles` claim ordering, so a user holding several app
        roles resolves to the highest privilege one rather than whichever the claim
        listed first. Roles the hierarchy does not rank (org_admin, team, customer)
        resolve by name to stay deterministic
        """
        resolved: Final = frozenset(
            role for role in (get_litellm_user_role(role_str) for role_str in app_roles or ()) if role is not None
        )
        if not resolved:
            return None

        ranked: Final = next((role for role in LITELLM_USER_ROLE_HIERARCHY if role in resolved), None)
        return ranked if ranked is not None else min(resolved, key=lambda role: role.value)

    @staticmethod
    def get_app_roles_from_id_token(id_token: str | None) -> list[str]:
        """
        Extract app roles from the Microsoft Entra ID (Azure AD) id_token JWT.

        App roles are assigned in the Azure AD Enterprise Application and appear
        in the 'app_roles' claim of the id_token.

        Args:
            id_token (Optional[str]): The JWT id_token from Microsoft SSO

        Returns:
            List[str]: List of app role names assigned to the user
        """
        if not id_token:
            verbose_proxy_logger.debug("No id_token provided for app role extraction")
            return []

        try:
            import jwt

            # Decode the JWT without signature verification
            # (signature is already verified by fastapi_sso)
            decoded_token: Final = jwt.decode(id_token, options={"verify_signature": False})

            # Extract app_roles claim from the token
            ## check for both 'roles' and 'app_roles' claims
            roles: Final = decoded_token.get("app_roles", []) or decoded_token.get("roles", [])

            if roles and isinstance(roles, list):
                verbose_proxy_logger.debug("Found %s app role(s) in id_token: %s", len(roles), roles)
                return roles
            else:
                verbose_proxy_logger.debug("No app roles found in id_token or roles claim is not a list")
                return []

        except Exception as e:
            verbose_proxy_logger.error("Error extracting app roles from id_token: %s", e)
            return []

    @staticmethod
    async def get_user_groups_from_graph_api(
        access_token: str | None = None,
    ) -> list[str]:
        """
        Returns a list of `team_ids` the user belongs to from the Microsoft Graph API

        Args:
            access_token (Optional[str]): Microsoft Graph API access token

        Returns:
            List[str]: List of group IDs the user belongs to
        """
        try:
            async_client: Final = get_async_httpx_client(llm_provider=httpxSpecialProvider.SSO_HANDLER)

            # Handle MSFT Enterprise Application Groups
            service_principal_id: Final = os.getenv("MICROSOFT_SERVICE_PRINCIPAL_ID", None)
            service_principal_group_ids: list[str] | None = []
            service_principal_teams: list[MicrosoftServicePrincipalTeam] | None = []
            if service_principal_id:
                (
                    service_principal_group_ids,
                    service_principal_teams,
                ) = await MicrosoftSSOHandler.get_group_ids_from_service_principal(
                    service_principal_id=service_principal_id,
                    async_client=async_client,
                    access_token=access_token,
                )
                verbose_proxy_logger.debug("Service principal group IDs: %s", service_principal_group_ids)
                if len(service_principal_group_ids) > 0:
                    await MicrosoftSSOHandler.create_litellm_teams_from_service_principal_team_ids(
                        service_principal_teams=service_principal_teams,
                    )

            # Fetch user membership from Microsoft Graph API
            all_group_ids = []
            next_link: str | None = MicrosoftSSOHandler.get_graph_api_user_groups_endpoint()
            auth_headers: Final = {"Authorization": f"Bearer {access_token}"}
            page_count = 0

            while next_link is not None and page_count < MicrosoftSSOHandler.MAX_GRAPH_API_PAGES:
                group_ids, next_link = await MicrosoftSSOHandler.fetch_and_parse_groups(
                    url=next_link, headers=auth_headers, async_client=async_client
                )
                all_group_ids.extend(group_ids)
                page_count += 1

            if next_link is not None and page_count >= MicrosoftSSOHandler.MAX_GRAPH_API_PAGES:
                verbose_proxy_logger.warning(
                    "Reached maximum page limit of %s. Some groups may not be included.",
                    MicrosoftSSOHandler.MAX_GRAPH_API_PAGES,
                )

            # If service_principal_group_ids is not empty, only return group_ids that are in both all_group_ids and service_principal_group_ids
            if service_principal_group_ids and len(service_principal_group_ids) > 0:
                all_group_ids = [group_id for group_id in all_group_ids if group_id in service_principal_group_ids]

            return all_group_ids

        except Exception as e:
            verbose_proxy_logger.error("Error getting user groups from Microsoft Graph API: %s", e)
            return []

    @staticmethod
    async def fetch_and_parse_groups(
        url: str, headers: dict, async_client: AsyncHTTPHandler
    ) -> tuple[list[str], str | None]:
        """Helper function to fetch and parse group data from a URL"""
        response: Final = await async_client.get(url, headers=headers)
        response_json: Final[dict[str, object]] = response.json()
        response_typed: Final = await MicrosoftSSOHandler._cast_graph_api_response_dict(response=response_json)
        group_ids: Final = MicrosoftSSOHandler._get_group_ids_from_graph_api_response(response=response_typed)
        return group_ids, response_typed.get("odata_nextLink")

    @staticmethod
    def _get_group_ids_from_graph_api_response(
        response: MicrosoftGraphAPIUserGroupResponse,
    ) -> list[str]:
        group_ids: Final = []
        for _object in response.get("value", []) or []:
            _group_id = _object.get("id")
            if _group_id is not None:
                group_ids.append(_group_id)
        return group_ids

    @staticmethod
    async def _cast_graph_api_response_dict(
        response: dict,
    ) -> MicrosoftGraphAPIUserGroupResponse:
        directory_objects: Final[list[MicrosoftGraphAPIUserGroupDirectoryObject]] = []
        for _object in response.get("value", []):
            directory_objects.append(
                MicrosoftGraphAPIUserGroupDirectoryObject(
                    odata_type=_object.get("@odata.type"),
                    id=_object.get("id"),
                    deletedDateTime=_object.get("deletedDateTime"),
                    description=_object.get("description"),
                    displayName=_object.get("displayName"),
                    roleTemplateId=_object.get("roleTemplateId"),
                )
            )
        return MicrosoftGraphAPIUserGroupResponse(
            odata_context=response.get("@odata.context"),
            odata_nextLink=response.get("@odata.nextLink"),
            value=directory_objects,
        )

    @staticmethod
    async def get_group_ids_from_service_principal(
        service_principal_id: str,
        async_client: AsyncHTTPHandler,
        access_token: str | None = None,
    ) -> tuple[list[str], list[MicrosoftServicePrincipalTeam]]:
        """
        Gets the groups belonging to the Service Principal Application

        Service Principal Id is an `Enterprise Application` in Azure AD

        Users use Enterprise Applications to manage Groups and Users on Microsoft Entra ID
        """
        base_url: Final = MicrosoftSSOHandler.get_graph_api_base_url()
        # Endpoint to get app role assignments for the given service principal
        endpoint: Final = f"/servicePrincipals/{service_principal_id}/appRoleAssignedTo"
        next_link: str | None = base_url + endpoint

        headers: Final = {
            "Authorization": f"Bearer {access_token}",
            "Content-Type": "application/json",
        }

        group_ids: Final[list[str]] = []
        service_principal_teams: Final[list[MicrosoftServicePrincipalTeam]] = []
        page_count = 0

        while next_link is not None and page_count < MicrosoftSSOHandler.MAX_GRAPH_API_PAGES:
            response = await async_client.get(next_link, headers=headers)
            response_json: _ServicePrincipalPage = response.json()
            verbose_proxy_logger.debug("Response from service principal app role assigned to: %s", response_json)

            for _object in response_json.get("value", []):
                if _object.get("principalType") == "Group":
                    # Append the group ID to the list
                    group_ids.append(_object.get("principalId"))
                    # Append the service principal team to the list
                    service_principal_teams.append(
                        MicrosoftServicePrincipalTeam(
                            principalDisplayName=_object.get("principalDisplayName"),
                            principalId=_object.get("principalId"),
                        )
                    )

            next_link = response_json.get("@odata.nextLink")
            page_count += 1

        if next_link is not None and page_count >= MicrosoftSSOHandler.MAX_GRAPH_API_PAGES:
            verbose_proxy_logger.warning(
                "Reached maximum page limit of %s. Some service principal group assignments may not be included.",
                MicrosoftSSOHandler.MAX_GRAPH_API_PAGES,
            )

        return group_ids, service_principal_teams

    @staticmethod
    async def create_litellm_teams_from_service_principal_team_ids(
        service_principal_teams: list[MicrosoftServicePrincipalTeam],
    ):
        """
        Creates Litellm Teams from the Service Principal Group IDs

        When a user sets a `SERVICE_PRINCIPAL_ID` in the env, litellm will fetch groups under that service principal and create Litellm Teams from them
        """
        verbose_proxy_logger.debug("Creating Litellm Teams from Service Principal Teams: %s", service_principal_teams)
        for service_principal_team in service_principal_teams:
            litellm_team_id: str | None = service_principal_team.get("principalId")
            litellm_team_name: str | None = service_principal_team.get("principalDisplayName")
            if not litellm_team_id:
                verbose_proxy_logger.debug(
                    "Skipping team creation for %s because it has no principalId", litellm_team_name
                )
                continue

            await SSOAuthenticationHandler.create_litellm_team_from_sso_group(
                litellm_team_id=litellm_team_id,
                litellm_team_name=litellm_team_name,
            )


class GoogleSSOHandler:
    """
    Handles Google SSO callback response and returns a CustomOpenID object
    """

    @staticmethod
    async def get_google_callback_response(
        request: Request,
        google_client_id: str,
        redirect_url: str,
        return_raw_sso_response: bool = False,
    ) -> OpenID | dict:
        """
        Get the Google SSO callback response

        Args:
            return_raw_sso_response: If True, return the raw SSO response
        """
        from fastapi_sso.sso.google import GoogleSSO

        google_client_secret: Final = os.getenv("GOOGLE_CLIENT_SECRET", None)
        if google_client_secret is None:
            raise ProxyException(
                message="GOOGLE_CLIENT_SECRET not set. Set it in .env file",
                type=ProxyErrorTypes.auth_error,
                param="GOOGLE_CLIENT_SECRET",
                code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            )
        google_sso: Final = GoogleSSO(
            client_id=google_client_id,
            redirect_uri=redirect_url,
            client_secret=google_client_secret,
        )

        # if user is trying to get the raw sso response for debugging, return the raw sso response
        if return_raw_sso_response:
            return (
                await google_sso.verify_and_process(
                    request=request,
                    convert_response=False,
                )
                or {}
            )

        result: Final = await google_sso.verify_and_process(request)
        return result or {}


@router.get("/sso/debug/login", tags=["experimental"], include_in_schema=False)
async def debug_sso_login(request: Request):
    """
    Create Proxy API Keys using Google Workspace SSO. Requires setting PROXY_BASE_URL in .env
    PROXY_BASE_URL should be the your deployed proxy endpoint, e.g. PROXY_BASE_URL="https://litellm-production-7002.up.railway.app/"
    Example:
    """
    from litellm.proxy.proxy_server import premium_user

    microsoft_client_id: Final = os.getenv("MICROSOFT_CLIENT_ID", None)
    google_client_id: Final = os.getenv("GOOGLE_CLIENT_ID", None)
    generic_client_id: Final = os.getenv("GENERIC_CLIENT_ID", None)

    ####### Check if user is a Enterprise / Premium User #######
    if microsoft_client_id is not None or google_client_id is not None or generic_client_id is not None:
        if premium_user is not True:
            raise ProxyException(
                message="You must be a LiteLLM Enterprise user to use SSO. If you have a license please set `LITELLM_LICENSE` in your env. If you want to obtain a license meet with us here: https://enterprise.litellm.ai/demo You are seeing this error message because You set one of `MICROSOFT_CLIENT_ID`, `GOOGLE_CLIENT_ID`, or `GENERIC_CLIENT_ID` in your env. Please unset this",
                type=ProxyErrorTypes.auth_error,
                param="premium_user",
                code=status.HTTP_403_FORBIDDEN,
            )

    # get url from request
    redirect_url: Final = SSOAuthenticationHandler.get_redirect_url_for_sso(
        request=request,
        sso_callback_route="sso/debug/callback",
    )

    # Check if we should use SSO handler
    if (
        SSOAuthenticationHandler.should_use_sso_handler(
            microsoft_client_id=microsoft_client_id,
            google_client_id=google_client_id,
            generic_client_id=generic_client_id,
        )
        is True
    ):
        return await SSOAuthenticationHandler.get_sso_login_redirect(
            redirect_url=redirect_url,
            microsoft_client_id=microsoft_client_id,
            google_client_id=google_client_id,
            generic_client_id=generic_client_id,
            request=request,
        )


@router.get("/sso/debug/callback", tags=["experimental"], include_in_schema=False)
async def debug_sso_callback(request: Request):
    """
    Returns the OpenID object returned by the SSO provider
    """
    import json

    from fastapi.responses import HTMLResponse

    from litellm.proxy._types import LiteLLM_JWTAuth
    from litellm.proxy.auth.handle_jwt import JWTHandler
    from litellm.proxy.proxy_server import (
        general_settings,
        jwt_handler,
        prisma_client,
        user_api_key_cache,
    )

    sso_jwt_handler: JWTHandler | None = None
    ui_access_mode: Final = general_settings.get("ui_access_mode", None)
    if ui_access_mode is not None and isinstance(ui_access_mode, dict):
        sso_jwt_handler = JWTHandler()
        sso_jwt_handler.update_environment(
            prisma_client=prisma_client,
            user_api_key_cache=user_api_key_cache,
            litellm_jwtauth=LiteLLM_JWTAuth(
                team_ids_jwt_field=general_settings.get("ui_access_mode", {}).get("sso_group_jwt_field", None),
            ),
            leeway=0,
        )

    microsoft_client_id: Final = os.getenv("MICROSOFT_CLIENT_ID", None)
    google_client_id: Final = os.getenv("GOOGLE_CLIENT_ID", None)
    generic_client_id: Final = os.getenv("GENERIC_CLIENT_ID", None)

    redirect_url = os.getenv("PROXY_BASE_URL", str(request.base_url))
    if redirect_url.endswith("/"):
        redirect_url += "sso/debug/callback"
    else:
        redirect_url += "/sso/debug/callback"

    result = None
    received_response: dict | None = None
    access_token_payload: dict | None = None
    if google_client_id is not None:
        result = await GoogleSSOHandler.get_google_callback_response(
            request=request,
            google_client_id=google_client_id,
            redirect_url=redirect_url,
            return_raw_sso_response=True,
        )
    elif microsoft_client_id is not None:
        result = await MicrosoftSSOHandler.get_microsoft_callback_response(
            request=request,
            microsoft_client_id=microsoft_client_id,
            redirect_url=redirect_url,
            return_raw_sso_response=True,
        )

    elif generic_client_id is not None:
        (
            result,
            received_response,
            access_token_payload,
            _sso_assertion,
        ) = await get_generic_sso_response(
            request=request,
            jwt_handler=jwt_handler,
            generic_client_id=generic_client_id,
            redirect_url=redirect_url,
            sso_jwt_handler=sso_jwt_handler,
        )

    # If result is None, return a basic error message
    if result is None:
        return HTMLResponse(
            content="<h1>SSO Authentication Failed</h1><p>No data was returned from the SSO provider.</p>",
            status_code=400,
        )

    # Convert the OpenID object to a dictionary
    if hasattr(result, "__dict__"):
        result_dict = result.__dict__
    else:
        result_dict = dict(result)

    # Filter out any None values and convert to JSON serializable format
    filtered_result: Final = {}
    for key, value in result_dict.items():
        if value is not None and not key.startswith("_"):
            if isinstance(value, (str, int, float, bool)) or value is None:
                filtered_result[key] = value
            else:
                try:
                    # Try to convert to string or another JSON serializable format
                    filtered_result[key] = str(value)
                except Exception as e:
                    filtered_result[key] = f"Complex value (not displayable): {e}"

    # Defense-in-depth: ensure no bearer tokens leak into the rendered HTML even if
    # a non-conforming IdP places them in its userinfo response.
    safe_raw_claims: Final = {k: v for k, v in (received_response or {}).items() if k not in _OAUTH_TOKEN_FIELDS}
    safe_access_token_claims = {k: v for k, v in (access_token_payload or {}).items() if k not in _OAUTH_TOKEN_FIELDS}

    sso_payload: Final = {
        "parsed_by_proxy": filtered_result,
        "raw_claims": safe_raw_claims,
        "access_token_claims": safe_access_token_claims,
    }

    # Replace the placeholder in the template with the actual data
    sso_payload_json: Final = json.dumps(sso_payload, indent=2, default=str).replace("</", "<\\/")
    html_content: Final = jwt_display_template.replace(
        "const ssoData = SSO_DATA;",
        f"const ssoData = {sso_payload_json};",
    )

    return HTMLResponse(content=html_content)
