fix(guardrails): apply SSRF protection to custom code http_request() primitive

litellm/litellm_core_utils/url_utils.py

@@ -426,3 +426,35 @@ async def async_safe_get(client: Any, url: str, **kwargs: Any) -> Any:
         # relative Location headers keep the original hostname.
         url = _extract_redirect_url(response, url)
     raise SSRFError("Too many redirects")
+
+
+async def async_safe_request(client: Any, method: str, url: str, **kwargs: Any) -> Any:
+    """Method-generic async version of ``safe_get`` with SSRF protection.
+
+    ``client`` must be an ``httpx.AsyncClient`` (exposing ``request``), not a
+    LiteLLM handler, because per-hop redirect control requires disabling the
+    client's default redirect following on every method. Each redirect hop is
+    validated, the connection is made to the validated IP with the original
+    Host header, and redirects are never auto-followed.
+
+    When ``litellm.user_url_validation`` is False, validation is bypassed and
+    this delegates to ``client.request(method, url, follow_redirects=True)``.
+    """
+    if not getattr(litellm, "user_url_validation", True):
+        kwargs.setdefault("follow_redirects", True)
+        return await client.request(method, url, **kwargs)
+    kwargs.pop("follow_redirects", None)
+    caller_headers = kwargs.pop("headers", None) or {}
+    for _ in range(_MAX_REDIRECTS):
+        validated_url, original_host = validate_url(url)
+        response = await client.request(
+            method,
+            validated_url,
+            headers={**caller_headers, "Host": original_host},
+            follow_redirects=False,
+            **kwargs,
+        )
+        if not response.is_redirect:
+            return response
+        url = _extract_redirect_url(response, url)
+    raise SSRFError("Too many redirects")

litellm/proxy/guardrails/guardrail_hooks/custom_code/primitives.py

@@ -13,6 +13,7 @@ from urllib.parse import urlparse
 import httpx
 
 from litellm._logging import verbose_proxy_logger
+from litellm.litellm_core_utils.url_utils import SSRFError, async_safe_request
 from litellm.llms.custom_httpx.http_handler import get_async_httpx_client
 from litellm.types.llms.custom_http import httpxSpecialProvider
 
@@ -468,10 +469,23 @@ async def http_request(
         params={"timeout": httpx.Timeout(timeout=timeout, connect=5.0)},
     )
 
+    json_body, data_body = _prepare_http_body(body)
+
     try:
-        response = await _execute_http_request(client, method, url, headers, body, timeout)
+        response = await async_safe_request(
+            client.client,
+            method,
+            url,
+            headers=headers,
+            json=json_body,
+            data=data_body,
+            timeout=httpx.Timeout(timeout=timeout, connect=5.0),
+        )
         return _http_success_response(response)
 
+    except SSRFError as e:
+        verbose_proxy_logger.warning(f"Custom code http_request blocked by SSRF protection: {e}")
+        return _http_error_response(f"Blocked request: {e}")
     except httpx.TimeoutException as e:
         verbose_proxy_logger.warning(f"Custom code http_request timeout: {e}")
         return _http_error_response(f"Request timeout after {timeout}s")
@@ -486,31 +500,6 @@ async def http_request(
         return _http_error_response(f"Unexpected error: {str(e)}")
 
 
-async def _execute_http_request(
-    client: Any,
-    method: str,
-    url: str,
-    headers: Optional[Dict[str, str]],
-    body: Optional[Any],
-    timeout: float,
-) -> httpx.Response:
-    """Execute the HTTP request using the appropriate client method."""
-    json_body, data_body = _prepare_http_body(body)
-
-    if method == "GET":
-        return await client.get(url=url, headers=headers)
-    elif method == "POST":
-        return await client.post(url=url, headers=headers, json=json_body, data=data_body, timeout=timeout)
-    elif method == "PUT":
-        return await client.put(url=url, headers=headers, json=json_body, data=data_body, timeout=timeout)
-    elif method == "DELETE":
-        return await client.delete(url=url, headers=headers, json=json_body, data=data_body, timeout=timeout)
-    elif method == "PATCH":
-        return await client.patch(url=url, headers=headers, json=json_body, data=data_body, timeout=timeout)
-    else:
-        raise ValueError(f"Unsupported HTTP method: {method}")
-
-
 async def http_get(
     url: str,
     headers: Optional[Dict[str, str]] = None,

tests/test_litellm/litellm_core_utils/test_url_utils.py

@@ -8,6 +8,7 @@ from litellm.litellm_core_utils.url_utils import (
     SSRFError,
     _is_blocked_ip,
     assert_same_origin,
+    async_safe_request,
     encode_url_path_segment,
     encode_url_path_segments,
     validate_url,
@@ -535,3 +536,109 @@ def test_assert_same_origin_error_message_does_not_leak_hostnames():
     detail = str(exc.value)
     assert "attacker.example.com" not in detail
     assert "api.internal-corp.example" not in detail
+
+
+class _FakeResponse:
+    def __init__(self, status, location=None):
+        self.status_code = status
+        self.headers = {"location": location} if location else {}
+        self.is_redirect = 300 <= status < 400
+
+
+class TestAsyncSafeRequest:
+    """async_safe_request must apply the same SSRF protection as safe_get to
+    every HTTP method, not just GET, and validate every redirect hop."""
+
+    async def test_blocks_private_ip_for_post(self, monkeypatch):
+        def fake(host, port, *a, **kw):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", port))]
+
+        monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake)
+        monkeypatch.setattr(litellm, "user_url_validation", True)
+
+        class FakeClient:
+            async def request(self, *a, **kw):
+                raise AssertionError("request must not be issued for a blocked IP")
+
+        with pytest.raises(SSRFError):
+            await async_safe_request(FakeClient(), "POST", "http://internal.example.com/x")
+
+    async def test_metadata_ip_blocked_for_post(self, monkeypatch):
+        def fake(host, port, *a, **kw):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("169.254.169.254", port))]
+
+        monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake)
+        monkeypatch.setattr(litellm, "user_url_validation", True)
+
+        class FakeClient:
+            async def request(self, *a, **kw):
+                raise AssertionError("request must not be issued for metadata IP")
+
+        with pytest.raises(SSRFError):
+            await async_safe_request(FakeClient(), "PUT", "http://metadata.example.com/token")
+
+    async def test_public_post_rewrites_to_ip_and_preserves_method_and_host(self, monkeypatch):
+        def fake(host, port, *a, **kw):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port))]
+
+        monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake)
+        monkeypatch.setattr(litellm, "user_url_validation", True)
+
+        calls = []
+
+        class FakeClient:
+            async def request(self, method, url, headers=None, follow_redirects=False, **kw):
+                calls.append(
+                    {
+                        "method": method,
+                        "url": url,
+                        "host": (headers or {}).get("Host"),
+                        "follow_redirects": follow_redirects,
+                        "kw": kw,
+                    }
+                )
+                return _FakeResponse(200)
+
+        await async_safe_request(
+            FakeClient(),
+            "POST",
+            "http://example.com/api",
+            json={"a": 1},
+        )
+        assert len(calls) == 1
+        assert calls[0]["method"] == "POST"
+        assert calls[0]["host"] == "example.com"
+        assert "93.184.216.34" in calls[0]["url"]
+        assert calls[0]["follow_redirects"] is False
+        assert calls[0]["kw"]["json"] == {"a": 1}
+
+    async def test_redirect_hop_is_revalidated_and_blocked(self, monkeypatch):
+        resolved = {"example.com": "93.184.216.34", "evil.example.com": "127.0.0.1"}
+
+        def fake(host, port, *a, **kw):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (resolved[host], port))]
+
+        monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake)
+        monkeypatch.setattr(litellm, "user_url_validation", True)
+
+        class FakeClient:
+            async def request(self, method, url, headers=None, follow_redirects=False, **kw):
+                return _FakeResponse(302, "http://evil.example.com/internal")
+
+        with pytest.raises(SSRFError):
+            await async_safe_request(FakeClient(), "POST", "http://example.com/start")
+
+    async def test_master_switch_disabled_bypasses_validation(self, monkeypatch):
+        monkeypatch.setattr(litellm, "user_url_validation", False)
+
+        calls = []
+
+        class FakeClient:
+            async def request(self, method, url, **kwargs):
+                calls.append((method, url, kwargs))
+                return _FakeResponse(200)
+
+        await async_safe_request(FakeClient(), "DELETE", "http://127.0.0.1/internal")
+        assert calls and calls[0][0] == "DELETE"
+        assert calls[0][1] == "http://127.0.0.1/internal"
+        assert calls[0][2].get("follow_redirects") is True

tests/test_litellm/proxy/guardrails/guardrail_hooks/custom_code/test_primitives.py

@@ -0,0 +1,122 @@
+import socket
+
+import pytest
+
+import litellm
+from litellm.litellm_core_utils import url_utils
+from litellm.proxy.guardrails.guardrail_hooks.custom_code import primitives
+
+
+class _FakeResponse:
+    def __init__(self, status=200, body=None, location=None):
+        self.status_code = status
+        self._body = body if body is not None else {"ok": True}
+        self.headers = {"location": location} if location else {}
+        self.is_redirect = 300 <= status < 400
+
+    def json(self):
+        return self._body
+
+    @property
+    def text(self):
+        return str(self._body)
+
+
+class _RecordingHandler:
+    def __init__(self, recorder, response=None):
+        self.client = _RecordingClient(recorder, response)
+
+
+class _RecordingClient:
+    def __init__(self, recorder, response=None):
+        self._recorder = recorder
+        self._response = response or _FakeResponse()
+
+    async def request(self, method, url, headers=None, follow_redirects=False, **kw):
+        self._recorder.append(
+            {"method": method, "url": url, "host": (headers or {}).get("Host"), "kw": kw}
+        )
+        return self._response
+
+
+def _patch_dns(monkeypatch, ip):
+    def fake(host, port, *a, **kw):
+        return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (ip, port))]
+
+    monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake)
+
+
+@pytest.fixture(autouse=True)
+def _enable_validation(monkeypatch):
+    monkeypatch.setattr(litellm, "user_url_validation", True)
+
+
+class TestHttpRequestSSRF:
+    """Regression for #32889: http_request() must not reach internal targets."""
+
+    async def test_blocks_loopback(self, monkeypatch):
+        _patch_dns(monkeypatch, "127.0.0.1")
+        recorder = []
+        monkeypatch.setattr(
+            primitives, "get_async_httpx_client", lambda **kw: _RecordingHandler(recorder)
+        )
+
+        result = await primitives.http_request("http://internal.example.com/secret")
+
+        assert result["success"] is False
+        assert "Blocked" in (result["error"] or "")
+        assert recorder == []
+
+    async def test_blocks_cloud_metadata(self, monkeypatch):
+        _patch_dns(monkeypatch, "169.254.169.254")
+        recorder = []
+        monkeypatch.setattr(
+            primitives, "get_async_httpx_client", lambda **kw: _RecordingHandler(recorder)
+        )
+
+        result = await primitives.http_post(
+            "http://metadata.example.com/latest/meta-data/", body={"x": 1}
+        )
+
+        assert result["success"] is False
+        assert "Blocked" in (result["error"] or "")
+        assert recorder == []
+
+    async def test_allows_public_host(self, monkeypatch):
+        _patch_dns(monkeypatch, "93.184.216.34")
+        recorder = []
+        monkeypatch.setattr(
+            primitives, "get_async_httpx_client", lambda **kw: _RecordingHandler(recorder)
+        )
+
+        result = await primitives.http_request(
+            "http://example.com/api", method="POST", body={"a": 1}
+        )
+
+        assert result["success"] is True
+        assert len(recorder) == 1
+        assert recorder[0]["method"] == "POST"
+        assert recorder[0]["host"] == "example.com"
+        assert "93.184.216.34" in recorder[0]["url"]
+        assert recorder[0]["kw"]["json"] == {"a": 1}
+
+    async def test_redirect_to_internal_is_blocked(self, monkeypatch):
+        resolved = {"example.com": "93.184.216.34", "evil.example.com": "127.0.0.1"}
+
+        def fake(host, port, *a, **kw):
+            return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", (resolved[host], port))]
+
+        monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake)
+
+        redirect = _FakeResponse(302, location="http://evil.example.com/internal")
+        recorder = []
+        monkeypatch.setattr(
+            primitives,
+            "get_async_httpx_client",
+            lambda **kw: _RecordingHandler(recorder, response=redirect),
+        )
+
+        result = await primitives.http_get("http://example.com/start")
+
+        assert result["success"] is False
+        assert "Blocked" in (result["error"] or "")