Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
40 changes: 33 additions & 7 deletions lib/crewai-tools/src/crewai_tools/security/safe_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -150,7 +150,8 @@ def validate_directory_path(path: str, base_dir: str | None = None) -> str:
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 / cloud metadata
ipaddress.ip_network("0.0.0.0/32"),
ipaddress.ip_network("0.0.0.0/8"), # "this" network (was /32 only)
ipaddress.ip_network("100.64.0.0/10"), # CGNAT / shared address space
]

_BLOCKED_IPV6_NETWORKS = [
Expand All @@ -161,15 +162,40 @@ def validate_directory_path(path: str, base_dir: str | None = None) -> str:
]


_NAT64_WELL_KNOWN = ipaddress.ip_network("64:ff9b::/96")


def _unwrap_embedded_ipv4(addr: ipaddress.IPv4Address | ipaddress.IPv6Address):
"""Return embedded IPv4 for IPv4-mapped or well-known NAT64 addresses."""
if isinstance(addr, ipaddress.IPv6Address):
if addr.ipv4_mapped:
return addr.ipv4_mapped
if addr in _NAT64_WELL_KNOWN:
# Last 32 bits carry the IPv4 (RFC 6052 well-known prefix).
return ipaddress.IPv4Address(int(addr) & 0xFFFFFFFF)
return addr


def _is_private_or_reserved(ip_str: str) -> bool:
"""Check if an IP address is private, reserved, or otherwise unsafe."""
"""Check if an IP address is private, reserved, or otherwise unsafe.

Uses ``ipaddress`` global/private classification so ranges the hand-maintained
blocklists omitted (CGNAT ``100.64/10``, documentation nets, Class E, the
rest of ``0.0.0.0/8``) stay blocked. Explicit network lists remain as a
belt-and-suspenders check. Multicast and NAT64-wrapped non-global IPv4 are
also rejected (``is_global`` is True for multicast and for many NAT64 embeds).
"""
try:
addr = ipaddress.ip_address(ip_str)
# Unwrap IPv4-mapped IPv6 addresses (e.g., ::ffff:127.0.0.1) to IPv4
# so they are only checked against IPv4 networks (avoids TypeError when
# an IPv4Address is compared against an IPv6Network).
if isinstance(addr, ipaddress.IPv6Address) and addr.ipv4_mapped:
addr = addr.ipv4_mapped
# Unwrap IPv4-mapped (::ffff:) and well-known NAT64 (64:ff9b::/96) so
# embedded loopback/link-local/private IPv4 cannot bypass the check.
addr = _unwrap_embedded_ipv4(addr)
if addr.is_multicast:
return True
# ``is_global`` is False for CGNAT (100.64/10) even though ``is_private``
# is also False there — that gap let SSRF past the old list check.
if not addr.is_global:
return True
networks = (
_BLOCKED_IPV4_NETWORKS
if isinstance(addr, ipaddress.IPv4Address)
Expand Down
32 changes: 32 additions & 0 deletions lib/crewai-tools/tests/utilities/test_safe_path.py
Original file line number Diff line number Diff line change
Expand Up @@ -154,6 +154,38 @@ def test_blocks_cloud_metadata(self):
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://169.254.169.254/latest/meta-data/")

def test_blocks_cgnat_100_64(self):
"""CGNAT shared address space must be blocked (not covered by is_private)."""
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://100.64.0.1/internal")

def test_blocks_multicast(self):
"""Multicast is is_global=True in ipaddress; still unsafe for fetch."""
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://224.0.0.1/")
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://239.255.255.250/")

def test_blocks_nat64_embedded_link_local(self):
"""Well-known NAT64 prefix can wrap 169.254.169.254 / loopback."""
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://[64:ff9b::a9fe:a9fe]/")
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://[64:ff9b::7f00:1]/")

def test_blocks_benchmarking_198_18(self):
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://198.18.0.1/")

def test_blocks_documentation_192_0_2(self):
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://192.0.2.1/")

def test_blocks_this_network_0_0_0_1(self):
"""Full 0.0.0.0/8, not only 0.0.0.0/32."""
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://0.0.0.1/")

def test_blocks_private_10_range(self):
with pytest.raises(ValueError, match="private/reserved IP"):
validate_url("http://10.0.0.1/internal")
Expand Down