SSRF protection: Edge case handling for CGNAT and v4/v6 mapping (Z#23236468)

This commit is contained in:
Raphael Michel
2026-06-09 11:16:30 +02:00
parent 93469d33e5
commit 62f35f0c10
2 changed files with 5 additions and 1 deletions

View File

@@ -148,13 +148,14 @@ def monkeypatch_urllib3_ssrf_protection():
if not getattr(settings, "ALLOW_HTTP_TO_PRIVATE_NETWORKS", False): if not getattr(settings, "ALLOW_HTTP_TO_PRIVATE_NETWORKS", False):
ip_addr = ipaddress.ip_address(sa[0]) ip_addr = ipaddress.ip_address(sa[0])
check_ip4 = ip_addr.ipv4_mapped if getattr(ip_addr, "ipv4_mapped", None) else ip_addr
if ip_addr.is_multicast: if ip_addr.is_multicast:
raise HTTPError(f"Request to multicast address {sa[0]} blocked") raise HTTPError(f"Request to multicast address {sa[0]} blocked")
if ip_addr.is_loopback or ip_addr.is_link_local: if ip_addr.is_loopback or ip_addr.is_link_local:
raise HTTPError(f"Request to local address {sa[0]} blocked") raise HTTPError(f"Request to local address {sa[0]} blocked")
if ip_addr.is_private: if ip_addr.is_private:
raise HTTPError(f"Request to private address {sa[0]} blocked") raise HTTPError(f"Request to private address {sa[0]} blocked")
if ip_addr in _cgnat_net: if check_ip4 in _cgnat_net:
raise HTTPError(f"Request to RFC 6598 address {sa[0]} blocked") raise HTTPError(f"Request to RFC 6598 address {sa[0]} blocked")
sock = None sock = None

View File

@@ -43,6 +43,8 @@ def test_private_ip_blocked():
requests.get("https://10.0.0.1", timeout=0.1) requests.get("https://10.0.0.1", timeout=0.1)
with pytest.raises(HTTPError, match="Request to RFC 6598 address.*"): with pytest.raises(HTTPError, match="Request to RFC 6598 address.*"):
requests.get("https://100.100.100.100", timeout=0.1) requests.get("https://100.100.100.100", timeout=0.1)
with pytest.raises(HTTPError, match="Request to RFC 6598 address.*"):
requests.get("https://[::ffff:100.64.0.1]", timeout=0.1)
@pytest.mark.django_db @pytest.mark.django_db
@@ -58,6 +60,7 @@ def test_private_ip_blocked():
[(AF_INET6, SOCK_STREAM, 6, '', ('fe80::1', 443, 0, 0))], [(AF_INET6, SOCK_STREAM, 6, '', ('fe80::1', 443, 0, 0))],
[(AF_INET6, SOCK_STREAM, 6, '', ('ff00::1', 443, 0, 0))], [(AF_INET6, SOCK_STREAM, 6, '', ('ff00::1', 443, 0, 0))],
[(AF_INET6, SOCK_STREAM, 6, '', ('fc00::1', 443, 0, 0))], [(AF_INET6, SOCK_STREAM, 6, '', ('fc00::1', 443, 0, 0))],
[(AF_INET6, SOCK_STREAM, 6, "", ("::ffff:100.64.0.1", 443, 0, 0))],
]) ])
def test_dns_resolving_to_local_blocked(res): def test_dns_resolving_to_local_blocked(res):
with mock.patch('socket.getaddrinfo') as mock_addr: with mock.patch('socket.getaddrinfo') as mock_addr: