diff --git a/litellm/proxy/common_utils/url_utils.py b/litellm/proxy/common_utils/url_utils.py index 2a160d3cec7..72cbc776589 100644 --- a/litellm/proxy/common_utils/url_utils.py +++ b/litellm/proxy/common_utils/url_utils.py @@ -15,6 +15,8 @@ from ipaddress import ip_address, ip_network from typing import Any, Optional, Tuple, Union from urllib.parse import urlparse, urlunparse +import litellm + _BLOCKED_NETWORKS = [ ip_network("0.0.0.0/8"), ip_network("10.0.0.0/8"), @@ -43,7 +45,7 @@ def _is_blocked_ip(addr: str) -> bool: try: ip = ip_address(addr) except ValueError: - return False + 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) @@ -104,10 +106,13 @@ def validate_url(url: str) -> Tuple[str, str]: "provider configuration instead of a user-supplied URL." ) - # For HTTPS, TLS certificate validation binds the connection to the - # hostname — DNS rebinding can't redirect to a different server because - # the cert wouldn't match. Return the original URL. - if parsed.scheme == "https": + # For HTTPS with SSL verification enabled, TLS certificate validation + # binds the connection to the hostname — DNS rebinding can't redirect + # to a different server because the cert wouldn't match. + # When SSL verification is disabled, this defense doesn't apply, so + # we rewrite to the validated IP like HTTP. + ssl_verify = getattr(litellm, "ssl_verify", True) + if parsed.scheme == "https" and ssl_verify is not False: return url, hostname # For HTTP, rewrite URL to connect to the validated IP directly diff --git a/tests/test_litellm/proxy/common_utils/test_url_utils.py b/tests/test_litellm/proxy/common_utils/test_url_utils.py index 73f465cfbc4..4dbda6a815e 100644 --- a/tests/test_litellm/proxy/common_utils/test_url_utils.py +++ b/tests/test_litellm/proxy/common_utils/test_url_utils.py @@ -1,6 +1,18 @@ import pytest -from litellm.proxy.common_utils.url_utils import SSRFError, validate_url +import litellm +from litellm.proxy.common_utils.url_utils import SSRFError, _is_blocked_ip, validate_url + + +class TestIsBlockedIp: + def test_blocks_private(self): + assert _is_blocked_ip("10.0.0.1") is True + + def test_allows_public(self): + assert _is_blocked_ip("8.8.8.8") is False + + def test_unparseable_is_blocked(self): + assert _is_blocked_ip("not-an-ip") is True class TestValidateUrl: @@ -62,3 +74,14 @@ class TestValidateUrl: def test_blocks_ipv6_loopback(self): with pytest.raises(SSRFError): validate_url("http://[::1]/") + + def test_https_rewrites_when_ssl_verify_disabled(self, monkeypatch): + monkeypatch.setattr(litellm, "ssl_verify", False) + rewritten, host = validate_url("https://example.com/image.png") + assert host == "example.com" + assert "example.com" not in rewritten # rewritten to IP + + def test_https_not_rewritten_when_ssl_verify_enabled(self, monkeypatch): + monkeypatch.setattr(litellm, "ssl_verify", True) + rewritten, host = validate_url("https://example.com/image.png") + assert rewritten == "https://example.com/image.png"