From 3e8c5c42d2aa475a1540dc52fa8847bdbc6e20ba Mon Sep 17 00:00:00 2001 From: Sasha Mitchell Date: Fri, 9 Oct 2026 08:29:25 +0700 Subject: [PATCH] fix(server): reject push URLs in shared address space 100.64.0.0/10 is not a public destination, but ipaddress.is_private is false for it, so the push URL screen accepted a webhook that resolved there. Judge an IPv4-mapped address as the IPv4 it carries. --- src/a2a/utils/push_url_validator.py | 34 ++++++++++++++----- .../tasks/test_push_notification_sender.py | 15 ++++++++ 2 files changed, 41 insertions(+), 8 deletions(-) diff --git a/src/a2a/utils/push_url_validator.py b/src/a2a/utils/push_url_validator.py index 7eb3072fa..fa9002e37 100644 --- a/src/a2a/utils/push_url_validator.py +++ b/src/a2a/utils/push_url_validator.py @@ -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 @@ -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 diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index 0c87a6d4e..e1aaa848a 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -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()