fix(server): validate push-notification URLs before dispatch (SSRF hardening)

src/a2a/server/tasks/base_push_notification_sender.py

@@ -1,5 +1,8 @@
 import asyncio
+import ipaddress
 import logging
+import socket
+import urllib.parse
 
 import httpx
 
@@ -20,6 +23,51 @@
 logger = logging.getLogger(__name__)
 
 
+def _ip_is_blocked(ip_str: str) -> bool:
+    """Whether an address is not a public unicast destination."""
+    try:
+        addr = ipaddress.ip_address(ip_str.split('%', maxsplit=1)[0])
+    except ValueError:
+        return True
+    return (
+        addr.is_private
+        or addr.is_loopback
+        or addr.is_link_local
+        or addr.is_multicast
+        or addr.is_reserved
+        or addr.is_unspecified
+    )
+
+
+def push_url_validation_error(url: str) -> str | None:
+    """Return an error string if a push-notification URL is not safe.
+
+    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.
+    """
+    try:
+        parsed = urllib.parse.urlparse(url)
+    except ValueError:
+        return 'unparseable URL'
+    if parsed.scheme not in ('http', 'https'):
+        return f"scheme '{parsed.scheme}' is not http/https"
+    host = parsed.hostname
+    if not host:
+        return 'no hostname'
+    port = parsed.port or (443 if parsed.scheme == 'https' else 80)
+    try:
+        infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    except socket.gaierror:
+        return f"host '{host}' could not be resolved"
+    for info in infos:
+        if _ip_is_blocked(info[4][0]):
+            return f"host '{host}' resolves to a non-public address"
+    return None
+
+
 class BasePushNotificationSender(PushNotificationSender):
     """Base implementation of PushNotificationSender interface."""
 
@@ -28,6 +76,8 @@
         httpx_client: httpx.AsyncClient,
         config_store: PushNotificationConfigStore,
         context: ServerCallContext | None = None,
+        *,
+        allow_private_push_urls: bool = False,
     ) -> None:
         """Initializes the BasePushNotificationSender.
 
@@ -41,6 +91,13 @@
               Pass None (the default) in new code. A non-None
               value logs a deprecation warning and is otherwise
               ignored.
+            allow_private_push_urls: Push-notification URLs are
+              client-supplied and the server POSTs to them, which makes
+              them an SSRF vector (cloud metadata endpoints, internal
+              services). By default each URL is validated at dispatch
+              time and non-public targets are dropped. Set this to True
+              only in deployments whose legitimate webhooks live on
+              private networks (validation is then skipped entirely).
         """
         if context is not None:
             logger.warning(
@@ -54,6 +111,7 @@
             )
         self._client = httpx_client
         self._config_store = config_store
+        self._allow_private_push_urls = allow_private_push_urls
 
     async def send_notification(
         self, task_id: str, event: PushNotificationEvent
@@ -81,6 +139,15 @@
         task_id: str,
     ) -> bool:
         url = push_info.url
+        if not self._allow_private_push_urls:
+            validation_error = push_url_validation_error(url)
+            if validation_error:
+                logger.warning(
+                    'Push-notification URL for task_id=%s rejected: %s',
+                    task_id,
+                    validation_error,
+                )
+                return False
         try:
             headers = None
             if push_info.token:

tests/integration/push_notifications/agent_app.py

@@ -154,6 +154,8 @@ def create_agent_app(
         push_sender=BasePushNotificationSender(
             httpx_client=notification_client,
             config_store=push_config_store,
+            # e2e webhooks are real local test servers (loopback).
+            allow_private_push_urls=True,
         ),
     )
     rest_routes = create_rest_routes(request_handler=handler)
@@ -225,6 +227,8 @@ def create_multi_user_agent_app(
         push_sender=BasePushNotificationSender(
             httpx_client=notification_client,
             config_store=push_config_store,
+            # e2e webhooks are real local test servers (loopback).
+            allow_private_push_urls=True,
         ),
     )
 

tests/server/tasks/test_inmemory_push_notifications.py

@@ -67,6 +67,14 @@ class TestInMemoryPushNotifier(unittest.IsolatedAsyncioTestCase):
     def setUp(self) -> None:
         self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient)
         self.config_store = InMemoryPushNotificationConfigStore()
+        # Keep DNS hermetic: pretend every test URL resolves to a public IP
+        # (push-URL SSRF validation is on by default now).
+        getaddrinfo_patch = patch(
+            'a2a.server.tasks.base_push_notification_sender.socket.getaddrinfo',
+            return_value=[(2, 1, 6, '', ('93.184.216.34', 80))],
+        )
+        self.addCleanup(getaddrinfo_patch.stop)
+        getaddrinfo_patch.start()
         self.notifier = BasePushNotificationSender(
             httpx_client=self.mock_httpx_client,
             config_store=self.config_store,
@@ -446,6 +454,14 @@ def setUp(self) -> None:
 
         self.config_store = InMemoryPushNotificationConfigStore()
 
+        # Keep DNS hermetic: pretend every test URL resolves to a public IP
+        # (push-URL SSRF validation is on by default now).
+        getaddrinfo_patch = patch(
+            'a2a.server.tasks.base_push_notification_sender.socket.getaddrinfo',
+            return_value=[(2, 1, 6, '', ('93.184.216.34', 80))],
+        )
+        self.addCleanup(getaddrinfo_patch.stop)
+        getaddrinfo_patch.start()
         self.sender = BasePushNotificationSender(
             httpx_client=self.mock_httpx_client,
             config_store=self.config_store,

tests/server/tasks/test_push_notification_sender.py

@@ -42,6 +42,13 @@ class TestBasePushNotificationSender(unittest.IsolatedAsyncioTestCase):
     def setUp(self) -> None:
         self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient)
         self.mock_config_store = AsyncMock()
+        # Keep DNS hermetic: pretend every test URL resolves to a public IP.
+        getaddrinfo_patch = patch(
+            'a2a.server.tasks.base_push_notification_sender.socket.getaddrinfo',
+            return_value=[(2, 1, 6, '', ('93.184.216.34', 80))],
+        )
+        self.addCleanup(getaddrinfo_patch.stop)
+        getaddrinfo_patch.start()
         self.sender = BasePushNotificationSender(
             httpx_client=self.mock_httpx_client,
             config_store=self.mock_config_store,
@@ -228,3 +235,78 @@ async def test_send_notification_artifact_update_event(self) -> None:
             json=MessageToDict(StreamResponse(artifact_update=event)),
             headers=None,
         )
+
+
+_GAI = 'a2a.server.tasks.base_push_notification_sender.socket.getaddrinfo'
+
+
+def _gai_result(ip: str, port: int = 80):
+    return [(2, 1, 6, '', (ip, port))]
+
+
+class TestPushUrlValidation(unittest.IsolatedAsyncioTestCase):
+    """SSRF hardening: client-supplied push URLs must not reach non-public
+    destinations unless the operator explicitly opts out."""
+
+    def setUp(self) -> None:
+        self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient)
+        self.mock_config_store = AsyncMock()
+        self.sender = BasePushNotificationSender(
+            httpx_client=self.mock_httpx_client,
+            config_store=self.mock_config_store,
+        )
+
+    async def _dispatch(self, url: str) -> None:
+        task = _create_sample_task()
+        config = _create_sample_push_config(url=url)
+        self.mock_config_store.get_info_for_dispatch.return_value = [config]
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.status_code = 200
+        self.mock_httpx_client.post.return_value = mock_response
+        await self.sender.send_notification(task.id, task)
+
+    async def test_metadata_endpoint_blocked(self) -> None:
+        with patch(_GAI, return_value=_gai_result('169.254.169.254')):
+            await self._dispatch('http://metadata.google.internal/latest')
+        self.mock_httpx_client.post.assert_not_called()
+
+    async def test_loopback_blocked(self) -> None:
+        with patch(_GAI, return_value=_gai_result('127.0.0.1')):
+            await self._dispatch('http://localhost:8080/admin')
+        self.mock_httpx_client.post.assert_not_called()
+
+    async def test_private_range_blocked(self) -> None:
+        with patch(_GAI, return_value=_gai_result('10.0.0.5')):
+            await self._dispatch('http://internal-service/endpoint')
+        self.mock_httpx_client.post.assert_not_called()
+
+    async def test_non_http_scheme_blocked(self) -> None:
+        await self._dispatch('ftp://example.com/file')
+        self.mock_httpx_client.post.assert_not_called()
+
+    async def test_unresolvable_host_blocked_fail_closed(self) -> None:
+        import socket as _socket
+
+        with patch(_GAI, side_effect=_socket.gaierror('no DNS')):
+            await self._dispatch('http://does-not-resolve.invalid/')
+        self.mock_httpx_client.post.assert_not_called()
+
+    async def test_public_host_allowed(self) -> None:
+        with patch(_GAI, return_value=_gai_result('93.184.216.34')):
+            await self._dispatch('http://notify.me/here')
+        self.mock_httpx_client.post.assert_awaited_once()
+
+    async def test_allow_private_opt_out(self) -> None:
+        sender = BasePushNotificationSender(
+            httpx_client=self.mock_httpx_client,
+            config_store=self.mock_config_store,
+            allow_private_push_urls=True,
+        )
+        task = _create_sample_task()
+        config = _create_sample_push_config(url='http://localhost:9000/hook')
+        self.mock_config_store.get_info_for_dispatch.return_value = [config]
+        mock_response = AsyncMock(spec=httpx.Response)
+        mock_response.status_code = 200
+        self.mock_httpx_client.post.return_value = mock_response
+        await sender.send_notification(task.id, task)
+        self.mock_httpx_client.post.assert_awaited_once()