diff --git a/litellm/__init__.py b/litellm/__init__.py index 3b67d9e0021..273af465b29 100644 --- a/litellm/__init__.py +++ b/litellm/__init__.py @@ -274,6 +274,8 @@ use_client: bool = False ssl_verify: Union[str, bool] = True ssl_security_level: Optional[str] = None ssl_certificate: Optional[str] = None +user_url_validation: bool = True +user_url_allowed_hosts: List[str] = [] ssl_ecdh_curve: Optional[ str ] = None # Set to 'X25519' to disable PQC and improve performance diff --git a/litellm/litellm_core_utils/url_utils.py b/litellm/litellm_core_utils/url_utils.py index aaeb2bee7ef..e920a044583 100644 --- a/litellm/litellm_core_utils/url_utils.py +++ b/litellm/litellm_core_utils/url_utils.py @@ -7,11 +7,21 @@ input (image_url, file_url, spec_path, etc.) to prevent SSRF attacks. validate_url() resolves DNS once, validates all IPs, and rewrites the URL to connect to the validated IP directly — no TOCTOU gap, no DNS rebinding. Redirects are followed manually with validation at each hop. + +Admins can opt out via two ``litellm`` globals (wired from proxy config): + +- ``litellm.user_url_validation`` (bool, default True): master switch. + When False, ``safe_get``/``async_safe_get`` perform a plain fetch with + no DNS check, no block list, and no rewrite. +- ``litellm.user_url_allowed_hosts`` (List[str], default []): per-host + allowlist. Entries are ``hostname`` or ``hostname:port`` (IPv6 hosts as + ``[addr]`` / ``[addr]:port``). Matching hosts skip the blocked-networks + check but still resolve DNS and still rewrite HTTP to the resolved IP. """ import socket from ipaddress import ip_address, ip_network -from typing import Any, Tuple +from typing import Any, List, Set, Tuple from urllib.parse import urlparse, urlunparse import httpx @@ -52,6 +62,36 @@ def _is_blocked_ip(addr: str) -> bool: return any(ip in net for net in _BLOCKED_NETWORKS) +def _normalize_host(host: str) -> str: + """Lowercase and strip a trailing dot from a hostname.""" + return host.lower().rstrip(".") + + +def _format_host_header(hostname: str, port: int, default_port: int) -> str: + """Build an RFC 7230 Host header value, bracketing IPv6 literals.""" + bracketed = f"[{hostname}]" if ":" in hostname else hostname + if port == default_port: + return bracketed + return f"{bracketed}:{port}" + + +def _is_host_allowlisted(hostname: str, effective_port: int) -> bool: + """Check whether a host is in the admin-configured allowlist. + + Admin entries may be ``hostname`` (any port) or ``hostname:port``. IPv6 + literals are written bracketed (``[::1]`` / ``[::1]:8080``). Matching + is case-insensitive on the hostname. + """ + configured: List[str] = getattr(litellm, "user_url_allowed_hosts", []) or [] + if not configured: + return False + normalized_host = _normalize_host(hostname) + host_repr = f"[{normalized_host}]" if ":" in normalized_host else normalized_host + candidates: Set[str] = {host_repr, f"{host_repr}:{effective_port}"} + allowlist: Set[str] = {_normalize_host(entry) for entry in configured if entry} + return bool(candidates & allowlist) + + def validate_url(url: str) -> Tuple[str, str]: """ Validate a user-supplied URL and rewrite it to connect to a validated IP. @@ -68,9 +108,9 @@ def validate_url(url: str) -> Tuple[str, str]: url: The user-supplied URL to validate. Returns: - Tuple of (rewritten_url, original_hostname). + Tuple of (rewritten_url, host_header). The rewritten URL has the hostname replaced with the validated IP. - The original hostname should be set as the Host header. + The host_header value should be sent as the Host header. Raises: SSRFError: If the URL scheme is invalid or the hostname resolves @@ -87,16 +127,15 @@ def validate_url(url: str) -> Tuple[str, str]: port = parsed.port default_port = 443 if parsed.scheme == "https" else 80 + effective_port = port if port is not None else default_port + host_header = _format_host_header(hostname, effective_port, default_port) - # Build the Host header value — include port when non-default - host_header = ( - hostname if (port is None or port == default_port) else f"{hostname}:{port}" - ) + is_allowlisted = _is_host_allowlisted(hostname, effective_port) # Resolve hostname and validate ALL addresses try: addrinfo = socket.getaddrinfo( - hostname, port or default_port, proto=socket.IPPROTO_TCP + hostname, effective_port, proto=socket.IPPROTO_TCP ) except socket.gaierror as e: raise SSRFError(f"DNS resolution failed for '{hostname}': {e}") @@ -104,13 +143,14 @@ def validate_url(url: str) -> Tuple[str, str]: if not addrinfo: raise SSRFError(f"No addresses found for '{hostname}'") - for family, type_, proto, canonname, sockaddr in addrinfo: - if _is_blocked_ip(sockaddr[0]): - raise SSRFError( - f"URL targets a blocked address ({sockaddr[0]}). " - "If this is a legitimate internal service, use a direct " - "provider configuration instead of a user-supplied URL." - ) + if not is_allowlisted: + for family, type_, proto, canonname, sockaddr in addrinfo: + if _is_blocked_ip(sockaddr[0]): + raise SSRFError( + f"URL targets a blocked address ({sockaddr[0]}). " + "If this is a legitimate internal service, add the host " + "to `user_url_allowed_hosts` in general_settings." + ) # For HTTPS with SSL verification enabled, TLS certificate validation # binds the connection to the hostname — DNS rebinding can't redirect @@ -159,6 +199,9 @@ def safe_get(client: Any, url: str, **kwargs: Any) -> Any: request. No DNS rebinding (resolve-and-rewrite). No redirect bypass (each hop validated). No breaking change for legitimate CDN redirects. + When ``litellm.user_url_validation`` is False, validation is bypassed + and this function delegates to ``client.get(url, follow_redirects=True)``. + Args: client: An httpx.Client (sync). url: The user-supplied URL. @@ -167,6 +210,9 @@ def safe_get(client: Any, url: str, **kwargs: Any) -> Any: Returns: The final httpx.Response. """ + if not getattr(litellm, "user_url_validation", True): + kwargs.setdefault("follow_redirects", True) + return client.get(url, **kwargs) kwargs.pop("follow_redirects", None) caller_headers = kwargs.pop("headers", {}) for _ in range(_MAX_REDIRECTS): @@ -185,6 +231,9 @@ def safe_get(client: Any, url: str, **kwargs: Any) -> Any: async def async_safe_get(client: Any, url: str, **kwargs: Any) -> Any: """Async version of safe_get.""" + if not getattr(litellm, "user_url_validation", True): + kwargs.setdefault("follow_redirects", True) + return await client.get(url, **kwargs) kwargs.pop("follow_redirects", None) caller_headers = kwargs.pop("headers", {}) for _ in range(_MAX_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 1b8121efaca..f6282c6fe79 100644 --- a/tests/test_litellm/litellm_core_utils/test_url_utils.py +++ b/tests/test_litellm/litellm_core_utils/test_url_utils.py @@ -114,3 +114,190 @@ class TestValidateUrl: monkeypatch.setattr(litellm, "ssl_verify", True) rewritten, host = validate_url("https://example.com/image.png") assert rewritten == "https://example.com/image.png" + + +class TestHostHeaderFormatting: + """RFC 7230 §5.4: IPv6 literals must be bracketed in the Host header.""" + + def test_ipv4_no_port(self, monkeypatch): + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("1.2.3.4", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + _, host = validate_url("http://example.com/") + assert host == "example.com" + + def test_ipv4_with_explicit_nondefault_port(self, monkeypatch): + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("1.2.3.4", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + _, host = validate_url("http://example.com:8080/") + assert host == "example.com:8080" + + def test_ipv4_with_explicit_default_port_strips_port(self, monkeypatch): + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("1.2.3.4", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + _, host = validate_url("http://example.com:80/") + assert host == "example.com" + + def test_ipv6_literal_is_bracketed_with_port(self, monkeypatch): + """Regression: IPv6 + port produced ambiguous `Host: 2001:db8::1:8080`.""" + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["[2001:db8::1]"]) + + def fake(host, port, *a, **kw): + return [ + ( + socket.AF_INET6, + socket.SOCK_STREAM, + 6, + "", + ("2001:db8::1", port, 0, 0), + ) + ] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + _, host = validate_url("http://[2001:db8::1]:8080/") + assert host == "[2001:db8::1]:8080" + + def test_ipv6_literal_is_bracketed_without_port(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["[2001:db8::1]"]) + + def fake(host, port, *a, **kw): + return [ + ( + socket.AF_INET6, + socket.SOCK_STREAM, + 6, + "", + ("2001:db8::1", port, 0, 0), + ) + ] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + _, host = validate_url("http://[2001:db8::1]/") + assert host == "[2001:db8::1]" + + +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.""" + monkeypatch.setattr(litellm, "user_url_validation", False) + + calls = [] + + class FakeClient: + def get(self, url, **kwargs): + calls.append((url, kwargs)) + + class R: + is_redirect = False + + return R() + + url_utils.safe_get(FakeClient(), "http://127.0.0.1/internal") + assert calls and calls[0][0] == "http://127.0.0.1/internal" + assert calls[0][1].get("follow_redirects") is True + + def test_enabled_still_blocks(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_validation", True) + with pytest.raises(SSRFError): + validate_url("http://127.0.0.1/") + + +class TestHostAllowlist: + def test_allowlisted_hostname_permits_private_ip(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + rewritten, host = validate_url("http://internal.corp/path") + assert host == "internal.corp" + assert "10.0.1.5" in rewritten + + def test_non_allowlisted_hostname_still_blocked(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + with pytest.raises(SSRFError): + validate_url("http://other.corp/") + + def test_allowlist_case_insensitive(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["Internal.Corp"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + rewritten, _ = validate_url("http://internal.corp/") + assert "10.0.1.5" in rewritten + + def test_allowlist_with_port_matches_explicit_port(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp:8080"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + rewritten, host = validate_url("http://internal.corp:8080/") + assert host == "internal.corp:8080" + assert "10.0.1.5" in rewritten + + def test_allowlist_with_port_matches_default_port(self, monkeypatch): + """Admin entry `host:443` matches `https://host/` (port=None, default 443).""" + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp:443"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + # Should succeed — no SSRFError raised + validate_url("https://internal.corp/") + + def test_allowlist_port_specific_does_not_match_other_port(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp:8080"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + with pytest.raises(SSRFError): + validate_url("http://internal.corp:9090/") + + def test_allowlist_host_entry_matches_any_port(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp"]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + validate_url("http://internal.corp:9090/") + validate_url("https://internal.corp:8443/") + + 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 + rewritten, host = validate_url("http://localhost:8080/") + assert host == "localhost:8080" + + def test_empty_allowlist_retains_default_deny(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", []) + with pytest.raises(SSRFError): + validate_url("http://127.0.0.1/") + + def test_allowlist_strips_trailing_dot(self, monkeypatch): + monkeypatch.setattr(litellm, "user_url_allowed_hosts", ["internal.corp."]) + + def fake(host, port, *a, **kw): + return [(socket.AF_INET, socket.SOCK_STREAM, 6, "", ("10.0.1.5", port))] + + monkeypatch.setattr(url_utils.socket, "getaddrinfo", fake) + validate_url("http://internal.corp/")