fix(security): block IPv4-mapped IPv6 addresses in SSRF check (#327)

2026-07-27 21:05:34 +00:00 · 2026-05-07 14:55:52 -07:00
parent ab46c89660
commit b8245136db
3 changed files with 356 additions and 5 deletions
@@ -13,6 +13,36 @@ static BLOCKED_HOSTS: Lazy<HashSet<&'static str>> = Lazy::new(|| {
    ])
 });

+/// Normalize an IPv6 address to its embedded IPv4 if it is an IPv4-mapped
+/// (`::ffff:a.b.c.d`) or IPv4-compatible (`::a.b.c.d`, deprecated) form.
+/// Returns `None` for `::` and `::1` (these are checked as IPv6 directly).
+fn embedded_ipv4(v6: &Ipv6Addr) -> Option<Ipv4Addr> {
+    if let Some(v4) = v6.to_ipv4_mapped() {
+        return Some(v4);
+    }
+    // IPv4-compatible (::a.b.c.d) — RFC 4291 deprecated but still parseable.
+    // Excludes `::` and `::1` which must be classified as IPv6.
+    let segments = v6.segments();
+    if segments[0] == 0
+        && segments[1] == 0
+        && segments[2] == 0
+        && segments[3] == 0
+        && segments[4] == 0
+        && segments[5] == 0
+        && !v6.is_unspecified()
+        && !v6.is_loopback()
+    {
+        let octets: [u8; 4] = [
+            (segments[6] >> 8) as u8,
+            (segments[6] & 0xff) as u8,
+            (segments[7] >> 8) as u8,
+            (segments[7] & 0xff) as u8,
+        ];
+        return Some(Ipv4Addr::from(octets));
+    }
+    None
+}
+
 /// Check if an IP address is private/reserved.
 pub fn is_private_ip(ip: &IpAddr) -> bool {
    match ip {
@@ -22,9 +52,22 @@ pub fn is_private_ip(ip: &IpAddr) -> bool {
                || v4.is_link_local()
                || is_in_cidr_v4(v4, Ipv4Addr::new(169, 254, 0, 0), 16)
                || *v4 == Ipv4Addr::UNSPECIFIED
+                || v4.is_broadcast()
+                || v4.is_multicast()
+                // 0.0.0.0/8 — current network, routes to localhost on Linux
+                || v4.octets()[0] == 0
        }
        IpAddr::V6(v6) => {
+            // IPv4-mapped (`::ffff:a.b.c.d`) and IPv4-compatible (`::a.b.c.d`)
+            // addresses must be classified by their embedded IPv4. Without
+            // this, the IPv6 checks below never match RFC1918 / loopback /
+            // link-local ranges and the address is treated as public.
+            if let Some(v4) = embedded_ipv4(v6) {
+                return is_private_ip(&IpAddr::V4(v4));
+            }
            v6.is_loopback()
+                || v6.is_unspecified()
+                || v6.is_multicast()
                || is_ula_v6(v6)
                || is_link_local_v6(v6)
        }
@@ -58,24 +101,59 @@ pub fn check_ssrf(url_str: &str) -> Option<String> {
        Err(_) => return Some("Invalid URL".into()),
    };

-    let hostname = match parsed.host_str() {
+    // Use the typed `Host` enum rather than `host_str()`. `host_str()` returns
+    // IPv6 addresses with brackets and in canonicalized hex form (e.g.
+    // `[::ffff:7f00:1]`), which neither `IpAddr::from_str` nor a string-based
+    // `BLOCKED_HOSTS` lookup can match against.
+    let host = match parsed.host() {
        Some(h) => h,
        None => return Some("No hostname in URL".into()),
    };

-    if BLOCKED_HOSTS.contains(hostname) {
+    let literal_ip: Option<IpAddr> = match &host {
+        url::Host::Ipv4(v4) => Some(IpAddr::V4(*v4)),
+        url::Host::Ipv6(v6) => Some(
+            // Normalize IPv4-mapped / IPv4-compatible IPv6 to embedded IPv4
+            // so the `BLOCKED_HOSTS` and private-range checks below catch
+            // them. This is the core SSRF fix.
+            embedded_ipv4(v6)
+                .map(IpAddr::V4)
+                .unwrap_or(IpAddr::V6(*v6)),
+        ),
+        url::Host::Domain(_) => None,
+    };
+
+    // Build the canonical host string for `BLOCKED_HOSTS` lookup. For IPv4
+    // (including normalized IPv4-mapped forms) this is the dotted-decimal
+    // representation; for IPv6 it is the un-bracketed canonical form.
+    let canonical_host: String = match (&literal_ip, &host) {
+        (Some(IpAddr::V4(v4)), _) => v4.to_string(),
+        (Some(IpAddr::V6(v6)), _) => v6.to_string(),
+        (None, url::Host::Domain(d)) => d.to_string(),
+        _ => unreachable!(),
+    };
+
+    if BLOCKED_HOSTS.contains(canonical_host.as_str()) {
        return Some(format!(
            "Blocked host: {} (cloud metadata endpoint)",
-            hostname
+            canonical_host
        ));
    }

+    if let Some(ip) = literal_ip {
+        if is_private_ip(&ip) {
+            return Some(format!("URL resolves to private IP: {}", ip));
+        }
+        return None;
+    }
+
+    // Domain hostname — fall through to DNS resolution.
    let port = parsed.port().unwrap_or(match parsed.scheme() {
        "https" => 443,
        _ => 80,
    });

-    let addr_str = format!("{}:{}", hostname, port);
+    let addr_str = format!("{}:{}", canonical_host, port);
    if let Ok(addrs) = addr_str.to_socket_addrs() {
        for addr in addrs {
            if is_private_ip(&addr.ip()) {
@@ -117,4 +195,126 @@ mod tests {
        let result = check_ssrf("not-a-url");
        assert!(result.is_some());
    }
+
+    #[test]
+    fn test_ipv4_mapped_ipv6_loopback_is_private() {
+        // ::ffff:127.0.0.1
+        let mapped: Ipv6Addr = "::ffff:127.0.0.1".parse().unwrap();
+        assert!(is_private_ip(&IpAddr::V6(mapped)));
+    }
+
+    #[test]
+    fn test_ipv4_mapped_ipv6_rfc1918_is_private() {
+        for s in ["::ffff:10.0.0.1", "::ffff:172.16.0.1", "::ffff:192.168.1.1"] {
+            let v6: Ipv6Addr = s.parse().unwrap();
+            assert!(is_private_ip(&IpAddr::V6(v6)), "{} should be private", s);
+        }
+    }
+
+    #[test]
+    fn test_ipv4_mapped_ipv6_link_local_is_private() {
+        let v6: Ipv6Addr = "::ffff:169.254.0.1".parse().unwrap();
+        assert!(is_private_ip(&IpAddr::V6(v6)));
+    }
+
+    #[test]
+    fn test_ipv4_mapped_public_ipv6_allowed() {
+        // ::ffff:8.8.8.8 — public IPv4 wrapped as IPv6 must NOT be flagged.
+        let v6: Ipv6Addr = "::ffff:8.8.8.8".parse().unwrap();
+        assert!(!is_private_ip(&IpAddr::V6(v6)));
+    }
+
+    #[test]
+    fn test_ipv6_unspecified_is_private() {
+        assert!(is_private_ip(&IpAddr::V6(Ipv6Addr::UNSPECIFIED)));
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_ipv4_mapped_loopback_url() {
+        let result = check_ssrf("http://[::ffff:127.0.0.1]:6666");
+        assert!(result.is_some(), "::ffff:127.0.0.1 must be blocked");
+        assert!(result.unwrap().contains("private IP"));
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_ipv4_mapped_rfc1918_url() {
+        for url in [
+            "http://[::ffff:10.0.0.1]/",
+            "http://[::ffff:192.168.1.1]/",
+            "http://[::ffff:172.16.0.1]/",
+        ] {
+            let result = check_ssrf(url);
+            assert!(result.is_some(), "{} must be blocked", url);
+        }
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_ipv4_mapped_metadata_url() {
+        // ::ffff:169.254.169.254 — IPv4-mapped form of AWS/GCP metadata IP.
+        let result = check_ssrf("http://[::ffff:169.254.169.254]/latest/meta-data/");
+        assert!(result.is_some());
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_ipv6_loopback_literal() {
+        // Bracketed IPv6 literal must be checked even though `host:port`
+        // string parsing of the unbracketed form is ambiguous.
+        let result = check_ssrf("http://[::1]:80/");
+        assert!(result.is_some());
+    }
+
+    #[test]
+    fn test_check_ssrf_allows_ipv4_mapped_public() {
+        let result = check_ssrf("http://[::ffff:8.8.8.8]/");
+        assert!(result.is_none(), "public IP wrapped as IPv6 must be allowed");
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_ipv4_mapped_alibaba_metadata() {
+        // ::ffff:100.100.100.200 — Alibaba metadata; not in any private CIDR,
+        // so must be caught by `BLOCKED_HOSTS` lookup on the normalized IPv4.
+        let result = check_ssrf("http://[::ffff:100.100.100.200]/");
+        assert!(result.is_some(), "Alibaba metadata via mapped form must be blocked");
+        assert!(result.unwrap().contains("Blocked host"));
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_ipv4_compatible_loopback() {
+        // ::127.0.0.1 — deprecated IPv4-compatible form; defense-in-depth.
+        let result = check_ssrf("http://[::127.0.0.1]/");
+        assert!(result.is_some(), "::127.0.0.1 must be blocked");
+    }
+
+    #[test]
+    fn test_check_ssrf_blocks_zero_dot_zero() {
+        // 0.0.0.0 — connects to localhost on Linux.
+        let result = check_ssrf("http://0.0.0.0/");
+        assert!(result.is_some());
+    }
+
+    #[test]
+    fn test_is_private_ip_unspecified_v4_v6() {
+        assert!(is_private_ip(&IpAddr::V4(Ipv4Addr::UNSPECIFIED)));
+        assert!(is_private_ip(&IpAddr::V6(Ipv6Addr::UNSPECIFIED)));
+    }
+
+    #[test]
+    fn test_is_private_ip_blocks_multicast_and_broadcast() {
+        assert!(is_private_ip(&IpAddr::V4(Ipv4Addr::new(239, 0, 0, 1))));
+        assert!(is_private_ip(&IpAddr::V4(Ipv4Addr::BROADCAST)));
+    }
+
+    #[test]
+    fn test_embedded_ipv4_compatibility() {
+        // IPv4-compatible: ::a.b.c.d
+        let v6: Ipv6Addr = "::127.0.0.1".parse().unwrap();
+        assert_eq!(embedded_ipv4(&v6), Some(Ipv4Addr::new(127, 0, 0, 1)));
+        // ::1 must NOT be treated as IPv4-compatible
+        assert_eq!(embedded_ipv4(&Ipv6Addr::LOCALHOST), None);
+        // :: must NOT be treated as IPv4-compatible
+        assert_eq!(embedded_ipv4(&Ipv6Addr::UNSPECIFIED), None);
+        // Mapped form
+        let v6: Ipv6Addr = "::ffff:127.0.0.1".parse().unwrap();
+        assert_eq!(embedded_ipv4(&v6), Some(Ipv4Addr::new(127, 0, 0, 1)));
+    }
 }
@@ -17,24 +17,56 @@ _BLOCKED_HOSTS = frozenset(
 )

 _BLOCKED_CIDR = [
+    ipaddress.ip_network("0.0.0.0/8"),  # current network — routes to localhost on Linux
    ipaddress.ip_network("10.0.0.0/8"),
    ipaddress.ip_network("172.16.0.0/12"),
    ipaddress.ip_network("192.168.0.0/16"),
    ipaddress.ip_network("127.0.0.0/8"),
    ipaddress.ip_network("169.254.0.0/16"),  # link-local
+    ipaddress.ip_network("224.0.0.0/4"),  # multicast
+    ipaddress.ip_network("255.255.255.255/32"),  # broadcast
+    ipaddress.ip_network("::/128"),  # IPv6 unspecified
    ipaddress.ip_network("::1/128"),
    ipaddress.ip_network("fc00::/7"),  # unique local
    ipaddress.ip_network("fe80::/10"),  # link-local v6
+    ipaddress.ip_network("ff00::/8"),  # IPv6 multicast
 ]


+def _embedded_ipv4(addr: ipaddress.IPv6Address) -> ipaddress.IPv4Address | None:
+    """Return the embedded IPv4 for IPv4-mapped (`::ffff:a.b.c.d`) and
+    IPv4-compatible (`::a.b.c.d`, deprecated) IPv6 addresses, else None.
+    Excludes `::` and `::1` so they remain classified as IPv6.
+    """
+    mapped = addr.ipv4_mapped
+    if mapped is not None:
+        return mapped
+    # IPv4-compatible (RFC 4291 §2.5.5.1) — first 80 bits zero, next 16 bits
+    # zero, embedded in last 32. Skip `::` (unspecified) and `::1` (loopback).
+    packed = addr.packed
+    if (
+        packed[:12] == b"\x00" * 12
+        and addr != ipaddress.IPv6Address("::")
+        and addr != ipaddress.IPv6Address("::1")
+    ):
+        return ipaddress.IPv4Address(packed[12:])
+    return None
+
+
 def is_private_ip(ip_str: str) -> bool:
    """Check if an IP address is private/reserved."""
    try:
        addr = ipaddress.ip_address(ip_str)
-        return any(addr in net for net in _BLOCKED_CIDR)
    except ValueError:
        return False
+    # Normalize IPv4-mapped / IPv4-compatible IPv6 to the embedded IPv4 so
+    # the IPv4 private-range CIDRs apply. Without this, e.g. ::ffff:127.0.0.1
+    # bypasses the loopback / RFC1918 checks.
+    if isinstance(addr, ipaddress.IPv6Address):
+        embedded = _embedded_ipv4(addr)
+        if embedded is not None:
+            addr = embedded
+    return any(addr in net for net in _BLOCKED_CIDR)


 def check_ssrf(url: str) -> Optional[str]:
@@ -58,6 +90,25 @@ def _check_ssrf_python(url: str) -> Optional[str]:
    if hostname in _BLOCKED_HOSTS:
        return f"Blocked host: {hostname} (cloud metadata endpoint)"

+    # If the hostname is itself an IP literal, classify it directly.
+    # Required so IPv6 literals (including IPv4-mapped forms) get checked
+    # without going through DNS, and so the metadata-host comparison
+    # catches forms like ::ffff:169.254.169.254.
+    try:
+        literal = ipaddress.ip_address(hostname)
+    except ValueError:
+        literal = None
+    if literal is not None:
+        if isinstance(literal, ipaddress.IPv6Address):
+            embedded = _embedded_ipv4(literal)
+            if embedded is not None:
+                mapped_str = str(embedded)
+                if mapped_str in _BLOCKED_HOSTS:
+                    return f"Blocked host: {mapped_str} (cloud metadata endpoint)"
+        if is_private_ip(hostname):
+            return f"URL resolves to private IP: {hostname}"
+        return None
+
    # DNS resolution check
    try:
        resolved = socket.getaddrinfo(
@@ -41,6 +41,39 @@ class TestIsPrivateIp:
    def test_empty_string(self):
        assert is_private_ip("") is False

+    def test_ipv4_mapped_ipv6_loopback(self):
+        # ::ffff:127.0.0.1 — IPv4-mapped form of loopback must be flagged.
+        assert is_private_ip("::ffff:127.0.0.1") is True
+
+    def test_ipv4_mapped_ipv6_rfc1918(self):
+        assert is_private_ip("::ffff:10.0.0.1") is True
+        assert is_private_ip("::ffff:172.16.0.1") is True
+        assert is_private_ip("::ffff:192.168.1.1") is True
+
+    def test_ipv4_mapped_ipv6_link_local(self):
+        assert is_private_ip("::ffff:169.254.0.1") is True
+
+    def test_ipv4_mapped_ipv6_public_allowed(self):
+        # Public IPv4 wrapped as IPv6 must NOT be flagged as private.
+        assert is_private_ip("::ffff:8.8.8.8") is False
+
+    def test_ipv4_compatible_loopback(self):
+        # ::127.0.0.1 — deprecated IPv4-compatible form, must still be blocked.
+        assert is_private_ip("::127.0.0.1") is True
+
+    def test_unspecified_addresses(self):
+        assert is_private_ip("0.0.0.0") is True
+        assert is_private_ip("::") is True
+
+    def test_zero_subnet_is_private(self):
+        # 0.0.0.0/8 routes to localhost on Linux.
+        assert is_private_ip("0.1.2.3") is True
+
+    def test_multicast_and_broadcast_blocked(self):
+        assert is_private_ip("239.0.0.1") is True
+        assert is_private_ip("255.255.255.255") is True
+        assert is_private_ip("ff02::1") is True
+

 class TestCheckSsrf:
    """Tests for SSRF protection.
@@ -119,5 +152,72 @@ class TestCheckSsrf:
        assert result is not None
        assert "private IP" in result

+    def test_blocks_ipv4_mapped_loopback_url(self):
+        """IPv4-mapped IPv6 form of 127.0.0.1 must be blocked (Rust path)."""
+        result = check_ssrf("http://[::ffff:127.0.0.1]:6666")
+        assert result is not None
+        assert "private IP" in result
+
+    def test_blocks_ipv4_mapped_rfc1918_url(self):
+        for url in (
+            "http://[::ffff:10.0.0.1]/",
+            "http://[::ffff:192.168.1.1]/",
+            "http://[::ffff:172.16.0.1]/",
+        ):
+            result = check_ssrf(url)
+            assert result is not None, f"{url} must be blocked"
+            assert "private IP" in result
+
+    def test_blocks_ipv4_mapped_metadata_url(self):
+        # ::ffff:169.254.169.254 — IPv4-mapped form of cloud metadata IP.
+        result = check_ssrf("http://[::ffff:169.254.169.254]/latest/meta-data/")
+        assert result is not None
+
+    def test_blocks_ipv6_loopback_literal_url(self):
+        result = check_ssrf("http://[::1]:80/")
+        assert result is not None
+
+    def test_allows_ipv4_mapped_public_url(self):
+        result = check_ssrf("http://[::ffff:8.8.8.8]/")
+        assert result is None
+
+    def test_python_impl_blocks_ipv4_mapped_loopback(self):
+        """Legacy Python impl must also catch ::ffff:127.0.0.1."""
+        result = _check_ssrf_python("http://[::ffff:127.0.0.1]:6666")
+        assert result is not None
+        assert "private IP" in result
+
+    def test_python_impl_blocks_ipv4_mapped_metadata(self):
+        result = _check_ssrf_python(
+            "http://[::ffff:169.254.169.254]/latest/meta-data/"
+        )
+        assert result is not None
+
+    def test_blocks_ipv4_mapped_alibaba_metadata(self):
+        # 100.100.100.200 isn't private — relies on BLOCKED_HOSTS lookup
+        # against the embedded v4. Verifies the canonical-host fix.
+        result = check_ssrf("http://[::ffff:100.100.100.200]/")
+        assert result is not None
+        assert "Blocked host" in result
+
+    def test_blocks_ipv4_compatible_loopback(self):
+        result = check_ssrf("http://[::127.0.0.1]/")
+        assert result is not None
+
+    def test_blocks_zero_dot_zero(self):
+        result = check_ssrf("http://0.0.0.0/")
+        assert result is not None
+
+    def test_url_userinfo_does_not_bypass(self):
+        # Userinfo + bracketed IPv6 literal still gets blocked.
+        result = check_ssrf("http://user:pass@[::ffff:127.0.0.1]:8080/admin")
+        assert result is not None
+
+    def test_canonicalized_hex_ipv6_form_blocked(self):
+        # url crate canonicalizes [::ffff:127.0.0.1] to [::ffff:7f00:1]
+        # internally; ensure this form blocks too.
+        result = check_ssrf("http://[::ffff:7f00:1]/")
+        assert result is not None
+

 __all__ = ["TestCheckSsrf", "TestIsPrivateIp"]