From 0602564b66bd20147f0e0d9fd3bc9e4c5e4fd221 Mon Sep 17 00:00:00 2001 From: user <70670632+stuxf@users.noreply.github.com> Date: Thu, 16 Apr 2026 22:03:47 +0000 Subject: [PATCH] fix: switch blocklist to RFC 6890 via ipaddress.is_global, block multicast and Azure Wire Server MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the hand-maintained _BLOCKED_NETWORKS CIDR list with a default-deny check based on ipaddress.is_global (RFC 6890 semantics, implemented by Python's stdlib). Also reject multicast explicitly — is_global returns True for public multicast allocations, which are not legitimate HTTP targets. Only globally-routable cloud-fabric IPs need explicit exceptions; the canonical list contains one entry today: Azure Wire Server (168.63.129.16), an in-fabric service reachable from any Azure VM. Coverage delta picked up automatically via is_global: - Alibaba Cloud metadata (100.100.100.200, CGNAT) - Legacy Oracle metadata (192.0.0.192, IETF Protocol Assignments) - IPv4 documentation ranges (192.0.2.0/24, 198.51.100.0/24, 203.0.113.0/24) - IPv4 reserved/future-use (240.0.0.0/4) and broadcast - IPv6 documentation (2001:db8::/32) Also fix two issues Greptile flagged: - HTTP relative-redirect hops lost the original hostname because _extract_redirect_url joined the Location against the rewritten (IP-based) URL. Join against the pre-rewrite URL so the next hop's Host header keeps the original hostname. - Two unit tests performed real socket.getaddrinfo('localhost') calls. Monkeypatch them. Add coverage tests for every cloud-metadata IP from the canonical SSRF dictionary (AWS/GCP/Azure/Alibaba/Oracle/DO/OpenStack) plus the new multicast/reserved/documentation/broadcast ranges, and a regression test for redirect-hostname preservation. --- litellm/litellm_core_utils/url_utils.py | 39 +++++--- .../litellm_core_utils/test_url_utils.py | 97 ++++++++++++++++++- 2 files changed, 118 insertions(+), 18 deletions(-) diff --git a/litellm/litellm_core_utils/url_utils.py b/litellm/litellm_core_utils/url_utils.py index e920a044583..d22f92642a3 100644 --- a/litellm/litellm_core_utils/url_utils.py +++ b/litellm/litellm_core_utils/url_utils.py @@ -28,19 +28,13 @@ import httpx import litellm -_BLOCKED_NETWORKS = [ - ip_network("0.0.0.0/8"), - ip_network("10.0.0.0/8"), - ip_network("100.64.0.0/10"), - ip_network("127.0.0.0/8"), - ip_network("169.254.0.0/16"), - ip_network("172.16.0.0/12"), - ip_network("192.0.0.0/24"), - ip_network("192.168.0.0/16"), - ip_network("198.18.0.0/15"), - ip_network("::1/128"), - ip_network("fc00::/7"), - ip_network("fe80::/10"), +# Globally-routable IPs that are cloud-internal. Everything else +# non-public is caught by ``not ip.is_global`` (RFC 6890, as implemented by +# Python's ``ipaddress`` module). This list only holds IPs that are +# publicly routable *and* point to cloud-fabric services reachable from +# inside a VM via special in-fabric routing. +_CLOUD_METADATA_EXCEPTIONS = [ + ip_network("168.63.129.16/32"), # Azure Wire Server ] _ALLOWED_SCHEMES = ("http", "https") @@ -53,13 +47,22 @@ class SSRFError(ValueError): def _is_blocked_ip(addr: str) -> bool: + """Return True for any IP not safe to reach from a user-supplied URL. + + Policy: default-deny via ``ip.is_global`` (RFC 6890), plus an explicit + exception list for globally-routable cloud-fabric IPs that are still + dangerous from inside a cloud VM (currently just Azure Wire Server). + Unparseable addresses fail closed. + """ try: ip = ip_address(addr) except ValueError: return True # fail-closed: unparseable addresses are blocked if ip.version == 6 and hasattr(ip, "ipv4_mapped") and ip.ipv4_mapped: ip = ip.ipv4_mapped - return any(ip in net for net in _BLOCKED_NETWORKS) + if not ip.is_global or ip.is_multicast: + return True + return any(ip in net for net in _CLOUD_METADATA_EXCEPTIONS) def _normalize_host(host: str) -> str: @@ -225,7 +228,9 @@ def safe_get(client: Any, url: str, **kwargs: Any) -> Any: ) if not response.is_redirect: return response - url = _extract_redirect_url(response, validated_url) + # Resolve the next hop against the ORIGINAL (pre-rewrite) URL so + # relative Location headers keep the original hostname. + url = _extract_redirect_url(response, url) raise SSRFError("Too many redirects") @@ -246,5 +251,7 @@ async def async_safe_get(client: Any, url: str, **kwargs: Any) -> Any: ) if not response.is_redirect: return response - url = _extract_redirect_url(response, validated_url) + # Resolve the next hop against the ORIGINAL (pre-rewrite) URL so + # relative Location headers keep the original hostname. + url = _extract_redirect_url(response, 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 f6282c6fe79..4579c203218 100644 --- a/tests/test_litellm/litellm_core_utils/test_url_utils.py +++ b/tests/test_litellm/litellm_core_utils/test_url_utils.py @@ -39,6 +39,46 @@ class TestIsBlockedIp: def test_unparseable_is_blocked(self): assert _is_blocked_ip("not-an-ip") is True + # Coverage delta picked up by switching to `not ip.is_global` (RFC 6890) + # over the old hand-maintained CIDR list. + def test_blocks_cgnat_alibaba_metadata(self): + """100.100.100.200 is Alibaba Cloud metadata; lives in CGNAT.""" + assert _is_blocked_ip("100.100.100.200") is True + + def test_blocks_ietf_protocol_assignments_old_oracle_metadata(self): + """192.0.0.192 was the legacy Oracle Cloud metadata IP.""" + assert _is_blocked_ip("192.0.0.192") is True + + def test_blocks_documentation_ranges(self): + assert _is_blocked_ip("192.0.2.1") is True + assert _is_blocked_ip("198.51.100.1") is True + assert _is_blocked_ip("203.0.113.1") is True + + def test_blocks_multicast(self): + assert _is_blocked_ip("224.0.0.1") is True + + def test_blocks_reserved_future_use(self): + assert _is_blocked_ip("240.0.0.1") is True + + def test_blocks_broadcast(self): + assert _is_blocked_ip("255.255.255.255") is True + + def test_blocks_azure_wire_server(self): + """168.63.129.16 is globally routable but cloud-internal — explicit exception.""" + assert _is_blocked_ip("168.63.129.16") is True + + def test_blocks_aws_ipv6_imds(self): + """fd00:ec2::254 is AWS's IPv6 IMDS, in IPv6 ULA (fc00::/7).""" + assert _is_blocked_ip("fd00:ec2::254") is True + + def test_blocks_ipv4_mapped_private(self): + """::ffff:10.0.0.1 must be unwrapped and blocked as 10.0.0.1.""" + assert _is_blocked_ip("::ffff:10.0.0.1") is True + + def test_blocks_ipv4_mapped_azure_wire_server(self): + """::ffff:168.63.129.16 must be unwrapped and blocked via the exception list.""" + assert _is_blocked_ip("::ffff:168.63.129.16") is True + class TestValidateUrl: def test_blocks_loopback(self): @@ -92,7 +132,13 @@ class TestValidateUrl: with pytest.raises(SSRFError, match="DNS resolution failed"): validate_url("http://this-domain-does-not-exist-xyz123.invalid/test") - def test_blocks_localhost_hostname(self): + def test_blocks_localhost_hostname(self, monkeypatch): + def fake(host, port, *a, **kw): + return [ + (socket.AF_INET, socket.SOCK_STREAM, 6, "", ("127.0.0.1", port or 80)) + ] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) with pytest.raises(SSRFError): validate_url("http://localhost/") @@ -181,6 +227,49 @@ class TestHostHeaderFormatting: assert host == "[2001:db8::1]" +class TestRedirectHostnamePreservation: + """Relative-location redirects must keep the original hostname, not the + rewritten IP, so the next hop's Host header still identifies the site.""" + + def test_relative_redirect_preserves_hostname_for_next_hop(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) + + 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 + + hops = [] + + class FakeClient: + def __init__(self): + self._n = 0 + + def get(self, url, headers=None, follow_redirects=False, **kw): + hops.append({"url": url, "host": (headers or {}).get("Host")}) + self._n += 1 + if self._n == 1: + return FakeResponse(302, "/redirected") + return FakeResponse(200) + + url_utils.safe_get(FakeClient(), "http://example.com/initial") + assert len(hops) == 2 + # Both hops must carry the ORIGINAL hostname in the Host header. + assert hops[0]["host"] == "example.com" + assert hops[1]["host"] == "example.com" + # Both outbound URLs go to the resolved IP (rewritten), not the hostname. + assert "93.184.216.34" in hops[0]["url"] + assert "93.184.216.34" in hops[1]["url"] + # The second hop resolved /redirected relative to the original, not the IP. + assert hops[1]["url"].endswith("/redirected") + + class TestValidationMasterSwitch: def test_disabled_bypasses_fetch_in_safe_get(self, monkeypatch): """When user_url_validation is False, safe_get delegates to client.get without validation.""" @@ -284,7 +373,11 @@ class TestHostAllowlist: def test_allowlist_permits_loopback(self, monkeypatch): """Admin may opt into loopback if they explicitly configure it.""" monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["localhost"]) - # localhost resolves locally without needing mocks + + 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) rewritten, host = validate_url("http://localhost:8080/") assert host == "localhost:8080"