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
34 changes: 26 additions & 8 deletions src/a2a/utils/push_url_validator.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,31 @@

logger = logging.getLogger(__name__)

# RFC 6598. ipaddress leaves is_private false for this range on every
# Python this package supports (3.10 through 3.14), while is_global is
# also false. The flag checks below would accept it.
_SHARED_ADDRESS_SPACE = ipaddress.ip_network('100.64.0.0/10')


def _ip_is_blocked(ip_str: str) -> bool:
"""Whether an address is not a public unicast destination."""
"""Whether an address is not a public unicast destination.

RFC 6598 shared address space (100.64.0.0/10) is not public, but
``ipaddress`` leaves ``is_private`` false for it. An IPv4-mapped
IPv6 address is judged as the IPv4 address inside it.
"""
try:
addr = ipaddress.ip_address(ip_str.split('%', maxsplit=1)[0])
except ValueError:
return True
mapped = getattr(addr, 'ipv4_mapped', None)
if mapped is not None:
addr = mapped
if (
isinstance(addr, ipaddress.IPv4Address)
and addr in _SHARED_ADDRESS_SPACE
):
return True
return (
addr.is_private
or addr.is_loopback
Expand All @@ -30,14 +48,14 @@ async def validate_push_notification_url(url: str) -> bool:
"""Return True if a push-notification URL is safe to fetch.

Blocks non-HTTP(S) schemes and hosts that resolve to loopback,
link-local, private, reserved, multicast, or unspecified addresses
(e.g. 169.254.169.254 cloud metadata, internal services). A host
that cannot be resolved is rejected: the POST would fail anyway,
and failing closed avoids treating resolution errors as a bypass.
link-local, private, shared (100.64.0.0/10), reserved, multicast,
or unspecified addresses (e.g. 169.254.169.254 cloud metadata,
internal services). A host that cannot be resolved is rejected:
the POST would fail anyway, and failing closed avoids treating
resolution errors as a bypass.

IPv4-mapped IPv6 forms are covered: ``ipaddress`` maps them to the
underlying IPv4 address, so the ``is_private``/``is_loopback``
checks apply to the mapped value.
IPv4-mapped IPv6 forms are judged as the IPv4 address they carry,
so the private, shared, and loopback checks apply to that address.

Uses the running event-loop resolver so request handlers and the
sender stay non-blocking. Deployments can pass this function as
Expand Down
15 changes: 15 additions & 0 deletions tests/server/tasks/test_push_notification_sender.py
Original file line number Diff line number Diff line change
Expand Up @@ -360,6 +360,21 @@ async def test_private_range_blocked(self) -> None:
await self._dispatch('http://internal-service/endpoint')
self.mock_httpx_client.post.assert_not_called()

async def test_shared_address_space_blocked(self) -> None:
with self._patch_gai(return_value=_gai_result('100.64.0.1')):
await self._dispatch('http://shared.example/hook')
self.mock_httpx_client.post.assert_not_called()

async def test_ipv4_mapped_shared_address_blocked(self) -> None:
with self._patch_gai(return_value=_gai_result('::ffff:100.64.0.1')):
await self._dispatch('http://shared.example/hook')
self.mock_httpx_client.post.assert_not_called()

async def test_address_below_shared_space_allowed(self) -> None:
with self._patch_gai(return_value=_gai_result('100.63.255.255')):
await self._dispatch('http://notify.me/here')
self.mock_httpx_client.post.assert_awaited_once()

async def test_non_http_scheme_blocked(self) -> None:
await self._dispatch('ftp://example.com/file')
self.mock_httpx_client.post.assert_not_called()
Expand Down
Loading