Merge 056629e6ab into 2d0c9eed4d

litellm/proxy/auth/auth_utils.py

@@ -2,7 +2,7 @@ import os
 import re
 import sys
 from collections.abc import Collection, Iterator, Mapping
-from functools import lru_cache
+from functools import lru_cache, partial
 from logging import Logger
 from typing import Any, Final, Protocol
 
@@ -328,19 +328,36 @@ _BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = (
     *sorted(CustomPricingLiteLLMParams.model_fields.keys()),
 )
 
+# Banned only on LLM-invocation routes. ``api_key`` overrides the
+# deployment's admin-configured provider credential (see the clientside
+# precedence check in ``litellm_pre_call_utils``), so it needs the same
+# opt-in as ``api_base``. It cannot join the list above because it is a
+# legitimate body field on control-plane routes that store third-party
+# credentials (e.g. POST /cloudzero/init) and on /health/test_connection.
+_LLM_ROUTE_BANNED_REQUEST_BODY_PARAMS: Final[tuple[str, ...]] = ("api_key",)
+
+
+def _banned_params_for_route(route: str) -> tuple[str, ...]:
+    from litellm.proxy.auth.route_checks import RouteChecks  # noqa: PLC0415  # proxy import cycle
+
+    if RouteChecks.is_llm_api_route(route):
+        return _BANNED_REQUEST_BODY_PARAMS + _LLM_ROUTE_BANNED_REQUEST_BODY_PARAMS
+    return _BANNED_REQUEST_BODY_PARAMS
+
 
 def _check_banned_params(
     body: dict,
     general_settings: dict,
     llm_router: Router | None,
     model: str,
+    banned_params: tuple[str, ...] = _BANNED_REQUEST_BODY_PARAMS,
 ) -> 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:
+    for param in banned_params:
         if param not in body:
             continue
         if general_settings.get("allow_client_side_credentials") is True:
@@ -426,7 +443,13 @@ def _reject_url_valued_fallback_target(value: str) -> None:
         )
 
 
-def is_request_body_safe(request_body: dict, general_settings: dict, llm_router: Router | None, model: str) -> bool:
+def is_request_body_safe(
+    request_body: dict,
+    general_settings: dict,
+    llm_router: Router | None,
+    model: str,
+    route: str = "",
+) -> bool:
     """
     Check if the request body is safe.
 
@@ -454,25 +477,27 @@ def is_request_body_safe(request_body: dict, general_settings: dict, llm_router:
     """
     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)
+    check_banned_params: Final = partial(
+        _check_banned_params,
+        general_settings=general_settings,
+        llm_router=llm_router,
+        model=model,
+        banned_params=_banned_params_for_route(route),
+    )
+    check_banned_params(request_body)
     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)
+            check_banned_params(nested)
     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)
+            check_banned_params(metadata)
         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,
-            )
+            check_banned_params(extract_nested_form_metadata(form_data=request_body, prefix=f"{metadata_key}["))
     for target in iter_request_fallback_targets(request_body):
         if isinstance(target, dict):
-            _check_banned_params(target, general_settings, llm_router, model)
+            check_banned_params(target)
             target_model = target.get("model")
             if isinstance(target_model, str):
                 _reject_url_valued_fallback_target(target_model)
@@ -482,12 +507,7 @@ def is_request_body_safe(request_body: dict, general_settings: dict, llm_router:
     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,
-            )
+            check_banned_params(litellm_params_metadata)
     return True
 
 
@@ -539,6 +559,7 @@ async def pre_db_read_auth_checks(
         general_settings=general_settings,
         llm_router=llm_router,
         model=request_data.get("model", ""),  # [TODO] use model passed in url as well (azure openai routes)
+        route=route,
     )
 
     # Check 3. Check if IP address is allowed

tests/test_litellm/proxy/auth/test_auth_utils.py

@@ -3279,3 +3279,121 @@ class TestIsRequestBodySafeBlocksAwsIdentitySelectors:
             )
             is True
         )
+
+
+class TestClientsideApiKeyRequiresAdminOptIn:
+    """A caller-supplied body ``api_key`` overrides the deployment's
+    admin-configured provider credential, so it needs the same opt-in as
+    ``api_base``. Before this fix the proxy silently signed the upstream
+    call with the caller's key on every LLM route."""
+
+    @pytest.mark.parametrize(
+        "route",
+        ["/v1/chat/completions", "/chat/completions", "/v1/messages", "/v1/embeddings", "/v1/responses"],
+    )
+    def test_body_api_key_rejected_on_llm_routes(self, route):
+        with pytest.raises(ValueError, match="api_key"):
+            is_request_body_safe(
+                request_body={"model": "gemini-flash", "api_key": "caller-supplied-key"},
+                general_settings={},
+                llm_router=None,
+                model="gemini-flash",
+                route=route,
+            )
+
+    def test_extra_body_api_key_rejected(self):
+        with pytest.raises(ValueError, match="api_key"):
+            is_request_body_safe(
+                request_body={"model": "gemini-flash", "extra_body": {"api_key": "caller-supplied-key"}},
+                general_settings={},
+                llm_router=None,
+                model="gemini-flash",
+                route="/v1/chat/completions",
+            )
+
+    def test_proxy_wide_opt_in_allows_body_api_key(self):
+        assert (
+            is_request_body_safe(
+                request_body={"model": "gemini-flash", "api_key": "caller-supplied-key"},
+                general_settings={"allow_client_side_credentials": True},
+                llm_router=None,
+                model="gemini-flash",
+                route="/v1/chat/completions",
+            )
+            is True
+        )
+
+    def test_per_deployment_opt_in_allows_body_api_key(self):
+        from litellm import Router
+
+        router = Router(
+            model_list=[
+                {
+                    "model_name": "gemini-flash",
+                    "litellm_params": {
+                        "model": "gemini/gemini-2.5-flash",
+                        "configurable_clientside_auth_params": ["api_key"],
+                    },
+                }
+            ]
+        )
+        assert (
+            is_request_body_safe(
+                request_body={"model": "gemini-flash", "api_key": "caller-supplied-key"},
+                general_settings={},
+                llm_router=router,
+                model="gemini-flash",
+                route="/v1/chat/completions",
+            )
+            is True
+        )
+
+    def test_body_without_api_key_still_accepted(self):
+        assert (
+            is_request_body_safe(
+                request_body={"model": "gemini-flash", "messages": [{"role": "user", "content": "hi"}]},
+                general_settings={},
+                llm_router=None,
+                model="gemini-flash",
+                route="/v1/chat/completions",
+            )
+            is True
+        )
+
+    def test_control_plane_route_still_accepts_body_api_key(self):
+        # /cloudzero/init and friends store a third-party credential sent in
+        # the body; they must not be caught by the LLM-route-only ban.
+        assert (
+            is_request_body_safe(
+                request_body={"api_key": "cz-key", "connection_id": "conn-1"},
+                general_settings={},
+                llm_router=None,
+                model="",
+                route="/cloudzero/init",
+            )
+            is True
+        )
+
+
+@pytest.mark.asyncio
+async def test_pre_db_read_auth_checks_rejects_body_api_key_on_llm_route():
+    """The route has to be plumbed through to ``is_request_body_safe``, otherwise
+    the ban never applies to a real request."""
+    from litellm.proxy.auth.auth_utils import pre_db_read_auth_checks
+
+    request = Request(
+        scope={
+            "type": "http",
+            "method": "POST",
+            "path": "/v1/chat/completions",
+            "headers": [],
+            "client": ("1.2.3.4", 1234),
+        }
+    )
+    with patch("litellm.proxy.proxy_server.general_settings", {}):
+        with pytest.raises(ValueError, match="api_key"):
+            await pre_db_read_auth_checks(
+                request=request,
+                request_data={"model": "gemini-flash", "api_key": "caller-supplied-key"},
+                route="/v1/chat/completions",
+            )