scriptoriumadmin/litellm
1
0
Fork 0
mirror of https://github.com/BerriAI/litellm.git synced 2026-09-13 23:11:40 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 101

feat: add admin opt-out for user URL validation

Browse source 
Two litellm-level flags wired through litellm_settings YAML:

- user_url_validation (bool, default True): master switch. When False,
  safe_get/async_safe_get bypass validation and call client.get
  directly.
- user_url_allowed_hosts (List[str], default []): per-host allowlist.
  Entries are 'host' (matches any port) or 'host:port' (port-specific).
  Matched hosts skip the blocked-networks check but still resolve DNS
  and still rewrite HTTP to the validated IP, preserving rebinding
  protection within the permitted name.

Also fix an existing Host header bug: IPv6 literals (e.g. 2001:db8::1)
were emitted unbracketed, producing ambiguous values like
'2001:db8::1:8080' per RFC 7230 5.4. Bracket them consistently in
_format_host_header.
This commit is contained in:
user 2026-04-16 21:40:19 +00:00
parent 1f50c6fa66
commit 1d3dda9342
No known key found for this signature in database
3 changed files with 253 additions and 15 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

2
litellm/__init__.py
View file

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

79
litellm/litellm_core_utils/url_utils.py
View file

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

187
tests/test_litellm/litellm_core_utils/test_url_utils.py
View file

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