fix(proxy)!: apply request-parameter checks consistently across body, path and form inputs

litellm/proxy/auth/auth_utils.py

@@ -20,6 +20,7 @@
     validate_url,
 )
 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,
 )
@@ -440,6 +441,13 @@ def is_request_body_safe(request_body: dict, general_settings: dict, llm_router:
         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)

litellm/proxy/common_request_processing.py

@@ -67,7 +67,10 @@
     ProxyConfig = _ProxyConfig
 else:
     ProxyConfig = Any
-from litellm.proxy.litellm_pre_call_utils import add_litellm_data_to_request
+from litellm.proxy.litellm_pre_call_utils import (
+    add_litellm_data_to_request,
+    reject_url_valued_destination,
+)
 from litellm.types.utils import (
     ModelResponse,
     ModelResponseStream,
@@ -1286,6 +1289,9 @@ async def common_processing_pre_call_logic(
                 self.data[_metadata_variable_name] = {}
             self.data[_metadata_variable_name]["queue_time_seconds"] = queue_time_seconds
 
+        if isinstance(model, str):
+            reject_url_valued_destination("model", model)
+
         self.data["model"] = (
             general_settings.get("completion_model", None)  # server default
             or user_model  # model name passed via cli args

litellm/proxy/health_endpoints/_health_endpoints.py

@@ -5,7 +5,7 @@
 import secrets
 import time
 import traceback
-from collections.abc import Iterable
+from collections.abc import Iterable, Mapping
 from datetime import datetime, timedelta
 from typing import Any, Final, Literal, TypedDict, cast
 
@@ -29,6 +29,9 @@
     UserAPIKeyAuth,
     WebhookEvent,
 )
+from litellm.proxy.auth.auth_utils import (
+    _BANNED_REQUEST_BODY_PARAMS,  # pyright: ignore[reportPrivateUsage]  # one canonical list, shared with the request-body check
+)
 from litellm.proxy.auth.user_api_key_auth import user_api_key_auth
 from litellm.proxy.db.exception_handler import PrismaDBExceptionHandler
 from litellm.proxy.health_check import (
@@ -43,6 +46,10 @@
     get_in_flight_requests,
 )
 from litellm.proxy.shutdown.graceful_shutdown_manager import GracefulShutdownManager
+from litellm.router_utils.clientside_credential_handler import (
+    _ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE,  # pyright: ignore[reportPrivateUsage]  # one canonical list, shared with the router path
+    clientside_credential_keys,
+)
 
 #### Health ENDPOINTS ####
 
@@ -80,6 +87,45 @@ def _reject_os_environ_references(params: dict) -> None:
                 stack.append(value)
 
 
+_CONFIG_CONNECTION_FIELDS: Final[frozenset[str]] = frozenset(
+    (
+        *_ADMIN_CONFIG_FIELDS_TO_CLEAR_ON_BASE_OVERRIDE,
+        *clientside_credential_keys,
+        "litellm_credential_name",
+    )
+)
+
+
+def _config_base_for_health_check(
+    config_params: Mapping[str, object],
+    request_params: Mapping[str, object],
+    allow_client_side_credentials: bool = False,
+) -> dict[str, object]:
+    """Return the configured parameters to merge under a connection-test request.
+
+    A request that sets its own connection fields describes a connection of its
+    own, so the configuration's credentials are not carried into it: they belong
+    to the endpoint the configuration names. Anything the request does not set
+    still comes from the configuration, which is what lets a request name a
+    configured model and test it as configured.
+
+    ``litellm_credential_name`` is dropped alongside the literal credential
+    fields: it names a stored credential that ``load_credentials_from_list``
+    resolves into the same secrets further down the call, so leaving it in place
+    would reintroduce them by reference.
+
+    ``general_settings.allow_client_side_credentials`` is the existing proxy-wide
+    opt-in for callers supplying their own connection parameters. Where an admin
+    has enabled it, a request may pair its own endpoint with the configured
+    credentials, as it could before.
+    """
+    if allow_client_side_credentials:
+        return dict(config_params)
+    if not any(param in request_params for param in _BANNED_REQUEST_BODY_PARAMS):
+        return dict(config_params)
+    return {key: value for key, value in config_params.items() if key not in _CONFIG_CONNECTION_FIELDS}
+
+
 def get_callback_identifier(callback):
     """
     Get the callback identifier string, handling both strings and objects.
@@ -1785,7 +1831,12 @@ async def test_model_connection(
     from litellm.proxy.management_endpoints.model_management_endpoints import (
         ModelManagementAuthChecks,
     )
-    from litellm.proxy.proxy_server import llm_router, premium_user, prisma_client
+    from litellm.proxy.proxy_server import (
+        general_settings,
+        llm_router,
+        premium_user,
+        prisma_client,
+    )
     from litellm.types.router import Deployment, LiteLLM_Params
 
     try:
@@ -1854,8 +1905,14 @@ async def test_model_connection(
                 )
 
         # Merge: config params (from proxy config) as base, request params override
-        # This allows users to override specific params while using config for credentials
-        litellm_params = {**config_litellm_params, **request_litellm_params}
+        litellm_params = {
+            **_config_base_for_health_check(
+                config_litellm_params,
+                request_litellm_params,
+                allow_client_side_credentials=general_settings.get("allow_client_side_credentials") is True,
+            ),
+            **request_litellm_params,
+        }
 
         ## Auth check
         auth_model_info: Final = loaded_model_info if loaded_model_info is not None else model_info

litellm/proxy/image_endpoints/endpoints.py

@@ -70,6 +70,7 @@ async def image_generation(
     user_api_key_dict: UserAPIKeyAuth = Depends(user_api_key_auth),
     model: str | None = None,
 ):
+    from litellm.proxy.litellm_pre_call_utils import reject_url_valued_destination
     from litellm.proxy.proxy_server import (
         add_litellm_data_to_request,
         general_settings,
@@ -96,6 +97,9 @@ async def image_generation(
             proxy_config=proxy_config,
         )
 
+        if isinstance(model, str):
+            reject_url_valued_destination("model", model)
+
         data["model"] = (
             model
             or general_settings.get("image_generation_model", None)  # server default

litellm/proxy/litellm_pre_call_utils.py

@@ -262,29 +262,37 @@ def _reject_url_valued_destinations(data: dict[str, Any]) -> None:
     are unaffected, while admins can opt specific hosts back in via
     ``litellm.provider_url_destination_allowed_hosts``.
     """
-    allowed_hosts: Final = getattr(litellm, "provider_url_destination_allowed_hosts", []) or []
     for field in _URL_DESTINATION_REQUEST_FIELDS:
         value = data.get(field)
-        if not isinstance(value, str):
+        if isinstance(value, str):
+            reject_url_valued_destination(field, value)
+
+
+def reject_url_valued_destination(field: str, value: str) -> None:
+    """Reject a URL-valued destination identifier unless admin-allowlisted.
+
+    Operates on one field/value pair. ``_reject_url_valued_destinations`` applies
+    it across ``_URL_DESTINATION_REQUEST_FIELDS`` for a request body.
+    """
+    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
-        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 HTTPException(
-                status_code=400,
-                detail={
-                    "error": "invalid_request",
-                    "param": field,
-                    "message": (
-                        f"URL-valued '{field}' is not allowed. Configure custom "
-                        "endpoints with api_base instead, or add the destination "
-                        "host to `provider_url_destination_allowed_hosts` in "
-                        "litellm_settings."
-                    ),
-                },
-            )
+        if is_url_destination_allowed_by_host(candidate, allowed_hosts):
+            continue
+        raise HTTPException(
+            status_code=400,
+            detail={
+                "error": "invalid_request",
+                "param": field,
+                "message": (
+                    f"URL-valued '{field}' 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 _strip_untrusted_request_header_controls(

tests/test_litellm/proxy/auth/test_auth_utils.py

@@ -3010,3 +3010,80 @@ def test_reads_tag_rpm_limit_from_metadata(self):
     def test_returns_none_when_unset(self):
         key = UserAPIKeyAuth(api_key="sk-123")
         assert get_key_tag_rpm_limit(key) is None
+
+
+class TestIsRequestBodySafeChecksBracketNotationMetadata:
+    """Bracket notation is how multipart callers express nested metadata; it is
+    validated the same way the dict form is."""
+
+    @pytest.mark.parametrize("metadata_key", ["metadata", "litellm_metadata"])
+    def test_bracket_notation_banned_param_is_rejected(self, metadata_key):
+        with pytest.raises(ValueError, match="langfuse_host"):
+            is_request_body_safe(
+                request_body={
+                    "purpose": "assistants",
+                    f"{metadata_key}[langfuse_host]": "https://example.invalid",
+                },
+                general_settings={},
+                llm_router=None,
+                model="gpt-4",
+            )
+
+    def test_bracket_notation_api_base_is_rejected(self):
+        with pytest.raises(ValueError, match="api_base"):
+            is_request_body_safe(
+                request_body={"litellm_metadata[api_base]": "https://example.invalid"},
+                general_settings={},
+                llm_router=None,
+                model="gpt-4",
+            )
+
+    def test_bracket_notation_allowed_under_proxy_wide_opt_in(self):
+        assert (
+            is_request_body_safe(
+                request_body={"litellm_metadata[langfuse_host]": "https://byok.example"},
+                general_settings={"allow_client_side_credentials": True},
+                llm_router=None,
+                model="gpt-4",
+            )
+            is True
+        )
+
+    def test_benign_bracket_notation_metadata_is_allowed(self):
+        assert (
+            is_request_body_safe(
+                request_body={
+                    "purpose": "assistants",
+                    "litellm_metadata[spend_logs_metadata][owner]": "john",
+                    "litellm_metadata[tags]": "production",
+                },
+                general_settings={},
+                llm_router=None,
+                model="gpt-4",
+            )
+            is True
+        )
+
+    def test_bracket_notation_matches_json_encoding_for_deeper_nesting(self):
+        """A value nested below the first level is treated the same either way:
+        the check descends one level into metadata, for both encodings."""
+        deep_bracket = {
+            "litellm_metadata[spend_logs_metadata][langfuse_host]": "https://example.invalid"
+        }
+        deep_json = {
+            "litellm_metadata": {"spend_logs_metadata": {"langfuse_host": "https://example.invalid"}}
+        }
+        kwargs = dict(general_settings={}, llm_router=None, model="gpt-4")
+        assert is_request_body_safe(request_body=deep_bracket, **kwargs) is True
+        assert is_request_body_safe(request_body=deep_json, **kwargs) is True
+
+    def test_body_without_bracket_keys_is_unaffected(self):
+        assert (
+            is_request_body_safe(
+                request_body={"model": "gpt-4", "messages": [{"role": "user", "content": "hi"}]},
+                general_settings={},
+                llm_router=None,
+                model="gpt-4",
+            )
+            is True
+        )