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>

litellm/llms/vertex_ai/batches/handler.py

@@ -4,7 +4,7 @@ from typing import Any, Coroutine, Dict, Optional, Union
 import httpx
 
 import litellm
-from litellm.litellm_core_utils.url_utils import safe_get
+from litellm.litellm_core_utils.url_utils import async_safe_get, safe_get
 from litellm.llms.custom_httpx.http_handler import (
     _get_httpx_client,
     get_async_httpx_client,
@@ -277,8 +277,13 @@ class VertexAIBatchPrediction(VertexLLM):
                     },
                 )
 
-        response = await client.get(
-            url=api_base,
+        # Mirror the sync path: ``api_base`` may come from caller-supplied
+        # 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,
         )
         if response.status_code != 200:

litellm/proxy/auth/auth_utils.py

@@ -6,6 +6,7 @@ from typing import Any, List, Optional, Tuple
 
 from fastapi import HTTPException, Request, status
 
+import litellm
 from litellm import Router, provider_list
 from litellm._logging import verbose_proxy_logger
 from litellm.constants import STANDARD_CUSTOMER_ID_HEADERS
@@ -80,17 +81,22 @@ def check_complete_credentials(request_body: dict) -> bool:
     if not (api_key_value and isinstance(api_key_value, str) and api_key_value.strip()):
         return 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})."
-            )
+    # ``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
 

litellm/router_utils/clientside_credential_handler.py

@@ -79,9 +79,16 @@ def get_dynamic_litellm_params(litellm_params: dict, request_kwargs: dict) -> di
     # 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:
-            if field not in request_kwargs:
-                litellm_params.pop(field, None)
+            litellm_params.pop(field, None)
+            if field in request_kwargs:
+                litellm_params[field] = request_kwargs[field]
 
     return litellm_params

tests/test_litellm/proxy/auth/test_auth_utils.py

@@ -670,8 +670,18 @@ class TestCheckCompleteCredentialsBlocksSSRF:
     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"],
@@ -687,28 +697,58 @@ class TestCheckCompleteCredentialsBlocksSSRF:
         ],
     )
     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,
-                }
-            )
+        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_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",
-            }
-        )
+    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:
     """
@@ -767,7 +807,14 @@ class TestGetDynamicLitellmParamsClearsAdminConfigOnBaseOverride:
         assert "vertex_credentials" not in out
         assert "vertex_project" not in out
 
-    def test_preserves_admin_config_when_caller_resupplies(self):
+    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,
         )
@@ -784,8 +831,32 @@ class TestGetDynamicLitellmParamsClearsAdminConfigOnBaseOverride:
                 "extra_body": {"attacker": "value"},
             },
         )
-        assert out["organization"] == "org-admin"
-        assert out["extra_body"] == {"admin": "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 (