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"