chore(auth): tighten clientside api_base handling (#26518)

* chore(auth): validate clientside api_base against SSRF guard; clear admin secrets on base override

Two related issues with how the proxy handles client-supplied
``api_base`` / ``base_url`` overrides on chat-completion requests:

1. **SSRF gate bypass** — ``check_complete_credentials()`` returned
   ``True`` for any non-empty ``api_key``, allowing the
   ``is_request_body_safe`` ``banned_params`` loop to admit ``api_base``
   / ``base_url`` values that point at private (RFC 1918), loopback,
   link-local, or cloud-metadata addresses. Now: when the gate sees a
   client-supplied ``api_base`` / ``base_url``, it runs the URL through
   ``litellm_core_utils.url_utils.validate_url`` (DNS-resolves, blocks
   internal/IMDS/LL networks, defends against rebinding). Rejection
   raises with a clear message.

2. **Admin-config leak on base override** —
   ``get_dynamic_litellm_params`` only carried the three clientside keys
   (``api_key``, ``api_base``, ``base_url``) from request to upstream
   call. Other admin-configured fields on ``litellm_params`` —
   ``organization``, ``extra_body``, ``extra_headers``, ``api_version``,
   ``azure_ad_token``, AWS / Vertex creds, etc. — flowed through
   unchanged. With base redirected to a client-controlled server, those
   admin secrets were sent to the attacker. Now: when ``api_base`` /
   ``base_url`` is in ``request_kwargs``, drop those admin-config
   fields from ``litellm_params`` unless the caller re-supplied them.

Tests cover the SSRF-target rejection per URL field, the admin-secret
clearing on base override, the don't-clear case when only ``api_key``
is overridden (BYOK pattern), and the don't-overwrite case when the
caller resupplies fields like ``organization`` themselves.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* chore(vertex-batches): wrap api_base GET in safe_get for defense-in-depth

The vertex batches status-poll fetches an attacker-influenceable
``api_base`` URL with a raw ``sync_handler.get()``. The proxy auth gate
already validates clientside ``api_base`` before reaching this sink, so
the proxy flow is covered. This adds the per-sink wrap so SDK callers
and any future code path that bypasses the proxy gate pick up the same
SSRF defense from ``url_utils.safe_get``.

Operators with a legitimate private Vertex base can either allowlist
the host via ``litellm.user_url_allowed_hosts`` or disable validation
with ``litellm.user_url_validation = False``.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* refactor(auth): hoist url_utils import; derive admin-config field list from CredentialLiteLLMParams

/simplify pass:
- Move ``from litellm.litellm_core_utils.url_utils import SSRFError, validate_url``
  to module top in ``proxy/auth/auth_utils.py``. CLAUDE.md prefers
  module-level imports unless avoiding a circular dependency, and
  there's no cycle here (``url_utils`` doesn't depend on ``proxy.auth``).
- Replace the hardcoded ``_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE``
  literal with ``_admin_config_fields_to_clear_on_base_override()`` that
  derives the typed-field portion from
  ``CredentialLiteLLMParams.model_fields``. Adds three fields the
  hardcoded list missed (``aws_bedrock_runtime_endpoint``,
  ``watsonx_region_name``, ``region_name``) and stays in sync as new
  provider fields are declared on the model. The kwargs-only set
  (``organization``, ``extra_body``, ``azure_ad_token``, ``aws_session_token``,
  ``aws_sts_endpoint``, ``aws_web_identity_token``, ``aws_role_name``, …)
  remains explicit since those fields aren't on the typed model.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(auth): close field-echo bypass; gate URL check on toggle; cover async batch path

Three issues from review:

1. ``get_dynamic_litellm_params`` used ``if field not in request_kwargs:
   pop`` to clear admin-set provider config when the caller redirected
   ``api_base``. A caller could *echo* any clear-list field name (with any
   value, including an empty string) to skip the pop, leaving the admin's
   value in ``litellm_params`` to be forwarded to the redirected upstream.
   Fix: always pop, then write the caller's value back if they resupplied
   the field.

2. ``check_complete_credentials`` called ``validate_url`` directly. That
   helper doesn't itself consult ``litellm.user_url_validation``; the
   toggle is honoured by ``safe_get`` / ``async_safe_get``. Mirror that
   here so admins who explicitly disabled URL validation aren't blocked
   at the proxy boundary.

3. ``VertexAIBatchesHandler._async_retrieve_batch`` still used a bare
   ``await client.get(api_base, ...)`` while the sync sibling was wrapped
   in ``safe_get``. Wrap the async call in ``async_safe_get`` so SDK
   callers on the async path get the same DNS-rebind / private /
   cloud-metadata defenses as the sync path.

Tests:

- ``TestCheckCompleteCredentialsBlocksSSRF`` is now mock-only; an autouse
  fixture flips the toggle on, ``validate_url`` is patched in the
  parametrized blocking tests, and the positive path no longer makes a
  real DNS call to api.openai.com.
- ``test_skips_url_validation_when_toggle_is_off`` documents the new
  toggle-off behaviour and asserts ``validate_url`` is not called.
- ``test_caller_resupplied_value_overrides_admin_value_on_base_override``
  replaces the prior test that asserted the buggy
  preserve-admin-value-on-echo behaviour.
- ``test_field_echo_does_not_preserve_admin_value`` is a focused
  regression test for the empty-string echo vector.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(auth): close provider-confusion credential exfil; expand banned-params; cover OCI

Three additions on top of the entry-point URL gate so the cluster is
fully closed against caller-supplied ``api_base`` redirection:

1. ``get_llm_provider_logic.py`` matched registered openai-compatible
   endpoints against ``api_base`` with an unanchored substring search
   (``if endpoint in api_base:``). A caller could pass an api_base like
   ``https://attacker.com/api.groq.com/openai/v1`` to coerce the proxy
   into reading ``GROQ_API_KEY`` from the environment and forwarding it
   as a Bearer credential to the attacker's host. Replaced with parsed-
   URL semantics (hostname exact-match plus segment-bounded path-prefix)
   in a new ``_endpoint_matches_api_base`` helper.

2. ``is_request_body_safe`` rejects ``api_base`` / ``base_url`` /
   ``user_config`` / a handful of AWS / vertex fields, but the list
   omitted three other endpoint-targeting fields:
   * ``aws_bedrock_runtime_endpoint`` — Bedrock endpoint redirect
   * ``langsmith_base_url`` / ``langfuse_host`` — observability callback
     hostnames; attacker-controlled values exfiltrate the entire request
     payload (incl. message content) via the logging hook.
   Added all three to the blocklist.

3. ``_admin_config_fields_to_clear_on_base_override`` derives its typed-
   field list from ``CredentialLiteLLMParams.model_fields``, which does
   not declare any of the OCI provider's auth fields. Added
   ``oci_signer``, ``oci_user``, ``oci_fingerprint``, ``oci_tenancy``,
   ``oci_key``, and ``oci_key_file`` to the kwargs-only fixed list so
   they are cleared on caller-redirected ``api_base`` like the AWS /
   Azure / Vertex equivalents.

Tests:

- ``TestEndpointMatchesApiBase`` — direct unit tests on the new
  matcher: legitimate provider URLs (5 shapes) match; attacker
  smuggling via path injection, suffix label, prefix label, userinfo
  ``@`` injection, and path-segment lookalikes (7 shapes) do not.
- ``TestGetLlmProviderRejectsAttackerSmuggledApiBase`` — end-to-end
  invariant that ``GROQ_API_KEY`` is never read against an attacker-
  controlled host while the legitimate ``api.groq.com`` path still
  resolves the provider correctly.
- ``TestIsRequestBodySafeBlocksEndpointTargetingFields`` — parametrized
  coverage that each of the three new banned-params raises a clear
  rejection naming the offending field.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(auth): remove implicit api-key bypass + add posthog/braintrust/slack to blocklist

The historical ``check_complete_credentials`` clause inside
``is_request_body_safe`` was a third, *implicit*, *caller-controlled*
BYOK path: any caller that supplied a non-empty ``api_key`` caused the
entire banned-params blocklist to be skipped. That turned every missing
entry on the blocklist into an exploitable SSRF / credential-exfil hole
and is the root cause of the chain of api_base advisories that have
been re-discovered with each new integration:

* GHSA-jh89-88fc-qrfp (critical, triage) — env-var exfil via api_base
* GHSA-3frq-6r6h-7j64 (high, triage) — admin org / extra_body leak
* veria-admin Dv_m860l, b_yRJeQ5, stN90yjP, LBlyOAc8, U2TD78kg —
  variations on "list X is missing field Y"

Two explicit, admin-controlled BYOK paths already exist and remain:
``general_settings.allow_client_side_credentials = true`` (proxy-wide)
and ``configurable_clientside_auth_params: [...]`` per deployment.
Removing the implicit bypass converts the failure mode of a missing
blocklist entry from "live credential leak" to "predictable 400 with
a clear remediation message," which is the structural fix.

Also adds the three remaining endpoint-targeting fields the dynamic
callback layer reads from request body: ``posthog_host``,
``braintrust_host``, ``slack_webhook_url``. ``slack_webhook_url`` in
particular was a direct exfil channel (caller-set webhook → proxy
mirrors every request to attacker's Slack).

Tests:

- ``test_api_key_does_not_bypass_blocklist`` — parametrized regression
  asserting api_key=anything no longer skips the gate for any of the
  five highest-risk fields.
- ``test_admin_opt_in_proxy_wide_still_allows`` — confirms the
  documented BYOK opt-in still works.
- Extends ``test_endpoint_targeting_field_in_request_body_is_rejected``
  to cover posthog / braintrust / slack.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

* fix(auth): block sagemaker_base_url, s3_endpoint_url, deployment_url

Provider-specific endpoint overrides surfaced by a wider audit of
``optional_params`` consumers in ``litellm/llms/``. Same threat as
``api_base``: a caller-supplied value redirects the outbound request
to an attacker host.

* ``s3_endpoint_url`` — read in ``litellm/llms/bedrock/files/transformation.py``
  to build the S3 upload URL for Bedrock files. Caller redirects file
  uploads to attacker-controlled S3.
* ``sagemaker_base_url`` — read in ``litellm/llms/sagemaker/{chat,completion}/*``.
  Caller redirects SageMaker traffic. This is the primary vector
  described in veria-admin mNqEBBtG.
* ``deployment_url`` — popped in ``litellm/llms/sap/chat/transformation.py``.
  Caller redirects SAP deployment requests.

Tests parametrize ``test_endpoint_targeting_field_in_request_body_is_rejected``
to cover the three new fields.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
stuxf 2026-04-29 17:27:22 -07:00 committed by GitHub
parent 4a7af1ff68
commit dedaf74a5e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
6 changed files with 629 additions and 15 deletions

View File

@ -1,4 +1,5 @@
from typing import Optional, Tuple from typing import Optional, Tuple
from urllib.parse import urlparse
import litellm import litellm
from litellm.constants import REPLICATE_MODEL_NAME_WITH_ID_LENGTH from litellm.constants import REPLICATE_MODEL_NAME_WITH_ID_LENGTH
@ -8,6 +9,43 @@ from litellm.secret_managers.main import get_secret, get_secret_str
from ..types.router import LiteLLM_Params from ..types.router import LiteLLM_Params
def _endpoint_matches_api_base(endpoint: str, api_base: str) -> bool:
"""
Match a registered openai-compatible endpoint against a caller-supplied
``api_base`` using parsed-URL semantics, not unanchored substring search.
Both inputs may be a bare hostname (``api.perplexity.ai``), host+path
(``api.deepinfra.com/v1/openai``), or a full URL
(``https://api.cerebras.ai/v1``). Hostnames must match exactly
(case-insensitive); if the registered endpoint has a non-trivial path,
the api_base path must start with it on a segment boundary.
The naive ``endpoint in api_base`` shape lets a caller pass
``https://attacker.com/api.groq.com/openai/v1`` to coerce the proxy
into reading the server's GROQ_API_KEY from the environment and
forwarding it to the attacker's host as a Bearer credential.
"""
def _parse(value: str):
# Ensure urlparse sees a scheme so it populates hostname / path.
normalized = value if "://" in value else f"https://{value}"
return urlparse(normalized)
parsed_endpoint = _parse(endpoint)
parsed_url = _parse(api_base)
endpoint_host = (parsed_endpoint.hostname or "").lower()
url_host = (parsed_url.hostname or "").lower()
if not endpoint_host or endpoint_host != url_host:
return False
endpoint_path = parsed_endpoint.path.rstrip("/")
if not endpoint_path:
return True
url_path = parsed_url.path.rstrip("/")
return url_path == endpoint_path or url_path.startswith(endpoint_path + "/")
def _is_non_openai_azure_model(model: str) -> bool: def _is_non_openai_azure_model(model: str) -> bool:
try: try:
model_name = model.split("/", 1)[1] model_name = model.split("/", 1)[1]
@ -210,7 +248,7 @@ def get_llm_provider( # noqa: PLR0915
# check if api base is a known openai compatible endpoint # check if api base is a known openai compatible endpoint
if api_base: if api_base:
for endpoint in litellm.openai_compatible_endpoints: for endpoint in litellm.openai_compatible_endpoints:
if endpoint in api_base: if _endpoint_matches_api_base(endpoint, api_base):
if endpoint == "api.perplexity.ai": if endpoint == "api.perplexity.ai":
custom_llm_provider = "perplexity" custom_llm_provider = "perplexity"
dynamic_api_key = get_secret_str("PERPLEXITYAI_API_KEY") dynamic_api_key = get_secret_str("PERPLEXITYAI_API_KEY")

View File

@ -4,6 +4,7 @@ from typing import Any, Coroutine, Dict, Optional, Union
import httpx import httpx
import litellm import litellm
from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
from litellm.llms.custom_httpx.http_handler import ( from litellm.llms.custom_httpx.http_handler import (
_get_httpx_client, _get_httpx_client,
get_async_httpx_client, get_async_httpx_client,
@ -224,8 +225,14 @@ class VertexAIBatchPrediction(VertexLLM):
}, },
) )
response = sync_handler.get( # ``api_base`` here can come from caller-supplied request kwargs
url=api_base, # (clientside override). Wrap the fetch in ``safe_get`` so DNS
# rebind / private / cloud-metadata targets are rejected; the
# proxy auth gate already blocks malicious clientside ``api_base``
# at the boundary — this is defense-in-depth for SDK callers.
response = safe_get(
sync_handler,
api_base,
headers=headers, headers=headers,
) )
@ -270,8 +277,13 @@ class VertexAIBatchPrediction(VertexLLM):
}, },
) )
response = await client.get( # Mirror the sync path: ``api_base`` may come from caller-supplied
url=api_base, # request kwargs, so wrap the fetch in ``async_safe_get`` to reject
# DNS-rebind / private / cloud-metadata targets. Defense-in-depth
# behind the proxy auth gate's clientside ``api_base`` check.
response = await async_safe_get(
client,
api_base,
headers=headers, headers=headers,
) )
if response.status_code != 200: if response.status_code != 200:

View File

@ -6,9 +6,11 @@ from typing import Any, List, Optional, Tuple
from fastapi import HTTPException, Request, status from fastapi import HTTPException, Request, status
import litellm
from litellm import Router, provider_list from litellm import Router, provider_list
from litellm._logging import verbose_proxy_logger from litellm._logging import verbose_proxy_logger
from litellm.constants import STANDARD_CUSTOMER_ID_HEADERS from litellm.constants import STANDARD_CUSTOMER_ID_HEADERS
from litellm.litellm_core_utils.url_utils import SSRFError, validate_url
from litellm.proxy._types import * from litellm.proxy._types import *
from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS from litellm.types.router import CONFIGURABLE_CLIENTSIDE_AUTH_PARAMS
@ -53,6 +55,12 @@ def _check_valid_ip(
def check_complete_credentials(request_body: dict) -> bool: def check_complete_credentials(request_body: dict) -> bool:
""" """
if 'api_base' in request body. Check if complete credentials given. Prevent malicious attacks. 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: Optional[str] = None given_model: Optional[str] = None
@ -70,10 +78,27 @@ def check_complete_credentials(request_body: dict) -> bool:
return False return False
api_key_value = request_body.get("api_key") api_key_value = request_body.get("api_key")
if api_key_value and isinstance(api_key_value, str) and api_key_value.strip(): if not (api_key_value and isinstance(api_key_value, str) and api_key_value.strip()):
return True return False
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} "
f"is rejected by the SSRF guard ({e})."
)
return True
def check_regex_or_str_match(request_body_value: Any, regex_str: str) -> bool: def check_regex_or_str_match(request_body_value: Any, regex_str: str) -> bool:
@ -159,15 +184,42 @@ def is_request_body_safe(
"aws_web_identity_token", "aws_web_identity_token",
"aws_role_name", "aws_role_name",
"vertex_credentials", "vertex_credentials",
# 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",
"langsmith_base_url",
"langfuse_host",
"posthog_host",
"braintrust_host",
"slack_webhook_url",
# 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",
] ]
# 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.
for param in banned_params: for param in banned_params:
if ( if param in request_body:
param in request_body
and not check_complete_credentials( # allow client-credentials to be passed to proxy
request_body=request_body
)
):
if general_settings.get("allow_client_side_credentials") is True: if general_settings.get("allow_client_side_credentials") is True:
return True return True
elif ( elif (
@ -182,7 +234,10 @@ def is_request_body_safe(
return True return True
raise ValueError( raise ValueError(
f"Rejected Request: {param} is not allowed in request body. " f"Rejected Request: {param} is not allowed in request body. "
"Enable with `general_settings::allow_client_side_credentials` on proxy config.yaml. " "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", "Relevant Issue: https://huntr.com/bounties/4001e1a2-7b7a-4776-a3ae-e6692ec3d997",
) )

View File

@ -11,9 +11,60 @@ If given, generate a unique model_id for the deployment.
Ensures cooldowns are applied correctly. Ensures cooldowns are applied correctly.
""" """
from typing import List
clientside_credential_keys = ["api_key", "api_base", "base_url"] clientside_credential_keys = ["api_key", "api_base", "base_url"]
def _admin_config_fields_to_clear_on_base_override() -> List[str]:
"""
Provider-specific credential / endpoint-targeting fields that must NOT
flow through to a client-redirected upstream.
Built dynamically from ``CredentialLiteLLMParams.model_fields`` so any
new provider field added there (Bedrock endpoint, Watsonx region, etc.)
is gated automatically plus a fixed list of kwargs-only fields that
aren't declared on the typed model.
"""
from litellm.types.router import CredentialLiteLLMParams
typed_fields = [
f
for f in CredentialLiteLLMParams.model_fields
if f not in clientside_credential_keys
]
kwargs_only_fields = [
# Caller-supplied via **kwargs, not declared on CredentialLiteLLMParams.
"organization",
"extra_body",
"extra_headers",
"default_headers",
"api_type",
"azure_ad_token",
"azure_ad_token_provider",
"aws_session_token",
"aws_sts_endpoint",
"aws_web_identity_token",
"aws_role_name",
# OCI provider — consumed by litellm/llms/oci/* via optional_params
# and not declared on CredentialLiteLLMParams. Without these here,
# an admin's OCI signing key / tenancy / fingerprint would flow
# through to an attacker-redirected upstream.
"oci_signer",
"oci_user",
"oci_fingerprint",
"oci_tenancy",
"oci_key",
"oci_key_file",
]
return typed_fields + kwargs_only_fields
_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE = (
_admin_config_fields_to_clear_on_base_override()
)
def is_clientside_credential(request_kwargs: dict) -> bool: def is_clientside_credential(request_kwargs: dict) -> bool:
""" """
Check if the credential is a clientside credential. Check if the credential is a clientside credential.
@ -34,4 +85,20 @@ def get_dynamic_litellm_params(litellm_params: dict, request_kwargs: dict) -> di
for key in clientside_credential_keys: for key in clientside_credential_keys:
if key in request_kwargs: if key in request_kwargs:
litellm_params[key] = request_kwargs[key] litellm_params[key] = request_kwargs[key]
# If the caller redirected api_base/base_url to a client-controlled value,
# don't forward the admin's organization / extra_body / region / token /
# vertex / aws fields — those were meant for the original upstream.
# Always drop the admin's value first, then write the caller's value back
# if they resupplied the field. The naive
# ``if field not in request_kwargs: pop`` shape lets a caller *echo* a
# field name (with any value, including an empty string) to keep the
# admin's value in ``litellm_params`` and have it forwarded to the
# redirected upstream.
if "api_base" in request_kwargs or "base_url" in request_kwargs:
for field in _ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE:
litellm_params.pop(field, None)
if field in request_kwargs:
litellm_params[field] = request_kwargs[field]
return litellm_params return litellm_params

View File

@ -0,0 +1,138 @@
"""
Regression tests for the parsed-URL hostname match used to identify a
caller-supplied ``api_base`` as a known openai-compatible provider.
The previous shape (``if endpoint in api_base:``) used unanchored
substring search, which let a caller pass
``https://attacker.com/api.groq.com/openai/v1`` and have the proxy
return ``GROQ_API_KEY`` as the dynamic credential exfiltrating the
server's real provider key to an attacker-controlled host on the
outbound request.
"""
import os
import sys
from unittest.mock import patch
import pytest
sys.path.insert(0, os.path.abspath("../../.."))
from litellm.litellm_core_utils.get_llm_provider_logic import (
_endpoint_matches_api_base,
get_llm_provider,
)
class TestEndpointMatchesApiBase:
"""Direct unit tests on the parsed-URL matcher."""
@pytest.mark.parametrize(
"endpoint, api_base",
[
# Bare hostname endpoint, exact host match.
("api.perplexity.ai", "https://api.perplexity.ai/v1"),
# Endpoint includes a path; api_base path starts with it.
("api.groq.com/openai/v1", "https://api.groq.com/openai/v1"),
# Endpoint with full URL scheme.
("https://api.cerebras.ai/v1", "https://api.cerebras.ai/v1/chat"),
# Trailing-slash on registered endpoint must not break match.
("https://llm.chutes.ai/v1/", "https://llm.chutes.ai/v1/chat"),
# Case-insensitive on hostname.
("api.groq.com/openai/v1", "https://API.GROQ.COM/openai/v1"),
],
)
def test_legitimate_provider_urls_match(self, endpoint, api_base):
assert _endpoint_matches_api_base(endpoint, api_base) is True
@pytest.mark.parametrize(
"endpoint, api_base",
[
# Attacker host, registered endpoint smuggled into path.
(
"api.groq.com/openai/v1",
"https://attacker.com/api.groq.com/openai/v1",
),
# Attacker host, registered endpoint smuggled into a path segment.
(
"api.groq.com/openai/v1",
"https://attacker.com/foo/api.groq.com/openai/v1",
),
# Lookalike host that contains the registered host as a suffix label.
(
"api.groq.com/openai/v1",
"https://api.groq.com.attacker.com/openai/v1",
),
# Lookalike host with the registered host as a prefix.
(
"api.groq.com/openai/v1",
"https://api.groq.com.evil.example/openai/v1",
),
# Right host, wrong path — endpoint requires ``/openai/v1`` prefix.
("api.groq.com/openai/v1", "https://api.groq.com/v1"),
# Path-segment lookalike: ``/openai/v10`` must not match ``/openai/v1``.
("api.groq.com/openai/v1", "https://api.groq.com/openai/v10"),
# Userinfo / @-injection trick — the ``hostname`` after ``@`` is
# what httpx connects to.
(
"api.groq.com/openai/v1",
"https://api.groq.com@attacker.com/openai/v1",
),
],
)
def test_attacker_smuggling_does_not_match(self, endpoint, api_base):
assert _endpoint_matches_api_base(endpoint, api_base) is False
class TestGetLlmProviderRejectsAttackerSmuggledApiBase:
"""
End-to-end: ``get_llm_provider`` must NOT return the server's stored
secret (e.g. ``GROQ_API_KEY``) for an api_base whose hostname is
attacker-controlled, even when the registered endpoint string appears
elsewhere in the URL.
"""
def test_attacker_host_does_not_yield_groq_secret(self):
# The function may either fall through (different provider) or
# raise BadRequestError because the model can't be identified.
# The invariant under test is that ``GROQ_API_KEY`` is never
# looked up against an attacker-controlled hostname.
import litellm
with patch(
"litellm.litellm_core_utils.get_llm_provider_logic.get_secret_str",
return_value="server-real-groq-key",
) as mocked_secret:
try:
_, _, dynamic_api_key, _ = get_llm_provider(
model="some-model",
api_base="https://attacker.com/api.groq.com/openai/v1",
)
# If it returned, the dynamic key must not be the secret.
assert dynamic_api_key != "server-real-groq-key"
except litellm.exceptions.BadRequestError:
# Acceptable outcome: provider unidentifiable, no secret
# was returned.
pass
# Regardless of return / raise, the secret must never have been
# read against this attacker-controlled api_base.
groq_lookups = [
call
for call in mocked_secret.call_args_list
if call.args and call.args[0] == "GROQ_API_KEY"
]
assert groq_lookups == []
def test_legitimate_groq_api_base_still_resolves(self):
with patch(
"litellm.litellm_core_utils.get_llm_provider_logic.get_secret_str",
return_value="server-real-groq-key",
):
_, provider, dynamic_api_key, _ = get_llm_provider(
model="some-model",
api_base="https://api.groq.com/openai/v1",
)
assert provider == "groq"
assert dynamic_api_key == "server-real-groq-key"

View File

@ -5,6 +5,8 @@ Unit tests for auth_utils functions related to rate limiting and customer ID ext
from typing import Optional from typing import Optional
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
import pytest
from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy._types import UserAPIKeyAuth
from litellm.proxy.auth.auth_utils import ( from litellm.proxy.auth.auth_utils import (
_get_customer_id_from_standard_headers, _get_customer_id_from_standard_headers,
@ -15,6 +17,7 @@ from litellm.proxy.auth.auth_utils import (
get_key_model_tpm_limit, get_key_model_tpm_limit,
get_project_model_rpm_limit, get_project_model_rpm_limit,
get_project_model_tpm_limit, get_project_model_tpm_limit,
is_request_body_safe,
) )
@ -660,3 +663,304 @@ class TestCheckCompleteCredentials:
def test_returns_true_when_api_key_is_valid(self): def test_returns_true_when_api_key_is_valid(self):
result = check_complete_credentials({"model": "gpt-4", "api_key": "sk-valid"}) result = check_complete_credentials({"model": "gpt-4", "api_key": "sk-valid"})
assert result is True assert result is True
class TestCheckCompleteCredentialsBlocksSSRF:
"""
Even with credentials supplied, ``api_base`` / ``base_url`` must not
point at private / internal / cloud-metadata addresses. Without this
the gate accepts ``api_key=anything`` plus a malicious target and the
proxy is used as an SSRF pivot.
The check only runs when ``litellm.user_url_validation`` is True, so
every test in this class flips the toggle. Tests stay mock-only no
real DNS is performed.
"""
@pytest.fixture(autouse=True)
def _enable_url_validation(self, monkeypatch):
import litellm
monkeypatch.setattr(litellm, "user_url_validation", True, raising=False)
@pytest.mark.parametrize(
"url_field",
["api_base", "base_url"],
)
@pytest.mark.parametrize(
"blocked_url",
[
"http://169.254.169.254/latest/meta-data/iam/security-credentials/",
"http://metadata.google.internal/computeMetadata/v1/",
"http://127.0.0.1:8080/admin",
"http://10.0.0.1/",
"http://192.168.1.1/",
],
)
def test_rejects_private_or_metadata_targets(self, url_field, blocked_url):
from litellm.litellm_core_utils.url_utils import SSRFError
with patch(
"litellm.proxy.auth.auth_utils.validate_url",
side_effect=SSRFError(f"blocked: {blocked_url}"),
):
with pytest.raises(ValueError) as exc_info:
check_complete_credentials(
{
"model": "gpt-4",
"api_key": "sk-some-clientside-key",
url_field: blocked_url,
}
)
assert url_field in str(exc_info.value)
assert "SSRF" in str(exc_info.value)
def test_allows_public_target_when_validate_url_passes(self):
# ``validate_url`` is mocked so no real DNS is performed.
with patch(
"litellm.proxy.auth.auth_utils.validate_url",
return_value=("https://api.openai.com/v1", "api.openai.com"),
):
result = check_complete_credentials(
{
"model": "gpt-4",
"api_key": "sk-some-clientside-key",
"api_base": "https://api.openai.com/v1",
}
)
assert result is True
def test_skips_url_validation_when_toggle_is_off(self, monkeypatch):
# Admins who disable ``user_url_validation`` (default) should not
# have requests rejected at the proxy boundary even if the URL
# would fail the SSRF guard.
import litellm
monkeypatch.setattr(litellm, "user_url_validation", False, raising=False)
with patch(
"litellm.proxy.auth.auth_utils.validate_url",
) as mocked:
result = check_complete_credentials(
{
"model": "gpt-4",
"api_key": "sk-some-clientside-key",
"api_base": "http://127.0.0.1:8080/admin",
}
)
assert result is True
mocked.assert_not_called()
class TestGetDynamicLitellmParamsClearsAdminConfigOnBaseOverride:
"""
When the caller redirects ``api_base`` / ``base_url`` to their own
server, admin-set fields like ``OpenAI-Organization``, ``extra_body``,
AWS / Vertex / Azure tokens, and per-deployment ``api_version`` must
NOT flow through to that destination.
"""
def test_clears_admin_organization_and_extra_body_on_base_override(self):
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
)
admin_params = {
"model": "gpt-4",
"api_key": "sk-admin-key",
"api_base": "https://admin.upstream/v1",
"organization": "org-admin-corp",
"extra_body": {"x-admin-secret": "super-secret"},
"api_version": "2026-04-01",
}
out = get_dynamic_litellm_params(
litellm_params=dict(admin_params),
request_kwargs={
"api_key": "sk-attacker",
"api_base": "https://attacker.example",
},
)
assert out["api_base"] == "https://attacker.example"
assert out["api_key"] == "sk-attacker"
assert "organization" not in out
assert "extra_body" not in out
assert "api_version" not in out
def test_clears_aws_and_vertex_secrets_on_base_override(self):
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
)
admin_params = {
"model": "bedrock/claude-3",
"aws_access_key_id": "AKIA-EXAMPLE",
"aws_secret_access_key": "secret-example",
"aws_session_token": "session-example",
"vertex_credentials": '{"private_key":"-----BEGIN..."}',
"vertex_project": "admin-gcp-project",
}
out = get_dynamic_litellm_params(
litellm_params=dict(admin_params),
request_kwargs={"base_url": "https://attacker.example"},
)
assert "aws_access_key_id" not in out
assert "aws_secret_access_key" not in out
assert "aws_session_token" not in out
assert "vertex_credentials" not in out
assert "vertex_project" not in out
def test_caller_resupplied_value_overrides_admin_value_on_base_override(self):
# When the caller redirects ``api_base`` and *also* supplies their
# own value for one of the admin fields (e.g. ``organization``),
# the caller's value must win — never the admin's. The naive
# ``if field not in request_kwargs: pop`` shape lets a caller echo
# the field name with any value (or empty string) to keep the
# admin's value forwarded, which is the exfiltration vector this
# test guards against.
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
)
out = get_dynamic_litellm_params(
litellm_params={
"api_base": "https://admin.upstream/v1",
"organization": "org-admin",
"extra_body": {"admin": "value"},
},
request_kwargs={
"api_base": "https://attacker.example",
"organization": "org-attacker",
"extra_body": {"attacker": "value"},
},
)
assert out["organization"] == "org-attacker"
assert out["extra_body"] == {"attacker": "value"}
def test_field_echo_does_not_preserve_admin_value(self):
# Regression: a caller that echoes an admin-config field name with
# an *empty* value (or any value) must not be able to keep the
# admin's value in ``litellm_params``.
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
)
out = get_dynamic_litellm_params(
litellm_params={
"api_base": "https://admin.upstream/v1",
"organization": "org-admin-secret",
"extra_body": {"x-admin-only": "secret"},
},
request_kwargs={
"api_base": "https://attacker.example",
"organization": "",
"extra_body": "",
},
)
assert out["organization"] == ""
assert out["extra_body"] == ""
assert "org-admin-secret" not in str(out)
def test_no_clearing_when_only_api_key_overridden(self):
from litellm.router_utils.clientside_credential_handler import (
get_dynamic_litellm_params,
)
# Caller only overrides api_key (BYOK pattern); admin's organization /
# extra_body / region still apply because the destination is unchanged.
out = get_dynamic_litellm_params(
litellm_params={
"api_base": "https://admin.upstream/v1",
"organization": "org-admin",
"api_version": "2026-04-01",
},
request_kwargs={"api_key": "sk-byok"},
)
assert out["organization"] == "org-admin"
assert out["api_version"] == "2026-04-01"
assert out["api_base"] == "https://admin.upstream/v1"
class TestIsRequestBodySafeBlocksEndpointTargetingFields:
"""
``is_request_body_safe`` rejects request-body fields that retarget the
outbound request to a caller-controlled host. Beyond the original
``api_base`` / ``base_url``, the same protection must apply to:
* ``aws_bedrock_runtime_endpoint`` Bedrock endpoint redirect; an
attacker-controlled value coerces the proxy to authenticate against
their host with the admin's AWS creds.
* ``langsmith_base_url`` Langsmith callback host; attacker-controlled
values exfiltrate the entire request payload (incl. message content)
via the observability hook.
* ``langfuse_host`` same exfil vector via the Langfuse hook.
"""
@pytest.fixture(autouse=True)
def _disable_url_validation(self, monkeypatch):
# The new banned-params entries should be rejected even when
# ``user_url_validation`` is off — the gate isn't the URL guard,
# it's the banned-params list.
import litellm
monkeypatch.setattr(litellm, "user_url_validation", False, raising=False)
@pytest.mark.parametrize(
"field",
[
"aws_bedrock_runtime_endpoint",
"langsmith_base_url",
"langfuse_host",
"posthog_host",
"braintrust_host",
"slack_webhook_url",
"s3_endpoint_url",
"sagemaker_base_url",
"deployment_url",
],
)
def test_endpoint_targeting_field_in_request_body_is_rejected(self, field):
with pytest.raises(ValueError) as exc:
is_request_body_safe(
request_body={"model": "gpt-4", field: "https://attacker.example"},
general_settings={},
llm_router=None,
model="gpt-4",
)
# The function lists the offending param name in the error.
assert field in str(exc.value)
@pytest.mark.parametrize(
"field",
["api_base", "base_url", "user_config", "langfuse_host", "slack_webhook_url"],
)
def test_api_key_does_not_bypass_blocklist(self, field):
# Regression: the historical ``check_complete_credentials`` clause
# made the entire blocklist a no-op for any caller that supplied
# a non-empty ``api_key``. That bypass turned every missing entry
# on the blocklist into an SSRF / credential-exfil hole. Verify
# that supplying an api_key (alongside the banned param) does NOT
# bypass the gate — it can only be opened by an admin opt-in.
with pytest.raises(ValueError) as exc:
is_request_body_safe(
request_body={
"model": "gpt-4",
"api_key": "sk-anything",
field: "https://attacker.example",
},
general_settings={},
llm_router=None,
model="gpt-4",
)
assert field in str(exc.value)
def test_admin_opt_in_proxy_wide_still_allows(self):
# ``general_settings.allow_client_side_credentials = True`` remains
# the documented proxy-wide BYOK opt-in.
assert (
is_request_body_safe(
request_body={"model": "gpt-4", "api_base": "https://my-byok.example"},
general_settings={"allow_client_side_credentials": True},
llm_router=None,
model="gpt-4",
)
is True
)