From b8245136dbd50d0083bd149b77c40afb8a266255 Mon Sep 17 00:00:00 2001 From: Jon Saad-Falcon <41205309+jonsaadfalcon@users.noreply.github.com> Date: Thu, 7 May 2026 14:55:52 -0700 Subject: [PATCH] fix(security): block IPv4-mapped IPv6 addresses in SSRF check (#327) --- rust/crates/openjarvis-security/src/ssrf.rs | 208 +++++++++++++++++++- src/openjarvis/security/ssrf.py | 53 ++++- tests/security/test_ssrf.py | 100 ++++++++++ 3 files changed, 356 insertions(+), 5 deletions(-) diff --git a/rust/crates/openjarvis-security/src/ssrf.rs b/rust/crates/openjarvis-security/src/ssrf.rs index 6ff0498d..20ec58ac 100644 --- a/rust/crates/openjarvis-security/src/ssrf.rs +++ b/rust/crates/openjarvis-security/src/ssrf.rs @@ -13,6 +13,36 @@ static BLOCKED_HOSTS: Lazy> = 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 { + 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 { 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 = 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))); + } } diff --git a/src/openjarvis/security/ssrf.py b/src/openjarvis/security/ssrf.py index 91f82462..199dd725 100644 --- a/src/openjarvis/security/ssrf.py +++ b/src/openjarvis/security/ssrf.py @@ -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( diff --git a/tests/security/test_ssrf.py b/tests/security/test_ssrf.py index c1484149..39081409 100644 --- a/tests/security/test_ssrf.py +++ b/tests/security/test_ssrf.py @@ -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"]