fix: switch blocklist to RFC 6890 via ipaddress.is_global, block multicast and Azure Wire Server

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

@@ -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")

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"