From 9c410fffb4e54123b553a287535f8b441fba00ef Mon Sep 17 00:00:00 2001 From: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com> Date: Sat, 11 Jul 2026 11:32:48 +0000 Subject: [PATCH] fix(guardrails): strip credential headers on cross-origin redirects in async_safe_request --- litellm/litellm_core_utils/url_utils.py | 34 +++++++++++- .../litellm_core_utils/test_url_utils.py | 54 +++++++++++++++++++ 2 files changed, 86 insertions(+), 2 deletions(-) diff --git a/litellm/litellm_core_utils/url_utils.py b/litellm/litellm_core_utils/url_utils.py index a7538f79fdf..03657d34ec9 100644 --- a/litellm/litellm_core_utils/url_utils.py +++ b/litellm/litellm_core_utils/url_utils.py @@ -21,7 +21,7 @@ Admins can opt out via two ``litellm`` globals (wired from proxy config): import socket from ipaddress import ip_address, ip_network -from typing import Any, List, Optional, Set, Tuple +from typing import Any, Dict, List, Optional, Set, Tuple from urllib.parse import quote, urlparse, urlunparse import httpx @@ -365,6 +365,34 @@ def _extract_redirect_url(response: Any, request_url: str) -> str: return str(httpx.URL(request_url).join(location)) +_SENSITIVE_REDIRECT_HEADERS = frozenset({"authorization", "proxy-authorization", "cookie"}) + + +def _same_origin(first_url: str, second_url: str) -> bool: + """Return True when both URLs share scheme, host, and effective port.""" + first = urlparse(first_url) + second = urlparse(second_url) + first_port = first.port if first.port is not None else _default_port_for_scheme(first.scheme) + second_port = second.port if second.port is not None else _default_port_for_scheme(second.scheme) + return ( + first.scheme == second.scheme + and _normalize_host(first.hostname or "") == _normalize_host(second.hostname or "") + and first_port == second_port + ) + + +def _headers_for_next_hop(headers: Dict[str, str], current_url: str, next_url: str) -> Dict[str, str]: + """Drop credential-bearing headers when a redirect crosses origins. + + Mirrors httpx's own redirect handling, which strips ``Authorization`` and + similar headers on cross-origin hops so a redirect can't exfiltrate the + caller's credentials to a different host. + """ + if _same_origin(current_url, next_url): + return headers + return {k: v for k, v in headers.items() if k.lower() not in _SENSITIVE_REDIRECT_HEADERS} + + def safe_get(client: Any, url: str, **kwargs: Any) -> Any: """ Fetch a user-supplied URL with SSRF protection on every redirect hop. @@ -456,5 +484,7 @@ async def async_safe_request(client: Any, method: str, url: str, **kwargs: Any) ) if not response.is_redirect: return response - url = _extract_redirect_url(response, url) + next_url = _extract_redirect_url(response, url) + caller_headers = _headers_for_next_hop(caller_headers, url, next_url) + url = next_url raise SSRFError("Too many redirects") diff --git a/tests/test_litellm/litellm_core_utils/test_url_utils.py b/tests/test_litellm/litellm_core_utils/test_url_utils.py index 5b672a17c9a..d6e50fdb521 100644 --- a/tests/test_litellm/litellm_core_utils/test_url_utils.py +++ b/tests/test_litellm/litellm_core_utils/test_url_utils.py @@ -628,6 +628,60 @@ class TestAsyncSafeRequest: with pytest.raises(SSRFError): await async_safe_request(FakeClient(), "POST", "http://example.com/start") + async def test_sensitive_headers_stripped_on_cross_origin_redirect(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(dict(headers or {})) + if len(calls) == 1: + return _FakeResponse(302, "https://other.example.org/next") + return _FakeResponse(200) + + await async_safe_request( + FakeClient(), + "GET", + "https://trusted.example.com/start", + headers={"Authorization": "Bearer secret", "Cookie": "sid=1", "X-Trace": "keep"}, + ) + assert len(calls) == 2 + assert calls[0]["Authorization"] == "Bearer secret" + assert calls[0]["Cookie"] == "sid=1" + assert "Authorization" not in calls[1] + assert "Cookie" not in calls[1] + assert calls[1]["X-Trace"] == "keep" + + async def test_sensitive_headers_kept_on_same_origin_redirect(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(dict(headers or {})) + if len(calls) == 1: + return _FakeResponse(302, "https://trusted.example.com/next") + return _FakeResponse(200) + + await async_safe_request( + FakeClient(), + "GET", + "https://trusted.example.com/start", + headers={"Authorization": "Bearer secret"}, + ) + assert len(calls) == 2 + assert calls[1]["Authorization"] == "Bearer secret" + async def test_too_many_redirects_raises(self, monkeypatch): def fake(host, port, *a, **kw): return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("93.184.216.34", port))]