diff --git a/litellm/proxy/auth/auth_utils.py b/litellm/proxy/auth/auth_utils.py index 448c975d123..a57eb2be271 100644 --- a/litellm/proxy/auth/auth_utils.py +++ b/litellm/proxy/auth/auth_utils.py @@ -53,6 +53,12 @@ def _check_valid_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: Optional[str] = None @@ -70,10 +76,24 @@ def check_complete_credentials(request_body: dict) -> bool: return False api_key_value = request_body.get("api_key") - if api_key_value and isinstance(api_key_value, str) and api_key_value.strip(): - return True + if not (api_key_value and isinstance(api_key_value, str) and api_key_value.strip()): + return False - return False + from litellm.litellm_core_utils.url_utils import SSRFError, validate_url + + 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: diff --git a/litellm/router_utils/clientside_credential_handler.py b/litellm/router_utils/clientside_credential_handler.py index c98f614335b..3ce8e37f8a6 100644 --- a/litellm/router_utils/clientside_credential_handler.py +++ b/litellm/router_utils/clientside_credential_handler.py @@ -13,6 +13,33 @@ Ensures cooldowns are applied correctly. clientside_credential_keys = ["api_key", "api_base", "base_url"] +# Admin-configured fields that carry secrets or environment-specific config +# meant for the *original* upstream. When the caller redirects ``api_base`` / +# ``base_url`` to their own server, these MUST NOT flow through unchanged or +# the admin's ``OpenAI-Organization`` header, ``extra_body`` payloads, AWS / +# Vertex / Azure credentials, etc. would be sent to the attacker. Only carry +# them through when the caller explicitly re-supplies the field. +_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE = [ + "organization", + "extra_body", + "extra_headers", + "default_headers", + "api_version", + "api_type", + "azure_ad_token", + "azure_ad_token_provider", + "aws_access_key_id", + "aws_secret_access_key", + "aws_session_token", + "aws_region_name", + "aws_sts_endpoint", + "aws_web_identity_token", + "aws_role_name", + "vertex_credentials", + "vertex_project", + "vertex_location", +] + def is_clientside_credential(request_kwargs: dict) -> bool: """ @@ -34,4 +61,13 @@ def get_dynamic_litellm_params(litellm_params: dict, request_kwargs: dict) -> di for key in clientside_credential_keys: if key in request_kwargs: 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. + if "api_base" in request_kwargs or "base_url" in request_kwargs: + for field in _ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE: + if field not in request_kwargs: + litellm_params.pop(field, None) + return litellm_params diff --git a/tests/test_litellm/proxy/auth/test_auth_utils.py b/tests/test_litellm/proxy/auth/test_auth_utils.py index 15cfc84c232..cadabc8af19 100644 --- a/tests/test_litellm/proxy/auth/test_auth_utils.py +++ b/tests/test_litellm/proxy/auth/test_auth_utils.py @@ -5,6 +5,8 @@ Unit tests for auth_utils functions related to rate limiting and customer ID ext from typing import Optional from unittest.mock import MagicMock, patch +import pytest + from litellm.proxy._types import UserAPIKeyAuth from litellm.proxy.auth.auth_utils import ( _get_customer_id_from_standard_headers, @@ -660,3 +662,146 @@ class TestCheckCompleteCredentials: def test_returns_true_when_api_key_is_valid(self): result = check_complete_credentials({"model": "gpt-4", "api_key": "sk-valid"}) 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. + """ + + @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): + 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_https_target(self): + # No DNS / SSRF guard objection on a normal public host. + result = check_complete_credentials( + { + "model": "gpt-4", + "api_key": "sk-some-clientside-key", + "api_base": "https://api.openai.com/v1", + } + ) + assert result is True + + +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_preserves_admin_config_when_caller_resupplies(self): + 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-admin" + assert out["extra_body"] == {"admin": "value"} + + 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"