litellm/litellm/proxy/auth/auth_utils.py

import importlib.util
import os
import re
import sys
from collections.abc import Collection, Iterator, Mapping
from functools import lru_cache
from logging import Logger
from typing import Any, Final, Protocol

from fastapi import HTTPException, Request, status
from pydantic import PositiveInt, TypeAdapter, ValidationError

import litellm
from litellm import Router, constants, provider_list
from litellm._logging import verbose_proxy_logger
from litellm.constants import (
    BATCH_ENQUEUED_TOKEN_LIMIT_METADATA_KEY,
    EMPTY_MAPPING,
    INVALID_VIRTUAL_KEY_ERROR_MARKER,
    MINIMUM_CUSTOM_KEY_LENGTH,
    STANDARD_CUSTOMER_ID_HEADERS,
)
from litellm.litellm_core_utils.safe_json_loads import safe_json_loads
from litellm.litellm_core_utils.url_utils import (
    SSRFError,
    is_url_destination_allowed_by_host,
    provider_url_destination_candidates,
    validate_url,
)
from litellm.llms.azure.passthrough.transformation import azure_router_model_in_endpoint
from litellm.proxy._types import *
from litellm.proxy.common_utils.http_parsing_utils import extract_nested_form_metadata
from litellm.types.passthrough_endpoints.pass_through_endpoints import (
    LITELLM_PASS_THROUGH_ENDPOINT_MARKER,
)
from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS
from litellm.types.utils import CustomPricingLiteLLMParams


def is_invalid_virtual_key_error(exception: BaseException | None) -> bool:
    """True when an authentication error rejects a malformed virtual key.

    Classifies only by the marker stamped where that 401 is raised. Message
    content is never inspected: other 401s interpolate caller-supplied values
    (vector store ids, organization ids) into their messages, so a phrase
    match would let a request body demote an authorization failure to the
    quiet log path.
    """
    if not isinstance(exception, (HTTPException, ProxyException)):
        return False

    code: Final[object] = getattr(exception, "code", None)
    status_code: Final[object] = code if code is not None else getattr(exception, "status_code", None)
    if str(status_code) != str(status.HTTP_401_UNAUTHORIZED):
        return False

    return getattr(exception, INVALID_VIRTUAL_KEY_ERROR_MARKER, False) is True


def mark_invalid_virtual_key_error(exception: ProxyException, is_invalid_virtual_key: bool) -> ProxyException:
    """Return an independently marked malformed-key exception after callback transformations."""
    if not is_invalid_virtual_key or str(exception.code) != str(status.HTTP_401_UNAUTHORIZED):
        return exception
    marked_exception: Final = ProxyException(
        message=exception.message,
        type=exception.type,
        param=exception.param,
        code=exception.code,
        headers=exception.headers.copy(),
        openai_code=None if exception.openai_code is None else str(exception.openai_code),
        provider_specific_fields=exception.provider_specific_fields,
    )
    setattr(marked_exception, INVALID_VIRTUAL_KEY_ERROR_MARKER, True)
    return marked_exception


def _get_request_ip_address(request: Request, use_x_forwarded_for: bool | None = False) -> str | None:
    client_ip = None
    if use_x_forwarded_for is True and "x-forwarded-for" in request.headers:
        client_ip = request.headers["x-forwarded-for"]
    elif request.client is not None:
        client_ip = request.client.host
    else:
        client_ip = ""

    return client_ip


def _check_valid_ip(
    allowed_ips: list[str] | None,
    request: Request,
    use_x_forwarded_for: bool | None = False,
) -> tuple[bool, str | None]:
    """
    Returns if ip is allowed or not
    """
    if allowed_ips is None:  # if not set, assume true
        return True, None

    # if general_settings.get("use_x_forwarded_for") is True then use x-forwarded-for
    client_ip: Final = _get_request_ip_address(request=request, use_x_forwarded_for=use_x_forwarded_for)

    # Check if IP address is allowed
    if client_ip not in allowed_ips:
        return False, client_ip

    return True, client_ip


def check_complete_credentials(request_body: dict) -> bool:
    """
    if 'api_base' in request body. Check if complete credentials given. Prevent malicious attacks.

    Supplying an ``api_key`` is necessary but not sufficient: even with
    credentials supplied, an ``api_base`` / ``base_url`` that resolves to a
    private/internal/cloud-metadata address would still allow the proxy to
    be used as an SSRF pivot. Validate any URL fields here so the gate
    can't be bypassed with ``api_key=anything`` plus a malicious target.
    """
    given_model: str | None = None

    given_model = request_body.get("model")
    if given_model is None:
        return False

    if (
        "sagemaker" in given_model
        or "bedrock" in given_model
        or "vertex_ai" in given_model
        or "vertex_ai_beta" in given_model
    ):
        # complex credentials - easier to make a malicious request
        return False

    api_key_value: Final = request_body.get("api_key")
    if not (api_key_value and isinstance(api_key_value, str) and api_key_value.strip()):
        return False

    # ``validate_url`` itself doesn't consult the toggle; ``safe_get`` /
    # ``async_safe_get`` do. Mirror that here so admins who explicitly
    # disabled URL validation (e.g. for an internal Ollama endpoint they
    # accept the SSRF risk for) aren't blocked at the proxy boundary.
    if getattr(litellm, "user_url_validation", False):
        for url_field in ("api_base", "base_url"):
            url_value = request_body.get(url_field)
            if not url_value or not isinstance(url_value, str):
                continue
            try:
                validate_url(url_value)
            except SSRFError as e:
                raise ValueError(
                    f"Rejected request: client-side {url_field}={url_value!r} is rejected by the SSRF guard ({e})."
                )

    return True


def check_regex_or_str_match(request_body_value: Any, regex_str: str) -> bool:
    """
    Check if request_body_value matches the regex_str or is equal to param
    """
    if re.match(regex_str, request_body_value) or regex_str == request_body_value:
        return True
    return False


def _is_param_allowed(
    param: str,
    request_body_value: Any,
    configurable_clientside_auth_params: CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS,
) -> bool:
    """
    Check if param is a str or dict and if request_body_value is in the list of allowed values
    """
    if configurable_clientside_auth_params is None:
        return False

    for item in configurable_clientside_auth_params:
        if isinstance(item, str) and param == item:
            return True
        elif isinstance(item, dict):
            if param == "api_base" and check_regex_or_str_match(
                request_body_value=request_body_value,
                regex_str=item["api_base"],
            ):  # assume param is a regex
                return True

    return False


def _allow_model_level_clientside_configurable_parameters(
    model: str, param: str, request_body_value: Any, llm_router: Router | None
) -> bool:
    """
    Check if model is allowed to use configurable client-side params
    - get matching model
    - check if 'clientside_configurable_parameters' is set for model
    -
    """
    if llm_router is None:
        return False
    # check if model is set
    model_info = llm_router.get_model_group_info(model_group=model)
    if model_info is None:
        # check if wildcard model is set
        if model.split("/", 1)[0] in provider_list:
            model_info = llm_router.get_model_group_info(model_group=model.split("/", 1)[0])

    if model_info is None:
        return False

    if model_info is None or model_info.configurable_clientside_auth_params is None:
        return False

    return _is_param_allowed(
        param=param,
        request_body_value=request_body_value,
        configurable_clientside_auth_params=model_info.configurable_clientside_auth_params,
    )


# Config dicts whose entries are spread as ``**dict`` into outbound LLM
# API calls. ``litellm_embedding_config`` is consumed by the Milvus
# vector store transformer. ``extra_body`` is the OpenAI-SDK passthrough
# container: provider modules pull provider-auth fields out of it
# (e.g. Azure's ``extra_body.azure_ad_token``, Bedrock's
# ``extra_body.aws_web_identity_token``) without re-validating, so the
# banned-key check has to descend into it the same way it descends into
# ``litellm_embedding_config``.
_NESTED_CONFIG_KEYS: Final[tuple[str, ...]] = ("litellm_embedding_config", "extra_body")

# Metadata containers that carry per-request configuration consumed by the
# observability callbacks. The same banned-param list applies — a value
# under ``metadata.langfuse_host`` redirects the same Langfuse client and
# leaks the same credentials as the root-level ``langfuse_host``, but the
# original check only walked the request-body root, so the metadata path
# was an unintentional bypass.
_NESTED_METADATA_KEYS: Final[tuple[str, ...]] = ("metadata", "litellm_metadata")

# Banned request-body params. The same list applies to every entry in
# ``_NESTED_CONFIG_KEYS`` (dicts spread as ``**kwargs`` into outbound
# calls) and ``_NESTED_METADATA_KEYS`` (dicts read directly by integration
# callbacks), so a single banned name is enforced wherever the field can
# reach the call path from.
# Per-request observability params that are SAFE to accept from clients.
# These describe the request being logged (prompt version, sampling rate)
# without choosing the destination or the credentials, so they don't
# contribute to the data-exfil primitive that the rest of
# ``_supported_callback_params`` does.
_SAFE_CLIENT_CALLBACK_PARAMS: Final[frozenset[str]] = frozenset(
    {
        "langfuse_prompt_version",
        "langsmith_sampling_rate",
    }
)

# Observability fields that integrations read from the request body or
# metadata but that are not (yet) listed in ``_supported_callback_params``.
# Listed here so the proxy bans them today; the long-term cleanup is to
# fold these into the canonical allowlist so they share one source of
# truth with the rest.
_EXTRA_BANNED_OBSERVABILITY_PARAMS: Final[frozenset[str]] = frozenset(
    {
        "posthog_api_url",
        # ``phoenix_project_name`` / ``phoenix_project_name_override`` are NOT
        # banned: on the proxy the Phoenix integrations only read them from
        # ``user_api_key_auth_metadata`` (key/team config), so the bare request
        # fields are inert and rejecting them just breaks SDK-style callers.
        # Server-reserved: written exclusively by add_user_api_key_auth_to_request_metadata
        # from the authenticated key's database record.  A caller-supplied value
        # would survive the server merge and let an authenticated user redirect
        # their Arize/Phoenix telemetry into arbitrary projects.
        "user_api_key_auth_metadata",
        "wandb_api_key",
        "weave_project_id",
    }
)


def _build_banned_observability_params() -> frozenset[str]:
    """Derive the observability ban list from the canonical allowlist.

    ``_supported_callback_params`` and ``_request_blocked_callback_params`` in
    ``litellm/litellm_core_utils/initialize_dynamic_callback_params.py`` is
    the single place that enumerates every observability field integrations
    resolve from kwargs/metadata, plus fields that integration code explicitly
    blocks from request-supplied callback params. Subtract the small set of
    informational fields (``_SAFE_CLIENT_CALLBACK_PARAMS``) and union with the
    extras the canonical allowlist hasn't caught up to yet. New integrations
    added to the canonical allowlist are banned by default, which is the safe
    failure mode.
    """
    from litellm.litellm_core_utils.initialize_dynamic_callback_params import (
        _request_blocked_callback_params,
        _supported_callback_params,
    )

    return (
        (frozenset(_supported_callback_params) - _SAFE_CLIENT_CALLBACK_PARAMS)
        | frozenset(_request_blocked_callback_params)
        | _EXTRA_BANNED_OBSERVABILITY_PARAMS
    )


_BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = (
    "api_base",
    "base_url",
    "user_config",
    "aws_sts_endpoint",
    "aws_web_identity_token",
    "aws_role_name",
    # Remaining AWS identity selectors. ``get_credentials`` prefers a named
    # profile over the deployment's static keys, so a caller-supplied
    # ``aws_profile_name`` signs Bedrock and S3 requests as any profile
    # present on the proxy host; the two AssumeRole knobs are banned with it
    # so the whole identity-selection family lives behind the same opt-in.
    "aws_profile_name",
    "aws_session_name",
    "aws_external_id",
    "vertex_credentials",
    # Azure managed-identity / federated-auth token. The Azure provider
    # transformer reads ``azure_ad_token`` (top-level or via
    # ``extra_body``) and resolves it through ``get_secret`` before
    # passing it as the bearer token to the Azure endpoint, so a
    # caller-supplied value is the same exfil shape as
    # ``aws_web_identity_token`` on the Bedrock path.
    "azure_ad_token",
    # Endpoint-targeting fields that retarget the outbound request or
    # an observability callback. An attacker-controlled value either
    # exfiltrates the request payload (incl. messages + admin-set
    # tokens) to the attacker's host, or coerces the proxy into
    # authenticating against the attacker's host with admin secrets.
    "aws_bedrock_runtime_endpoint",
    # Bedrock project/workspace association. Deployments pin this to
    # enforce a data-retention policy, so a caller-supplied value would
    # re-route the request's retention and accounting to any project
    # reachable with the deployment's shared AWS credentials.
    "aws_bedrock_project_id",
    "bedrock_tags",
    # Provider-specific endpoint overrides that flow into the outbound
    # request via ``optional_params``. Same threat as ``api_base``:
    # ``s3_endpoint_url`` redirects Bedrock file uploads to attacker
    # S3; ``sagemaker_base_url`` redirects all SageMaker traffic;
    # ``deployment_url`` redirects SAP deployments.
    "s3_endpoint_url",
    "sagemaker_base_url",
    "deployment_url",
    # NVIDIA Riva fields consumed by the audio-transcription handler
    # via ``optional_params``. Banned for the same reason as the
    # provider-specific entries above: a caller-supplied value retargets
    # the request away from the admin's pinned configuration.
    "nvcf_function_id",
    "use_ssl",
    # Per-deployment opt-in that hands the whole call to the Rust core. It is a
    # deployment decision, not a request one: the Rust path uses its own client
    # rather than the one the deployment configured, and reports no post_call,
    # so a caller-supplied value picks a transport and a callback surface the
    # admin did not choose.
    "rust",
    # SDK-only field; also rejected outright in is_request_body_safe.
    "model_list",
    "vertex_ai_credentials",
    # Observability credentials, hosts, and project identifiers: derived
    # from the canonical ``_supported_callback_params`` allowlist so new
    # integrations are covered automatically. Sorted for stable iteration
    # order and reviewable diffs.
    *sorted(_build_banned_observability_params()),
    *sorted(CustomPricingLiteLLMParams.model_fields.keys()),
)


def _check_banned_params(
    body: dict,
    general_settings: dict,
    llm_router: Router | None,
    model: str,
) -> None:
    """Raise ``ValueError`` if ``body`` carries a banned param without admin opt-in.

    Shared between the root-level check and the nested-config check so a
    new banned param only needs to be added in one place.
    """
    for param in _BANNED_REQUEST_BODY_PARAMS:
        if param not in body:
            continue
        if general_settings.get("allow_client_side_credentials") is True:
            # Proxy-wide opt-in: every banned param is permitted, exit
            # entirely so the rest of the loop doesn't waste work.
            return
        if (
            _allow_model_level_clientside_configurable_parameters(
                model=model,
                param=param,
                request_body_value=body[param],
                llm_router=llm_router,
            )
            is True
        ):
            # Per-param opt-in: only THIS param is permitted by the
            # deployment's ``configurable_clientside_auth_params``. Skip
            # to the next banned param so a body that pairs an allowed
            # ``api_base`` with an unallowed ``langfuse_host`` is still
            # rejected for the second field.
            continue
        raise ValueError(
            f"Rejected Request: {param} is not allowed in request body. "
            "Clientside passthrough requires explicit admin opt-in via "
            "either `general_settings.allow_client_side_credentials = true` "
            "(proxy-wide) or `configurable_clientside_auth_params` on the "
            "deployment in your proxy config.yaml. "
            "Relevant Issue: https://huntr.com/bounties/4001e1a2-7b7a-4776-a3ae-e6692ec3d997",
        )


_FALLBACK_FIELDS: Final[tuple[str, ...]] = (
    "fallbacks",
    "context_window_fallbacks",
    "content_policy_fallbacks",
)


def _iter_fallback_field_values(request_body: Mapping[str, object]) -> Iterator[object]:
    override: Final = request_body.get("router_settings_override")
    for source in (request_body, override):
        if isinstance(source, Mapping):
            for field in _FALLBACK_FIELDS:
                yield source.get(field)


def _iter_fallback_targets(value: object, depth: int) -> Iterator[str | Mapping[str, object]]:
    if depth > 2 * litellm.ROUTER_MAX_FALLBACKS:
        raise ValueError("Rejected Request: fallback nesting exceeds the allowed validation depth.")
    if not isinstance(value, list):
        return
    for item in value:
        if isinstance(item, str):
            yield item
        elif isinstance(item, Mapping):
            values = tuple(item.values())
            if not (values and all(isinstance(v, list) for v in values)):
                yield item
            if isinstance(item.get("model"), str):
                for field in _FALLBACK_FIELDS:
                    yield from _iter_fallback_targets(item.get(field), depth + 1)
            else:
                for target_list in values:
                    yield from _iter_fallback_targets(target_list, depth + 1)


def iter_request_fallback_targets(request_body: Mapping[str, object]) -> Iterator[str | Mapping[str, object]]:
    for value in _iter_fallback_field_values(request_body):
        yield from _iter_fallback_targets(value, 0)


def _reject_url_valued_fallback_target(value: str) -> None:
    allowed_hosts: Final = getattr(litellm, "provider_url_destination_allowed_hosts", []) or []
    for candidate in provider_url_destination_candidates(value):
        if not candidate.lower().startswith(("http://", "https://")):
            continue
        if is_url_destination_allowed_by_host(candidate, allowed_hosts):
            continue
        raise ValueError(
            f"Rejected Request: URL-valued fallback destination '{value}' is not allowed. "
            "Configure custom endpoints with api_base instead, or add the destination host to "
            "`provider_url_destination_allowed_hosts` in litellm_settings."
        )


def is_request_body_safe(request_body: dict, general_settings: dict, llm_router: Router | None, model: str) -> bool:
    """
    Check if the request body is safe.

    A malicious user can set the api_base to their own domain and invoke POST /chat/completions to intercept and steal the OpenAI API key.
    Relevant issue: https://huntr.com/bounties/4001e1a2-7b7a-4776-a3ae-e6692ec3d997

    The blocklist is enforced unconditionally. Legitimate clientside
    credential / endpoint passthrough goes through one of the two
    explicit admin opt-ins (``general_settings.allow_client_side_credentials``
    proxy-wide or ``configurable_clientside_auth_params`` per deployment).
    Historically there was a third, *implicit*, *caller-controlled* path:
    ``check_complete_credentials`` returned True when the caller supplied
    any non-empty ``api_key``, which made the entire blocklist a no-op.
    That bypass turned every missing entry on the blocklist into an
    exploitable SSRF / credential-exfil hole — see GHSA-jh89-88fc-qrfp,
    GHSA-3frq-6r6h-7j64, and the chain of veria-admin findings (Dv_m860l,
    b_yRJeQ5, stN90yjP, LBlyOAc8, U2TD78kg). Removed: the blocklist now
    has a single, predictable failure mode for missing entries (a 400),
    not a credential leak.

    Iterative single-level descent into ``_NESTED_CONFIG_KEYS`` (rather
    than recursion) covers nested-config attacks like Milvus's
    ``litellm_embedding_config.api_base`` (VERIA-6) without exposing a
    recursion-depth DoS surface.
    """
    if "model_list" in request_body:
        raise ValueError("Rejected Request: model_list is not allowed in the request body.")
    _check_banned_params(request_body, general_settings, llm_router, model)
    for nested_key in _NESTED_CONFIG_KEYS:
        nested = _coerce_metadata_to_dict(request_body.get(nested_key))
        if nested is not None:
            _check_banned_params(nested, general_settings, llm_router, model)
    for metadata_key in _NESTED_METADATA_KEYS:
        metadata = _coerce_metadata_to_dict(request_body.get(metadata_key))
        if metadata is not None:
            _check_banned_params(metadata, general_settings, llm_router, model)
        if any(isinstance(key, str) and key.startswith(f"{metadata_key}[") for key in request_body):
            _check_banned_params(
                extract_nested_form_metadata(form_data=request_body, prefix=f"{metadata_key}["),
                general_settings,
                llm_router,
                model,
            )
    for target in iter_request_fallback_targets(request_body):
        if isinstance(target, dict):
            _check_banned_params(target, general_settings, llm_router, model)
            target_model = target.get("model")
            if isinstance(target_model, str):
                _reject_url_valued_fallback_target(target_model)
        elif isinstance(target, str):
            _reject_url_valued_fallback_target(target)
    litellm_params: Final = _coerce_metadata_to_dict(request_body.get("litellm_params"))
    if litellm_params is not None:
        litellm_params_metadata: Final = _coerce_metadata_to_dict(litellm_params.get("metadata"))
        if litellm_params_metadata is not None:
            _check_banned_params(
                litellm_params_metadata,
                general_settings,
                llm_router,
                model,
            )
    return True


def _coerce_metadata_to_dict(value: Any) -> dict[str, Any] | None:
    """Return ``value`` as a dict, parsing it from JSON if delivered as a string.

    Multipart/form-data and ``extra_body`` callers send ``litellm_metadata``
    as a JSON-encoded string; the proxy parses it into a dict later in
    ``add_litellm_data_to_request``, but the auth-time bouncer runs first
    and would otherwise miss the banned-param check on a still-stringified
    metadata blob.
    """
    if isinstance(value, dict):
        return value
    if isinstance(value, str):
        from litellm.litellm_core_utils.safe_json_loads import safe_json_loads

        parsed: Final = safe_json_loads(value)
        if isinstance(parsed, dict):
            return parsed
    return None


async def pre_db_read_auth_checks(
    request: Request,
    request_data: dict,
    route: str,
):
    """
    1. Checks if request size is under max_request_size_mb (if set)
    2. Check if request body is safe (example user has not set api_base in request body)
    3. Check if IP address is allowed (if set)
    4. Check if request route is an allowed route on the proxy (if set)

    Returns:
    - True

    Raises:
    - HTTPException if request fails initial auth checks
    """
    from litellm.proxy.proxy_server import general_settings, llm_router, premium_user

    # Check 1. request size
    await check_if_request_size_is_safe(request=request)

    # Check 2. Request body is safe
    is_request_body_safe(
        request_body=request_data,
        general_settings=general_settings,
        llm_router=llm_router,
        model=request_data.get("model", ""),  # [TODO] use model passed in url as well (azure openai routes)
    )

    # Check 3. Check if IP address is allowed
    is_valid_ip, passed_in_ip = _check_valid_ip(
        allowed_ips=general_settings.get("allowed_ips", None),
        use_x_forwarded_for=general_settings.get("use_x_forwarded_for", False),
        request=request,
    )

    if not is_valid_ip:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail=f"Access forbidden: IP address {passed_in_ip} not allowed.",
        )

    # Check 4. Check if request route is an allowed route on the proxy
    if "allowed_routes" in general_settings:
        _allowed_routes: Final = general_settings["allowed_routes"]
        if premium_user is not True:
            verbose_proxy_logger.error(
                "Trying to set allowed_routes. This is an Enterprise feature. %s",
                CommonProxyErrors.not_premium_user.value,
            )
        if route not in _allowed_routes:
            verbose_proxy_logger.error("Route %s not in allowed_routes=%s", route, _allowed_routes)
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Access forbidden: Route {route} not allowed",
            )


def route_in_additonal_public_routes(current_route: str):
    """
    Helper to check if the user defined public_routes on config.yaml

    Parameters:
    - current_route: str - the route the user is trying to call

    Returns:
    - bool - True if the route is defined in public_routes
    - bool - False if the route is not defined in public_routes

    Supports wildcard patterns (e.g., "/api/*" matches "/api/users", "/api/users/123")

    In order to use this the litellm config.yaml should have the following in general_settings:

    ```yaml
    general_settings:
        master_key: sk-1234
        public_routes: ["LiteLLMRoutes.public_routes", "/spend/calculate", "/api/*"]
    ```
    """
    from litellm.proxy.auth.route_checks import RouteChecks
    from litellm.proxy.proxy_server import general_settings, premium_user

    try:
        if premium_user is not True:
            return False
        if general_settings is None:
            return False

        routes_defined: Final = general_settings.get("public_routes", [])

        # Check exact match first
        if current_route in routes_defined:
            return True

        # Check wildcard patterns
        for route_pattern in routes_defined:
            if RouteChecks.route_matches_wildcard_pattern(route=current_route, pattern=route_pattern):
                return True

        return False
    except Exception as e:
        verbose_proxy_logger.error("route_in_additonal_public_routes: %s", e)
        return False


def get_request_route(request: Request) -> str:
    """
    Resolve the request route from the ASGI scope, with ``root_path`` stripped.

    Prefer this over ``request.url.path`` for any auth, ACL, routing, or
    audit-log decision: Starlette reconstructs ``url.path`` by interpolating
    the Host header into a URL string and re-parsing with ``urlsplit``, so a
    malformed Host (e.g. ``localhost/?x=1``) collapses ``url.path`` to ``"/"``
    while FastAPI continues to dispatch on ``scope["path"]``. ``scope["path"]``
    is uvicorn's parse of the HTTP request line and matches the actual
    handler, so it's the authoritative route.

    Also normalizes sub-path deployments by stripping ``scope["root_path"]``
    e.g. ``/genai/chat/completions`` -> ``/chat/completions``.
    """
    try:
        scope: Final = request.scope
        if not isinstance(scope, dict):
            return str(request.url.path)
        raw_path: Final[str] = str(scope.get("path", request.url.path))
        root_path: Final[str] = str(scope.get("app_root_path", scope.get("root_path", ""))).rstrip("/")
        if not isinstance(raw_path, str):
            return str(request.url.path)
        # Strip root_path only when it matches whole path segments — guarding
        # against sibling paths like "/apifoo" being truncated under
        # root_path="/api". Trailing slashes on root_path are stripped above,
        # so bare "/" or "/prefix/" still leave the leading "/" intact.
        if root_path and (raw_path == root_path or raw_path.startswith(root_path + "/")):
            stripped: Final = raw_path[len(root_path) :]
            return stripped or "/"
        return raw_path
    except Exception as e:
        verbose_proxy_logger.debug(
            "error on get_request_route: %s, defaulting to request.url.path=%s", e, request.url.path
        )
        return str(request.url.path)


def get_request_route_template(request: Request) -> str | None:
    """
    Return the low-cardinality route template, e.g.
    ``/v1/threads/{thread_id}/runs`` (vs. the literal path from
    ``get_request_route``). FastAPI sets ``scope["route"]`` before endpoint
    dependencies run. Returns None if unavailable (unmatched path, Mount).
    """
    try:
        scope: Final = request.scope
        if not isinstance(scope, dict):
            return None
        route: Final = scope.get("route")
        template: Final = getattr(route, "path", None)
        return template if isinstance(template, str) and template else None
    except Exception as e:
        verbose_proxy_logger.debug("error on get_request_route_template: %s", e)
        return None


@lru_cache(maxsize=256)
def normalize_request_route(route: str) -> str:
    """
    Normalize request routes by replacing dynamic path parameters with placeholders.

    This prevents high cardinality in Prometheus metrics by collapsing routes like:
    - /v1/responses/1234567890 -> /v1/responses/{response_id}
    - /v1/threads/thread_123 -> /v1/threads/{thread_id}

    Args:
        route: The request route path

    Returns:
        Normalized route with dynamic parameters replaced by placeholders

    Examples:
        >>> normalize_request_route("/v1/responses/abc123")
        '/v1/responses/{response_id}'
        >>> normalize_request_route("/v1/responses/abc123/cancel")
        '/v1/responses/{response_id}/cancel'
        >>> normalize_request_route("/chat/completions")
        '/chat/completions'
    """
    # Define patterns for routes with dynamic IDs
    # Format: (regex_pattern, replacement_template)
    patterns: Final = [
        # Responses API - must come before generic patterns
        (r"^(/(?:openai/)?v1/responses)/([^/]+)(/input_items)$", r"\1/{response_id}\3"),
        (r"^(/(?:openai/)?v1/responses)/([^/]+)(/cancel)$", r"\1/{response_id}\3"),
        (r"^(/(?:openai/)?v1/responses)/([^/]+)$", r"\1/{response_id}"),
        (r"^(/responses)/([^/]+)(/input_items)$", r"\1/{response_id}\3"),
        (r"^(/responses)/([^/]+)(/cancel)$", r"\1/{response_id}\3"),
        (r"^(/responses)/([^/]+)$", r"\1/{response_id}"),
        # Threads API
        (
            r"^(/(?:openai/)?v1/threads)/([^/]+)(/runs)/([^/]+)(/steps)/([^/]+)$",
            r"\1/{thread_id}\3/{run_id}\5/{step_id}",
        ),
        (
            r"^(/(?:openai/)?v1/threads)/([^/]+)(/runs)/([^/]+)(/steps)$",
            r"\1/{thread_id}\3/{run_id}\5",
        ),
        (
            r"^(/(?:openai/)?v1/threads)/([^/]+)(/runs)/([^/]+)(/cancel)$",
            r"\1/{thread_id}\3/{run_id}\5",
        ),
        (
            r"^(/(?:openai/)?v1/threads)/([^/]+)(/runs)/([^/]+)(/submit_tool_outputs)$",
            r"\1/{thread_id}\3/{run_id}\5",
        ),
        (
            r"^(/(?:openai/)?v1/threads)/([^/]+)(/runs)/([^/]+)$",
            r"\1/{thread_id}\3/{run_id}",
        ),
        (r"^(/(?:openai/)?v1/threads)/([^/]+)(/runs)$", r"\1/{thread_id}\3"),
        (
            r"^(/(?:openai/)?v1/threads)/([^/]+)(/messages)/([^/]+)$",
            r"\1/{thread_id}\3/{message_id}",
        ),
        (r"^(/(?:openai/)?v1/threads)/([^/]+)(/messages)$", r"\1/{thread_id}\3"),
        (r"^(/(?:openai/)?v1/threads)/([^/]+)$", r"\1/{thread_id}"),
        # Vector Stores API
        (
            r"^(/(?:openai/)?v1/vector_stores)/([^/]+)(/files)/([^/]+)$",
            r"\1/{vector_store_id}\3/{file_id}",
        ),
        (
            r"^(/(?:openai/)?v1/vector_stores)/([^/]+)(/files)$",
            r"\1/{vector_store_id}\3",
        ),
        (
            r"^(/(?:openai/)?v1/vector_stores)/([^/]+)(/file_batches)/([^/]+)$",
            r"\1/{vector_store_id}\3/{batch_id}",
        ),
        (
            r"^(/(?:openai/)?v1/vector_stores)/([^/]+)(/file_batches)$",
            r"\1/{vector_store_id}\3",
        ),
        (r"^(/(?:openai/)?v1/vector_stores)/([^/]+)$", r"\1/{vector_store_id}"),
        # Assistants API
        (r"^(/(?:openai/)?v1/assistants)/([^/]+)$", r"\1/{assistant_id}"),
        # Files API
        (r"^(/(?:openai/)?v1/files)/([^/]+)(/content)$", r"\1/{file_id}\3"),
        (r"^(/(?:openai/)?v1/files)/([^/]+)$", r"\1/{file_id}"),
        # Batches API
        (r"^(/(?:openai/)?v1/batches)/([^/]+)(/cancel)$", r"\1/{batch_id}\3"),
        (r"^(/(?:openai/)?v1/batches)/([^/]+)$", r"\1/{batch_id}"),
        # Fine-tuning API
        (
            r"^(/(?:openai/)?v1/fine_tuning/jobs)/([^/]+)(/events)$",
            r"\1/{fine_tuning_job_id}\3",
        ),
        (
            r"^(/(?:openai/)?v1/fine_tuning/jobs)/([^/]+)(/cancel)$",
            r"\1/{fine_tuning_job_id}\3",
        ),
        (
            r"^(/(?:openai/)?v1/fine_tuning/jobs)/([^/]+)(/checkpoints)$",
            r"\1/{fine_tuning_job_id}\3",
        ),
        (r"^(/(?:openai/)?v1/fine_tuning/jobs)/([^/]+)$", r"\1/{fine_tuning_job_id}"),
        # Models API
        (r"^(/(?:openai/)?v1/models)/([^/]+)$", r"\1/{model}"),
    ]

    # Apply patterns in order
    for pattern, replacement in patterns:
        normalized = re.sub(pattern, replacement, route)
        if normalized != route:
            return normalized

    # Return original route if no pattern matched
    return route


async def check_if_request_size_is_safe(request: Request) -> bool:
    """
    Enterprise Only:
        - Checks if the request size is within the limit

    Args:
        request (Request): The incoming request.

    Returns:
        bool: True if the request size is within the limit

    Raises:
        ProxyException: If the request size is too large

    """
    from litellm.proxy.proxy_server import general_settings, premium_user

    max_request_size_mb: Final = general_settings.get("max_request_size_mb", None)

    if max_request_size_mb is not None:
        # Check if premium user
        if premium_user is not True:
            verbose_proxy_logger.warning(
                "using max_request_size_mb - not checking -  this is an enterprise only feature. %s",
                CommonProxyErrors.not_premium_user.value,
            )
            return True

        # Get the request body
        content_length: Final = request.headers.get("content-length")

        if content_length:
            header_size: Final = int(content_length)
            header_size_mb: Final = bytes_to_mb(bytes_value=header_size)
            verbose_proxy_logger.debug("content_length request size in MB=%s", header_size_mb)

            if header_size_mb > max_request_size_mb:
                raise ProxyException(
                    message=f"Request size is too large. Request size is {header_size_mb} MB. Max size is {max_request_size_mb} MB",
                    type=ProxyErrorTypes.bad_request_error.value,
                    code=400,
                    param="content-length",
                )
        else:
            # If Content-Length is not available, read the body
            body: Final = await request.body()
            body_size: Final = len(body)
            request_size_mb: Final = bytes_to_mb(bytes_value=body_size)

            verbose_proxy_logger.debug("request body request size in MB=%s", request_size_mb)
            if request_size_mb > max_request_size_mb:
                raise ProxyException(
                    message=f"Request size is too large. Request size is {request_size_mb} MB. Max size is {max_request_size_mb} MB",
                    type=ProxyErrorTypes.bad_request_error.value,
                    code=400,
                    param="content-length",
                )

    return True


async def check_response_size_is_safe(response: Any) -> bool:
    """
    Enterprise Only:
        - Checks if the response size is within the limit

    Args:
        response (Any): The response to check.

    Returns:
        bool: True if the response size is within the limit

    Raises:
        ProxyException: If the response size is too large

    """

    from litellm.proxy.proxy_server import general_settings, premium_user

    max_response_size_mb: Final = general_settings.get("max_response_size_mb", None)
    if max_response_size_mb is not None:
        # Check if premium user
        if premium_user is not True:
            verbose_proxy_logger.warning(
                "using max_response_size_mb - not checking -  this is an enterprise only feature. %s",
                CommonProxyErrors.not_premium_user.value,
            )
            return True

        response_size_mb: Final = bytes_to_mb(bytes_value=sys.getsizeof(response))
        verbose_proxy_logger.debug("response size in MB=%s", response_size_mb)
        if response_size_mb > max_response_size_mb:
            raise ProxyException(
                message=f"Response size is too large. Response size is {response_size_mb} MB. Max size is {max_response_size_mb} MB",
                type=ProxyErrorTypes.bad_request_error.value,
                code=400,
                param="content-length",
            )

    return True


def bytes_to_mb(bytes_value: int):
    """
    Helper to convert bytes to MB
    """
    return bytes_value / (1024 * 1024)


# helpers used by parallel request limiter to handle model rpm/tpm limits for a given api key
def _get_deployment_default_limit(model_name: str, field: str) -> int | None:
    """
    Return the minimum value of `field` across all deployments for model_name,
    or None if no deployment has the field set.

    When multiple deployments share the same model name, taking the minimum is
    the safest choice for load-balanced setups: it ensures no deployment is
    over-consumed regardless of which one actually serves a given request.
    """
    from litellm.proxy.proxy_server import llm_router

    if llm_router is None:
        return None
    deployments: Final = llm_router.get_model_list(model_name=model_name)
    if not deployments:
        return None
    limits: Final = []
    for deployment in deployments:
        raw = deployment.get("litellm_params", {}).get(field)
        if raw is not None:
            try:
                if isinstance(raw, (int, float, str, bytes, bytearray)):
                    limits.append(int(raw))
            except (ValueError, TypeError):
                pass
    return min(limits) if limits else None


def _get_deployment_default_rpm_limit(model_name: str) -> int | None:
    return _get_deployment_default_limit(model_name, "default_api_key_rpm_limit")


def _get_deployment_default_tpm_limit(model_name: str) -> int | None:
    return _get_deployment_default_limit(model_name, "default_api_key_tpm_limit")


def get_key_model_rpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
    model_name: str | None = None,
) -> dict[str, int] | None:
    """
    Get the model rpm limit for a given api key.

    Priority order (returns first found):
    1. Key metadata (model_rpm_limit)
    2. Key model_max_budget (rpm_limit per model)
    3. Team metadata (model_rpm_limit)
    4. Deployment default_api_key_rpm_limit (when model_name is provided)
    """
    # 1. Check key metadata first (takes priority)
    if user_api_key_dict.metadata:
        result: Final = user_api_key_dict.metadata.get("model_rpm_limit")
        if result:
            return result

    # 2. Check model_max_budget
    if user_api_key_dict.model_max_budget:
        model_rpm_limit: Final[dict[str, int]] = {}
        for model, budget in user_api_key_dict.model_max_budget.items():
            if isinstance(budget, dict) and budget.get("rpm_limit") is not None:
                model_rpm_limit[model] = budget["rpm_limit"]
        if model_rpm_limit:
            return model_rpm_limit

    # 3. Fallback to team metadata
    if user_api_key_dict.team_metadata:
        team_limit: Final = user_api_key_dict.team_metadata.get("model_rpm_limit")
        if team_limit is not None:
            return team_limit

    # 4. Fallback to deployment default_api_key_rpm_limit
    if model_name is not None:
        default_limit: Final = _get_deployment_default_rpm_limit(model_name)
        if default_limit is not None:
            return {model_name: default_limit}

    return None


def get_key_model_tpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
    model_name: str | None = None,
) -> dict[str, int] | None:
    """
    Get the model tpm limit for a given api key.

    Priority order (returns first found):
    1. Key metadata (model_tpm_limit)
    2. Key model_max_budget (tpm_limit per model)
    3. Team metadata (model_tpm_limit)
    4. Deployment default_api_key_tpm_limit (when model_name is provided)
    """
    # 1. Check key metadata first (takes priority)
    if user_api_key_dict.metadata:
        result: Final = user_api_key_dict.metadata.get("model_tpm_limit")
        if result:
            return result

    # 2. Check model_max_budget (iterate per-model like RPM does)
    if user_api_key_dict.model_max_budget:
        model_tpm_limit: Final[dict[str, int]] = {}
        for model, budget in user_api_key_dict.model_max_budget.items():
            if isinstance(budget, dict) and budget.get("tpm_limit") is not None:
                model_tpm_limit[model] = budget["tpm_limit"]
        if model_tpm_limit:
            return model_tpm_limit

    # 3. Fallback to team metadata
    if user_api_key_dict.team_metadata:
        team_limit: Final = user_api_key_dict.team_metadata.get("model_tpm_limit")
        if team_limit is not None:
            return team_limit

    # 4. Fallback to deployment default_api_key_tpm_limit
    if model_name is not None:
        default_limit: Final = _get_deployment_default_tpm_limit(model_name)
        if default_limit is not None:
            return {model_name: default_limit}

    return None


ESTIMATED_OUTPUT_TOKENS_FIELD: Final = "default_estimated_output_tokens"
ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD: Final = "default_estimated_output_tokens_per_model"
ESTIMATED_OUTPUT_TOKENS_METADATA_FIELDS: Final = frozenset(
    {ESTIMATED_OUTPUT_TOKENS_FIELD, ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD}
)

_ESTIMATED_OUTPUT_TOKENS_ADAPTER: Final = TypeAdapter(PositiveInt)
_ESTIMATED_OUTPUT_TOKENS_PER_MODEL_ADAPTER: Final = TypeAdapter(Mapping[str, PositiveInt])


def _validated_output_token_estimate(raw: object) -> int | None:
    """Coerce one declared estimate to a positive int, or ignore it."""
    if raw is None:
        return None
    try:
        return _ESTIMATED_OUTPUT_TOKENS_ADAPTER.validate_python(raw)
    except ValidationError as validation_error:
        verbose_proxy_logger.warning(
            "Ignoring malformed %s in metadata: %s",
            ESTIMATED_OUTPUT_TOKENS_FIELD,
            validation_error,
        )
        return None


def _validated_output_token_estimates_per_model(raw: object) -> Mapping[str, int] | None:
    """Coerce a declared per-model estimate map, or ignore it."""
    if raw is None:
        return None
    try:
        return _ESTIMATED_OUTPUT_TOKENS_PER_MODEL_ADAPTER.validate_python(raw)
    except ValidationError as validation_error:
        verbose_proxy_logger.warning(
            "Ignoring malformed %s in metadata: %s",
            ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD,
            validation_error,
        )
        return None


def _estimated_output_tokens_from_metadata(
    metadata: Mapping[str, object] | None,
    model_name: str | None,
) -> int | None:
    """Resolve the per-model, then global, estimate out of one metadata blob.

    The two fields are validated independently so a malformed per-model map
    cannot discard a valid global estimate, or the other way round.
    """
    if not metadata or ESTIMATED_OUTPUT_TOKENS_METADATA_FIELDS.isdisjoint(metadata):
        return None

    if model_name is not None:
        per_model: Final = _validated_output_token_estimates_per_model(
            metadata.get(ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD)
        )
        per_model_estimate: Final = per_model.get(model_name) if per_model is not None else None
        if per_model_estimate is not None:
            return per_model_estimate

    return _validated_output_token_estimate(metadata.get(ESTIMATED_OUTPUT_TOKENS_FIELD))


def get_estimated_output_tokens(
    user_api_key_dict: UserAPIKeyAuth,
    model_name: str | None = None,
) -> int | None:
    """Resolve the operator-declared output-token estimate for TPM reservation.

    Priority order (returns first found):
    1. Key metadata ``default_estimated_output_tokens_per_model[model_name]``
    2. Key metadata ``default_estimated_output_tokens``
    3. Team metadata ``default_estimated_output_tokens_per_model[model_name]``
    4. Team metadata ``default_estimated_output_tokens``

    Returns ``None`` when nothing is configured, which leaves the static
    heuristic floor in place.
    """
    key_estimate: Final = _estimated_output_tokens_from_metadata(user_api_key_dict.metadata, model_name)
    if key_estimate is not None:
        return key_estimate
    return _estimated_output_tokens_from_metadata(user_api_key_dict.team_metadata, model_name)


class OutputTokenEstimateRequest(Protocol):
    """The shape of any management request that can carry an output-token estimate.

    Read-only members: the gate inspects a request, it never writes one back.
    """

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

    @property
    def default_estimated_output_tokens(self) -> int | None: ...

    @property
    def default_estimated_output_tokens_per_model(self) -> Mapping[str, int] | None: ...

    @property
    def model_fields_set(self) -> Collection[str]: ...


def _requested_output_token_estimates(
    data: OutputTokenEstimateRequest,
    existing_metadata: Mapping[str, object],
) -> tuple[object, object]:
    """The output-token estimates this request would leave stored on the entity.

    Mirrors how the management endpoints merge metadata: a supplied ``metadata``
    replaces the stored blob wholesale, an omitted one preserves it, and the
    dedicated top-level fields overlay whatever survives. Both sources are read
    because the same declaration reaches the same stored field either way.
    """
    base: Final[Mapping[str, object]] = (
        (data.metadata or {}) if "metadata" in data.model_fields_set else existing_metadata
    )
    return (
        data.default_estimated_output_tokens
        if data.default_estimated_output_tokens is not None
        else base.get(ESTIMATED_OUTPUT_TOKENS_FIELD),
        data.default_estimated_output_tokens_per_model
        if data.default_estimated_output_tokens_per_model is not None
        else base.get(ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD),
    )


def enforce_output_token_estimates_are_admin_only(
    data: OutputTokenEstimateRequest,
    existing_metadata: Mapping[str, object] | None,
    user_api_key_dict: UserAPIKeyAuth,
    entity: Literal["key", "team"],
) -> None:
    """Only a proxy admin may change what a key or team declares its models emit.

    That declaration is what the TPM limiter reserves for a request omitting
    ``max_tokens``, so lowering or clearing it under-reserves against every
    window the request is charged against, including the team and organization
    ones the writer may not own. A key's metadata is writable by its holder and
    a team's by its team admin, so neither is a trustworthy source for a value
    that weakens a limit set above them. Gated on the resulting value rather
    than on presence, so a form resending the stored declaration stays a no-op.
    """
    if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value:
        return
    stored: Final[Mapping[str, object]] = existing_metadata or {}
    if _requested_output_token_estimates(data, stored) == (
        stored.get(ESTIMATED_OUTPUT_TOKENS_FIELD),
        stored.get(ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD),
    ):
        return
    raise HTTPException(
        status_code=403,
        detail={
            "error": f"Only proxy admins can set {ESTIMATED_OUTPUT_TOKENS_FIELD} or "
            f"{ESTIMATED_OUTPUT_TOKENS_PER_MODEL_FIELD} on a {entity}. They decide how many output tokens "
            "the rate limiter reserves for a request that omits max_tokens."
        },
    )


class BatchEnqueuedTokenLimitRequest(Protocol):
    """The shape of any management request that can carry a batch enqueued-token limit."""

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

    @property
    def model_fields_set(self) -> Collection[str]: ...


def enforce_batch_enqueued_token_limit_is_admin_only(
    data: BatchEnqueuedTokenLimitRequest,
    existing_metadata: Mapping[str, object] | None,
    user_api_key_dict: UserAPIKeyAuth,
    entity: Literal["key", "team"],
) -> None:
    """Only a proxy admin may change a key or team's batch enqueued-token limit.

    When set, ``batch_enqueued_token_limit`` replaces the standard RPM/TPM checks
    for batch submissions, so a holder-writable copy would let a caller lift their
    own batch quota. Gated on the resulting value rather than on presence, so a
    form resending the stored value stays a no-op.
    """
    if user_api_key_dict.user_role == LitellmUserRoles.PROXY_ADMIN.value:
        return
    stored: Final[Mapping[str, object]] = existing_metadata or EMPTY_MAPPING
    requested: Final[Mapping[str, object]] = (
        (data.metadata or EMPTY_MAPPING) if "metadata" in data.model_fields_set else stored
    )
    if requested.get(BATCH_ENQUEUED_TOKEN_LIMIT_METADATA_KEY) == stored.get(BATCH_ENQUEUED_TOKEN_LIMIT_METADATA_KEY):
        return
    raise HTTPException(
        status_code=403,
        detail={  # mutable-ok: HTTPException.detail has no immutable form
            "error": f"Only proxy admins can set {BATCH_ENQUEUED_TOKEN_LIMIT_METADATA_KEY} on a {entity}. "
            "It replaces the standard rate limit checks for batch submissions."
        },
    )


def get_model_rate_limit_from_metadata(
    user_api_key_dict: UserAPIKeyAuth,
    metadata_accessor_key: Literal["team_metadata", "organization_metadata", "project_metadata"],
    rate_limit_key: Literal["model_rpm_limit", "model_tpm_limit", "model_itpm_limit", "model_otpm_limit"],
) -> dict[str, int] | None:
    if getattr(user_api_key_dict, metadata_accessor_key):
        return getattr(user_api_key_dict, metadata_accessor_key).get(rate_limit_key)
    return None


def get_team_model_rpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    if user_api_key_dict.team_metadata:
        return user_api_key_dict.team_metadata.get("model_rpm_limit")
    return None


def get_team_model_tpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    if user_api_key_dict.team_metadata:
        return user_api_key_dict.team_metadata.get("model_tpm_limit")
    return None


def get_key_mcp_rpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    """
    Get the per-MCP-server rpm limit for a given api key.

    Priority order (returns first found):
    1. Key metadata (mcp_rpm_limit)
    2. Team metadata (mcp_rpm_limit)

    The returned dict is keyed by MCP server name (alias if set, else the
    configured server name).
    """
    if user_api_key_dict.metadata:
        result: Final = user_api_key_dict.metadata.get("mcp_rpm_limit")
        if result is not None:
            return result

    if user_api_key_dict.team_metadata:
        team_limit: Final = user_api_key_dict.team_metadata.get("mcp_rpm_limit")
        if team_limit is not None:
            return team_limit

    return None


def get_team_mcp_rpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    if user_api_key_dict.team_metadata:
        return user_api_key_dict.team_metadata.get("mcp_rpm_limit")
    return None


def get_key_tag_rpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    """
    Get the per-request-tag rpm limit configured on a given api key.

    The returned dict is keyed by request tag, so each tag/group tracked on
    the key gets its own independent RPM counter.
    """
    if user_api_key_dict.metadata:
        return user_api_key_dict.metadata.get("tag_rpm_limit")
    return None


def get_project_model_rpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    if user_api_key_dict.project_metadata:
        return user_api_key_dict.project_metadata.get("model_rpm_limit")
    return None


def get_project_model_tpm_limit(
    user_api_key_dict: UserAPIKeyAuth,
) -> dict[str, int] | None:
    if user_api_key_dict.project_metadata:
        return user_api_key_dict.project_metadata.get("model_tpm_limit")
    return None


def custom_auth_common_checks_warning(
    *,
    custom_auth_configured: bool,
    run_common_checks: bool,
) -> str | None:
    if not custom_auth_configured or run_common_checks:
        return None
    return (
        "custom_auth is configured but 'custom_auth_run_common_checks' is not set. "
        "Problem: budgets, model-access allowlists, and per-model rate limits configured "
        "on your DB team/project records will NOT be enforced for custom-auth requests "
        "(rate limits set directly on the returned UserAPIKeyAuth still apply). "
        "Fix: set 'general_settings.custom_auth_run_common_checks: true'. "
        "Docs: https://docs.litellm.ai/docs/proxy/custom_auth"
    )


_custom_auth_common_checks_warning_emitted = False


def warn_once_if_custom_auth_skips_common_checks(
    *,
    custom_auth_configured: bool,
    run_common_checks: bool,
    logger: Logger = verbose_proxy_logger,
) -> None:
    global _custom_auth_common_checks_warning_emitted
    if _custom_auth_common_checks_warning_emitted:
        return
    message: Final = custom_auth_common_checks_warning(
        custom_auth_configured=custom_auth_configured,
        run_common_checks=run_common_checks,
    )
    if message is None:
        return
    logger.warning(message)
    _custom_auth_common_checks_warning_emitted = True


def log_once_if_budget_reservation_disabled(
    *,
    disabled: bool,
    logger: Logger = verbose_proxy_logger,
) -> None:
    if constants.budget_reservation_disabled_info_emitted or not disabled:
        return
    logger.info(
        "disable_budget_reservation is enabled: skipping optimistic budget "
        "reservation. Budget enforcement is read-time only. Concurrent "
        "requests can each pass the spend check before their cost is recorded, "
        "so a configured budget may be briefly exceeded under high concurrency. "
        "Set disable_budget_reservation to False or remove it to restore "
        "hard per-request budget enforcement."
    )
    constants.budget_reservation_disabled_info_emitted = True  # rebind-ok: process-wide one-shot sentinel


def is_pass_through_provider_route(route: str) -> bool:
    PROVIDER_SPECIFIC_PASS_THROUGH_ROUTES: Final = [
        "vertex-ai",
    ]

    # check if any of the prefixes are in the route
    for prefix in PROVIDER_SPECIFIC_PASS_THROUGH_ROUTES:
        if prefix in route:
            return True

    return False


def has_user_setup_sso() -> bool:
    """
    Check if the user has set up single sign-on (SSO).

    Covers OAuth providers (Microsoft, Google, generic) and SAML IdP metadata.
    Used by UI discovery (``sso_configured``) so the login button enables when
    any supported SSO path is configured — including SAML-only setups.
    """
    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)
    saml_idp_metadata_url: Final = os.getenv("SAML_IDP_METADATA_URL", None)
    saml_idp_metadata_xml: Final = os.getenv("SAML_IDP_METADATA_XML", None)

    return (
        microsoft_client_id is not None
        or google_client_id is not None
        or generic_client_id is not None
        or bool(saml_idp_metadata_url)
        or bool(saml_idp_metadata_xml)
    )


def _is_google_ready() -> bool:
    return bool(os.getenv("GOOGLE_CLIENT_ID")) and bool(os.getenv("GOOGLE_CLIENT_SECRET"))


def _is_microsoft_ready() -> bool:
    return (
        bool(os.getenv("MICROSOFT_CLIENT_ID"))
        and bool(os.getenv("MICROSOFT_CLIENT_SECRET"))
        and bool(os.getenv("MICROSOFT_TENANT"))
    )


def _is_generic_oauth_ready() -> bool:
    return (
        bool(os.getenv("GENERIC_CLIENT_ID"))
        and bool(os.getenv("GENERIC_CLIENT_SECRET"))
        and bool(os.getenv("GENERIC_AUTHORIZATION_ENDPOINT"))
        and bool(os.getenv("GENERIC_TOKEN_ENDPOINT"))
        and bool(os.getenv("GENERIC_USERINFO_ENDPOINT"))
    )


def _is_saml_ready() -> bool:
    if not (os.getenv("SAML_IDP_METADATA_URL") or os.getenv("SAML_IDP_METADATA_XML")):
        return False
    # SAML's runtime (python3-saml) is an optional dependency; the SAML
    # handler itself fails closed on every request when it is missing
    # (SAMLAuthHandler raises before touching the IdP), so metadata alone
    # is not "ready" either. find_spec raises ModuleNotFoundError (rather
    # than returning None) when the top-level package is absent entirely,
    # so this must not be a bare boolean expression or every password
    # login would 500 on a deployment that configured SAML metadata
    # without installing the optional extra.
    try:
        return importlib.util.find_spec("onelogin.saml2.auth") is not None
    except ModuleNotFoundError:
        return False


def is_sso_provider_fully_configured() -> bool:
    """Whether ANY configured SSO provider has every companion setting it
    needs to actually authenticate a user, not merely a client id.

    A lone ``MICROSOFT_CLIENT_ID`` with no secret or tenant makes
    ``has_user_setup_sso()`` return True while every real sign-in attempt
    fails, so a gate that BLOCKS the password fallback (unlike the UI
    discovery use of ``has_user_setup_sso()``, where a dead login button is
    merely confusing) must check readiness here, or it can lock every admin
    out with no way to sign in at all. Checks every provider independently
    (mirroring ``/sso/readiness``'s per-provider requirements) rather than
    stopping at the first one with a client id set, so a stray leftover
    client id for an unused provider can never mask a different, fully
    configured provider that would otherwise satisfy this gate.
    """
    return _is_google_ready() or _is_microsoft_ready() or _is_generic_oauth_ready() or _is_saml_ready()


def get_customer_user_header_from_mapping(user_id_mapping) -> list | None:
    """Return the header_name mapped to CUSTOMER role, if any (dict-based)."""
    if not user_id_mapping:
        return None
    items: Final = user_id_mapping if isinstance(user_id_mapping, list) else [user_id_mapping]
    customer_headers_mappings: Final = []
    for item in items:
        if not isinstance(item, dict):
            continue
        role = item.get("litellm_user_role")
        header_name = item.get("header_name")
        if role is None or not header_name:
            continue
        if str(role).lower() == str(LitellmUserRoles.CUSTOMER).lower():
            customer_headers_mappings.append(header_name.lower())

    if customer_headers_mappings:
        return customer_headers_mappings

    return None


def _get_customer_id_from_standard_headers(
    request_headers: dict | None,
) -> str | None:
    """
    Check standard customer ID headers for a customer/end-user ID.

    This enables tools like Claude Code to pass customer IDs via ANTHROPIC_CUSTOM_HEADERS.
    No configuration required - these headers are always checked.

    Args:
        request_headers: The request headers dict

    Returns:
        The customer ID if found in standard headers, None otherwise
    """
    if request_headers is None:
        return None

    for standard_header in STANDARD_CUSTOMER_ID_HEADERS:
        for header_name, header_value in request_headers.items():
            if header_name.lower() == standard_header.lower():
                user_id_str = _coerce_user_id_to_str(header_value)
                if user_id_str:
                    return user_id_str
    return None


def _coerce_user_id_to_str(value: Any) -> str | None:
    """Return a usable end-user identifier string, or None if the value isn't one.

    Always drops non-string structured values (dict/list/tuple/set) because
    stringifying them produces garbage spend-log rows like
    ``"{'device_id': ...}"``. Strings that *decode* to a structured payload
    are only rejected when ``litellm.validate_end_user_id_in_db`` is enabled
    — operators who currently pass JSON-encoded identifiers keep their
    existing behavior until they opt in. See
    auth_utils.py:get_end_user_id_from_request_body for the extraction chain.
    """
    if value is None:
        return None
    if isinstance(value, bool):
        # bool is an int subclass; handle explicitly to avoid "True"/"False".
        return None
    if isinstance(value, (int, float)):
        return str(value)
    if isinstance(value, str):
        stripped: Final = value.strip()
        if not stripped:
            return None
        # Reject strings that decode to a structured payload (JSON object/array)
        # only when the operator has opted into end-user validation. Gating
        # behind the flag preserves backwards compatibility for deployments
        # that intentionally pass JSON-encoded user identifiers.
        if litellm.validate_end_user_id_in_db and stripped[:1] in ("{", "["):
            parsed: Final = safe_json_loads(stripped)
            if isinstance(parsed, (dict, list)):
                return None
        return stripped
    # dict, list, tuple, set, arbitrary objects -> drop.
    return None


def get_end_user_id_from_request_body(request_body: dict, request_headers: dict | None = None) -> str | None:
    # Import general_settings here to avoid potential circular import issues at module level
    # and to ensure it's fetched at runtime.
    from litellm.proxy.proxy_server import general_settings

    # Check 1: Standard customer ID headers (always checked, no configuration required)
    customer_id: Final = _get_customer_id_from_standard_headers(request_headers=request_headers)
    if customer_id is not None:
        return customer_id

    # Check 2: Follow the user header mappings feature, if not found, then check for deprecated user_header_name (only if request_headers is provided)
    # User query: "system not respecting user_header_name property"
    # This implies the key in general_settings is 'user_header_name'.
    if request_headers is not None:
        custom_header_name_to_check: list | str | None = None

        # Prefer user mappings (new behavior)
        user_id_mapping: Final = general_settings.get("user_header_mappings", None)
        if user_id_mapping:
            custom_header_name_to_check = get_customer_user_header_from_mapping(user_id_mapping)

        # Fallback to deprecated user_header_name if mapping did not specify
        if not custom_header_name_to_check:
            user_id_header_config_key: Final = "user_header_name"
            value: Final = general_settings.get(user_id_header_config_key)
            if isinstance(value, str) and value.strip() != "":
                custom_header_name_to_check = value

        # If we have a header name to check, try to read it from request headers
        if isinstance(custom_header_name_to_check, list):
            headers_lower: Final = {k.lower(): v for k, v in request_headers.items()}
            for expected_header in custom_header_name_to_check:
                user_id_str = _coerce_user_id_to_str(headers_lower.get(expected_header))
                if user_id_str:
                    return user_id_str

        elif isinstance(custom_header_name_to_check, str):
            for header_name, header_value in request_headers.items():
                if header_name.lower() == custom_header_name_to_check.lower():
                    user_id_str = _coerce_user_id_to_str(header_value)
                    if user_id_str:
                        return user_id_str

    # Check 3: 'user' field in request_body (commonly OpenAI)
    if "user" in request_body:
        user_id_str = _coerce_user_id_to_str(request_body["user"])
        if user_id_str:
            return user_id_str

    def _as_dict(value: Any) -> dict:
        # metadata / litellm_metadata can arrive as JSON strings from
        # multipart/form-data or extra_body; coerce so string-encoded
        # payloads can't evade end-user attribution.
        if isinstance(value, dict):
            return value
        if isinstance(value, str):
            parsed: Final = safe_json_loads(value)
            return parsed if isinstance(parsed, dict) else {}
        return {}

    # Check 4: 'litellm_metadata.user' in request_body (commonly Anthropic)
    litellm_metadata: Final = _as_dict(request_body.get("litellm_metadata"))
    user_id_str = _coerce_user_id_to_str(litellm_metadata.get("user"))
    if user_id_str:
        return user_id_str

    # Check 5: 'metadata.user_id' in request_body (another common pattern)
    metadata_dict: Final = _as_dict(request_body.get("metadata"))
    user_id_str = _coerce_user_id_to_str(metadata_dict.get("user_id"))
    if user_id_str:
        return user_id_str

    # Check 6: 'safety_identifier' in request body (OpenAI Responses API parameter)
    # SECURITY NOTE: safety_identifier can be set by any caller in the request body.
    # Only use this for end-user identification in trusted environments where you control
    # the calling application. For untrusted callers, prefer using headers or server-side
    # middleware to set the end_user_id to prevent impersonation.
    user_id_str = _coerce_user_id_to_str(request_body.get("safety_identifier"))
    if user_id_str:
        return user_id_str

    return None


MODEL_ROUTING_HEADER_NAME: Final = "x-litellm-model"
_MODEL_ROUTING_ROUTE_MARKERS: Final = (
    "/files",
    "/batches",
    "/vector_stores",
    "/skills",
    "/evals",
    "/fine_tuning",
    "/videos",
)
_MODEL_ROUTING_HEADER_OR_QUERY_ROUTE_MARKERS: Final = (
    "/files",
    "/batches",
    "/skills",
    "/evals",
)
_MODEL_ROUTING_QUERY_TARGET_MODEL_ROUTE_MARKERS: Final = (
    "/files",
    "/batches",
    "/fine_tuning",
)
_MODEL_ROUTING_BODY_TARGET_MODEL_ROUTE_MARKERS: Final = (
    "/files",
    "/batches",
    "/vector_stores",
)
_MODEL_ROUTING_COMPLETION_MODEL_ROUTE_MARKERS: Final = ("/evals",)
# Realtime WebRTC routes carry the effective model inside the nested
# ``session.model`` field (see realtime_endpoints.endpoints), so the model the
# request will actually use is not present at the top level. Extract it here so
# can_key_call_model() validates the real target model.
_MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS: Final = (
    "/realtime/client_secrets",
    "/realtime/calls",
)
_MODEL_ROUTING_ID_FIELDS: Final = (
    "file_id",
    "input_file_id",
    "output_file_id",
    "error_file_id",
    "batch_id",
    "fine_tuning_job_id",
    "training_file",
    "validation_file",
    "vector_store_id",
    "video_id",
    "character_id",
)


def _append_model_candidates(candidates: list[str], value: Any) -> None:
    if value is None:
        return

    values: Final = value if isinstance(value, (list, tuple, set)) else [value]
    for item in values:
        if item is None:
            continue
        if isinstance(item, str):
            model_names = [model.strip() for model in item.split(",")]
        else:
            model_names = [str(item).strip()]
        candidates.extend(model for model in model_names if model)


def _dedupe_model_candidates(candidates: list[str]) -> list[str]:
    deduped: Final[list[str]] = []
    for model in candidates:
        if model not in deduped:
            deduped.append(model)
    return deduped


def _get_case_insensitive_mapping_value(mapping: Mapping[str, object] | None, key: str) -> object:
    if not mapping:
        return None
    if key in mapping:
        return mapping[key]
    key_lower: Final = key.lower()
    for mapping_key, value in mapping.items():
        if str(mapping_key).lower() == key_lower:
            return value
    return None


def _route_matches_any_marker(route: str, markers: tuple[str, ...]) -> bool:
    normalized_route: Final = route.lower()
    return any(marker in normalized_route for marker in markers)


def _route_uses_model_routing_sources(route: str) -> bool:
    return _route_matches_any_marker(route=route, markers=_MODEL_ROUTING_ROUTE_MARKERS)


def _extract_models_from_managed_resource_id(
    resource_id: Any,
    resource_id_field: str | None = None,
    llm_router: Router | None = None,
) -> list[str]:
    if not isinstance(resource_id, str) or not resource_id:
        return []

    candidates: Final[list[str]] = []

    try:
        from litellm.proxy.openai_files_endpoints.common_utils import (
            _is_base64_encoded_unified_file_id,
            decode_model_from_file_id,
            get_model_id_from_unified_batch_id,
            get_models_from_unified_file_id,
        )

        _append_model_candidates(candidates=candidates, value=decode_model_from_file_id(resource_id))
        unified_file_id: Final = _is_base64_encoded_unified_file_id(resource_id)
        if unified_file_id:
            _append_model_candidates(
                candidates=candidates,
                value=get_models_from_unified_file_id(unified_file_id),
            )
            _append_model_candidates(
                candidates=candidates,
                value=_resolve_model_id_with_router(get_model_id_from_unified_batch_id(unified_file_id), llm_router),
            )
    except Exception as e:
        verbose_proxy_logger.debug("Unable to extract model from managed file/batch ID: %s", str(e))

    try:
        from litellm.llms.base_llm.managed_resources.utils import parse_unified_id

        parsed_id: Final = parse_unified_id(resource_id)
        if parsed_id:
            _append_model_candidates(
                candidates=candidates,
                value=_resolve_model_id_with_router(parsed_id.get("model_id"), llm_router),
            )
            _append_model_candidates(candidates=candidates, value=parsed_id.get("target_model_names"))
    except Exception as e:
        verbose_proxy_logger.debug("Unable to extract model from unified managed resource ID: %s", str(e))

    if resource_id_field in ("video_id", "character_id"):
        try:
            from litellm.types.videos.utils import (
                decode_character_id_with_provider,
                decode_video_id_with_provider,
            )

            if resource_id_field == "video_id":
                model_id = decode_video_id_with_provider(resource_id).get("model_id")
                _append_model_candidates(
                    candidates=candidates,
                    value=_resolve_model_id_with_router(model_id, llm_router),
                )
            else:
                model_id = decode_character_id_with_provider(resource_id).get("model_id")
                _append_model_candidates(
                    candidates=candidates,
                    value=_resolve_model_id_with_router(model_id, llm_router),
                )
        except Exception as e:
            verbose_proxy_logger.debug("Unable to extract model from managed video/character ID: %s", str(e))

    return _dedupe_model_candidates(candidates)


def _resolve_model_id_with_router(model_id: str | None, llm_router: Router | None) -> str | None:
    if model_id is None or llm_router is None:
        return model_id
    try:
        return llm_router.resolve_model_name_from_model_id(model_id) or model_id
    except Exception as e:
        verbose_proxy_logger.debug("Unable to resolve model_id from managed resource ID: %s", str(e))
        return model_id


def _extract_model_candidates_from_request(
    request_data: dict,
    route: str,
    request_headers: Mapping[str, object] | None = None,
    request_query_params: Mapping[str, object] | None = None,
    llm_router: Router | None = None,
) -> list[str]:
    candidates: Final[list[str]] = []
    uses_model_routing_sources: Final = _route_uses_model_routing_sources(route=route)
    uses_header_or_query_model_sources: Final = _route_matches_any_marker(
        route=route, markers=_MODEL_ROUTING_HEADER_OR_QUERY_ROUTE_MARKERS
    )
    uses_query_target_model_sources: Final = _route_matches_any_marker(
        route=route, markers=_MODEL_ROUTING_QUERY_TARGET_MODEL_ROUTE_MARKERS
    )
    uses_body_target_model_sources: Final = _route_matches_any_marker(
        route=route, markers=_MODEL_ROUTING_BODY_TARGET_MODEL_ROUTE_MARKERS
    )
    uses_completion_model_sources: Final = _route_matches_any_marker(
        route=route, markers=_MODEL_ROUTING_COMPLETION_MODEL_ROUTE_MARKERS
    )

    body_model: Final = request_data.get("model")
    _append_model_candidates(candidates, body_model)
    if uses_body_target_model_sources or not body_model:
        _append_model_candidates(candidates, request_data.get("target_model_names"))
    if _route_matches_any_marker(route=route, markers=_MODEL_ROUTING_SESSION_MODEL_ROUTE_MARKERS):
        session: Final = request_data.get("session")
        if isinstance(session, dict):
            _append_model_candidates(candidates, session.get("model"))
    if uses_completion_model_sources and isinstance(request_data.get("completion"), dict):
        _append_model_candidates(candidates, request_data["completion"].get("model"))

    if uses_model_routing_sources:
        if uses_header_or_query_model_sources:
            _append_model_candidates(
                candidates,
                _get_case_insensitive_mapping_value(request_query_params, "model"),
            )
            _append_model_candidates(
                candidates,
                _get_case_insensitive_mapping_value(request_headers, MODEL_ROUTING_HEADER_NAME),
            )
        if uses_query_target_model_sources:
            _append_model_candidates(
                candidates,
                _get_case_insensitive_mapping_value(request_query_params, "target_model_names"),
            )

        for field in _MODEL_ROUTING_ID_FIELDS:
            _append_model_candidates(
                candidates,
                _extract_models_from_managed_resource_id(
                    request_data.get(field),
                    resource_id_field=field,
                    llm_router=llm_router,
                ),
            )

    return _dedupe_model_candidates(candidates)


def _format_model_candidates(
    candidates: list[str],
) -> str | list[str] | None:
    if not candidates:
        return None
    if len(candidates) == 1:
        return candidates[0]
    return candidates


def request_dispatched_to_pass_through_endpoint(request: Request | None) -> bool:
    """Whether FastAPI resolved this request to a user-defined pass-through handler.

    Reads the marker set by ``create_pass_through_route`` off the dispatched endpoint
    (``request.scope["endpoint"]``). Because routing has already run by the time auth
    dependencies execute, this reflects the handler that actually serves the request:
    a custom path colliding with a built-in route resolves to the built-in handler,
    which carries no marker, so model-access checks are never wrongly skipped.
    """
    if request is None:
        return False
    scope: Final = getattr(request, "scope", None)
    if not isinstance(scope, dict):
        return False
    endpoint: Final = scope.get("endpoint")
    # Identity check against True (not truthiness): the marker is set to the literal
    # True, and this keeps a spec'd Mock request (whose attribute access yields truthy
    # child mocks) from being misread as a pass-through dispatch.
    return getattr(endpoint, LITELLM_PASS_THROUGH_ENDPOINT_MARKER, False) is True


def get_model_from_request(
    request_data: dict,
    route: str,
    request_headers: Mapping[str, object] | None = None,
    request_query_params: Mapping[str, object] | None = None,
    llm_router: Router | None = None,
    request: Request | None = None,
) -> str | list[str] | None:
    """Resolve the model(s) a request targets, for model-access and budget checks.

    Returns ``None`` when the request was dispatched to a user-defined pass-through
    endpoint: its body is forwarded verbatim to the configured upstream, so a
    ``model`` field there names an upstream model, not a LiteLLM-managed one, and
    enforcing key/team model allowlists against it would reject valid requests. The
    check reads the FastAPI-resolved endpoint (``request.scope["endpoint"]``), not the
    request path, so a custom path that collides with a built-in route never
    suppresses model-access checks: on a collision the built-in handler is dispatched
    and does not carry the marker. Built-in provider passthrough routes
    (``/vertex_ai``, ``/gemini``, ...) are separate handlers and keep model enforcement.
    """
    if request_dispatched_to_pass_through_endpoint(request):
        return None

    candidates: Final = _extract_model_candidates_from_request(
        request_data=request_data,
        route=route,
        request_headers=request_headers,
        request_query_params=request_query_params,
        llm_router=llm_router,
    )
    model = _format_model_candidates(candidates)

    # If no explicit model was found, try to extract from route
    if model is None:
        # Parse model from route that follows the pattern /openai/deployments/{model}/*
        match: Final = re.match(r"/openai/deployments/([^/]+)", route)
        if match:
            model = match.group(1)

    # If still not found, extract model from Google generateContent-style routes.
    # These routes put the model in the path and allow "/" inside the model id.
    # Examples:
    # - /v1beta/models/gemini-2.0-flash:generateContent
    # - /v1beta/models/bedrock/claude-sonnet-3.7:generateContent
    # - /models/custom/ns/model:streamGenerateContent
    if model is None and not route.lower().startswith("/vertex"):
        google_match = re.search(r"/(?:v1beta|beta)/models/([^:]+):", route)
        if google_match:
            model = google_match.group(1)

    if model is None and not route.lower().startswith("/vertex"):
        google_match = re.search(r"^/models/([^:]+):", route)
        if google_match:
            model = google_match.group(1)

    # If still not found, extract from Vertex AI passthrough route
    # Pattern: /vertex_ai/.../models/{model_id}:*
    # Example: /vertex_ai/v1/.../models/gemini-1.5-pro:generateContent
    if model is None and route.lower().startswith("/vertex"):
        vertex_match: Final = re.search(r"/models/([^:]+)", route)
        if vertex_match:
            model = vertex_match.group(1)

    if route.lower().startswith("/bedrock"):
        bedrock_model: Final = _model_from_bedrock_route(route)
        return model if bedrock_model is None else bedrock_model

    if route.lower().startswith(("/azure/", "/azure_ai/")):
        azure_model: Final = _router_model_from_azure_route(route, llm_router)
        return model if azure_model is None else azure_model

    return model


def _router_model_from_azure_route(route: str, llm_router: Router | None) -> str | None:
    if llm_router is None:
        return None
    endpoint: Final = re.sub(r"^/azure(?:_ai)?/", "", route, flags=re.IGNORECASE)
    return azure_router_model_in_endpoint(endpoint, frozenset(llm_router.get_model_names()))


def _model_from_bedrock_route(route: str) -> str | None:
    from litellm.proxy.pass_through_endpoints.llm_passthrough_endpoints import (
        _extract_model_from_bedrock_endpoint,
        is_bedrock_count_tokens_endpoint,
    )

    bedrock_endpoint: Final = re.sub(r"^/bedrock/", "", route, flags=re.IGNORECASE)
    if is_bedrock_count_tokens_endpoint(bedrock_endpoint):
        return None
    try:
        return _extract_model_from_bedrock_endpoint(bedrock_endpoint)
    except ValueError:
        return None


def abbreviate_api_key(api_key: str) -> str:
    if len(api_key) < MINIMUM_CUSTOM_KEY_LENGTH:
        return "sk-..."
    return f"sk-...{api_key[-4:]}"