diff --git a/src/a2a/utils/push_url_validator.py b/src/a2a/utils/push_url_validator.py index 7eb3072fa..97240bde0 100644 --- a/src/a2a/utils/push_url_validator.py +++ b/src/a2a/utils/push_url_validator.py @@ -11,13 +11,22 @@ 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. + ``ipaddress`` leaves ``is_private`` false for it and ``is_global`` + false as well, so ``not is_global`` still rejects 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 return ( - addr.is_private + not addr.is_global or addr.is_loopback or addr.is_link_local or addr.is_multicast @@ -30,14 +39,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 afc6f982e..4772581b7 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -400,6 +400,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()