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

src/a2a/server/request_handlers/default_request_handler.py

@@ -95,36 +95,35 @@
         task_store: TaskStore,
         agent_card: AgentCard,
         queue_manager: QueueManager | None = None,
         push_config_store: PushNotificationConfigStore | None = None,
         push_sender: PushNotificationSender | None = None,
         request_context_builder: RequestContextBuilder | None = None,
         extended_agent_card: AgentCard | None = None,
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], Awaitable[AgentCard]
         ]
         | None = None,
-        push_url_validator: Callable[[str], Awaitable[str | None]]
-        | None = None,
+        push_url_validator: Callable[[str], Awaitable[bool]] | None = None,
     ) -> None:
         """Initializes the DefaultRequestHandler.
 
         Args:
             agent_executor: The `AgentExecutor` instance to run agent logic.
             task_store: The `TaskStore` instance to manage task persistence.
             agent_card: The `AgentCard` describing the agent's capabilities.
             queue_manager: The `QueueManager` instance to manage event queues. Defaults to `InMemoryQueueManager`.
             push_config_store: The `PushNotificationConfigStore` instance for managing push notification configurations. Defaults to None.
             push_sender: The `PushNotificationSender` instance for sending push notifications. Defaults to None.
             request_context_builder: The `RequestContextBuilder` instance used
               to build request contexts. Defaults to `SimpleRequestContextBuilder`.
             extended_agent_card: An optional, distinct `AgentCard` to be served at the extended card endpoint.
             extended_card_modifier: An optional callback to dynamically modify the extended `AgentCard` before it is served.
-            push_url_validator: Async callable that returns an error string
-              for a rejected push URL, or None to accept it. Defaults to
-              None (no library screening). The spec lists these checks as
-              SHOULD, so deployments that want the built-in policy should
-              pass ``push_url_validation_error``.
+            push_url_validator: Async callable that returns True to accept
+              a push URL, or False to reject it. Defaults to None (no
+              library screening). The spec lists these checks as SHOULD,
+              so deployments that want the built-in policy should pass
+              ``validate_push_notification_url``.
         """
         self.agent_executor = agent_executor
         self.task_store = task_store
         self._agent_card = agent_card
@@ -151,11 +150,8 @@
         """Apply the configured push-URL policy, if any."""
         if self._push_url_validator is None:
             return
-        url_error = await self._push_url_validator(url)
-        if url_error:
-            raise InvalidParamsError(
-                message=f'Invalid push notification URL: {url_error}'
-            )
+        if not await self._push_url_validator(url):
+            raise InvalidParamsError(message='Invalid push notification URL')
 
     @validate_request_params
     async def on_get_task(

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -90,18 +90,17 @@
         agent_card: AgentCard,
         queue_manager: Any
         | None = None,  # Accepted for signature compat; ignored in v2 (warns)
         push_config_store: PushNotificationConfigStore | None = None,
         push_sender: PushNotificationSender | None = None,
         request_context_builder: RequestContextBuilder | None = None,
         extended_agent_card: AgentCard | None = None,
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], Awaitable[AgentCard]
         ]
         | None = None,
-        push_url_validator: Callable[[str], Awaitable[str | None]]
-        | None = None,
+        push_url_validator: Callable[[str], Awaitable[bool]] | None = None,
     ) -> None:
         if queue_manager is not None:
             message = (
                 'A queue_manager was passed to DefaultRequestHandlerV2, but it '
                 'is not used: v2 delegates event streaming to an in-memory '
@@ -137,11 +136,8 @@
         """Apply the configured push-URL policy, if any."""
         if self._push_url_validator is None:
             return
-        url_error = await self._push_url_validator(url)
-        if url_error:
-            raise InvalidParamsError(
-                message=f'Invalid push notification URL: {url_error}'
-            )
+        if not await self._push_url_validator(url):
+            raise InvalidParamsError(message='Invalid push notification URL')
 
     async def aclose(self) -> None:
         """Shuts down the handler, draining all active tasks.

src/a2a/server/tasks/base_push_notification_sender.py

@@ -1,8 +1,7 @@
 import asyncio
-import ipaddress
 import logging
-import socket
-import urllib.parse
+
+from collections.abc import Awaitable, Callable
 
 import httpx
 
@@ -23,56 +22,6 @@
 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
-    )
-
-
-async 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.
-
-    Uses the running event-loop resolver so the default request
-    handlers stay non-blocking. Deployments can replace this with
-    their own policy via ``push_url_validator``.
-    """
-    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:
-        loop = asyncio.get_running_loop()
-        infos = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
-    except OSError:
-        return f"host '{host}' could not be resolved"
-    for info in infos:
-        if _ip_is_blocked(str(info[4][0])):
-            return f"host '{host}' resolves to a non-public address"
-    return None
-
-
 class BasePushNotificationSender(PushNotificationSender):
     """Base implementation of PushNotificationSender interface."""
 
@@ -81,6 +30,8 @@ def __init__(
         httpx_client: httpx.AsyncClient,
         config_store: PushNotificationConfigStore,
         context: ServerCallContext | None = None,
+        *,
+        push_url_validator: Callable[[str], Awaitable[bool]] | None = None,
     ) -> None:
         """Initializes the BasePushNotificationSender.
 
@@ -94,6 +45,11 @@ def __init__(
               Pass None (the default) in new code. A non-None
               value logs a deprecation warning and is otherwise
               ignored.
+            push_url_validator: Async callable that returns True to
+              accept a push URL, or False to reject it. Defaults to
+              None (no library screening). The spec lists these checks
+              as SHOULD, so deployments that want the built-in policy
+              should pass ``validate_push_notification_url``.
         """
         if context is not None:
             logger.warning(
@@ -107,6 +63,7 @@ def __init__(
             )
         self._client = httpx_client
         self._config_store = config_store
+        self._push_url_validator = push_url_validator
 
     async def send_notification(
         self, task_id: str, event: PushNotificationEvent
@@ -134,6 +91,11 @@ async def _dispatch_notification(
         task_id: str,
     ) -> bool:
         url = push_info.url
+        if (
+            self._push_url_validator is not None
+            and not await self._push_url_validator(url)
+        ):
+            return False
         try:
             headers = None
             if push_info.token:

src/a2a/utils/push_url_validator.py

@@ -0,0 +1,83 @@
+"""Shared policy for screening client-supplied push-notification URLs."""
+
+import asyncio
+import ipaddress
+import logging
+import socket
+import urllib.parse
+
+
+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
+    )
+
+
+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.
+
+    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.
+
+    Uses the running event-loop resolver so request handlers and the
+    sender stay non-blocking. Deployments can pass this function as
+    ``push_url_validator`` on ``DefaultRequestHandler`` /
+    ``DefaultRequestHandlerV2`` / ``BasePushNotificationSender``.
+    The default on those constructors is ``None`` (no library
+    screening).
+    """
+    try:
+        parsed = urllib.parse.urlparse(url)
+        explicit_port = parsed.port
+    except ValueError:
+        logger.warning('Push-notification URL is unparseable: %s', url)
+        return False
+    if parsed.scheme not in ('http', 'https'):
+        logger.warning(
+            'Push-notification URL scheme %r is not http/https: %s',
+            parsed.scheme,
+            url,
+        )
+        return False
+    host = parsed.hostname
+    if not host:
+        logger.warning('Push-notification URL has no hostname: %s', url)
+        return False
+    port = explicit_port or (443 if parsed.scheme == 'https' else 80)
+    try:
+        loop = asyncio.get_running_loop()
+        infos = await loop.getaddrinfo(host, port, type=socket.SOCK_STREAM)
+    except OSError:
+        logger.warning(
+            'Push-notification host %r could not be resolved: %s', host, url
+        )
+        return False
+    for info in infos:
+        if _ip_is_blocked(str(info[4][0])):
+            logger.warning(
+                'Push-notification host %r resolves to a non-public address: %s',
+                host,
+                url,
+            )
+            return False
+    return True

tests/server/request_handlers/test_default_request_handler.py

@@ -44,9 +44,6 @@
     TaskStore,
     TaskUpdater,
 )
-from a2a.server.tasks.base_push_notification_sender import (
-    push_url_validation_error,
-)
 from a2a.types import (
     InternalError,
     InvalidParamsError,
@@ -79,6 +76,9 @@
     TaskStatus,
     TaskStatusUpdateEvent,
 )
+from a2a.utils.push_url_validator import (
+    validate_push_notification_url,
+)
 
 
 class MockAgentExecutor(AgentExecutor):
@@ -3159,7 +3159,7 @@ async def test_on_create_task_push_notification_config_rejects_invalid_url(
         task_store=mock_task_store,
         push_config_store=push_store,
         agent_card=agent_card,
-        push_url_validator=push_url_validation_error,
+        push_url_validator=validate_push_notification_url,
     )
     context = create_server_call_context()
 
@@ -3219,7 +3219,7 @@ async def test_on_message_send_rejects_invalid_push_url(agent_card):
         task_store=mock_task_store,
         push_config_store=push_store,
         agent_card=agent_card,
-        push_url_validator=push_url_validation_error,
+        push_url_validator=validate_push_notification_url,
     )
     context = create_server_call_context()
     params = SendMessageRequest(

tests/server/tasks/test_push_notification_sender.py

@@ -1,3 +1,4 @@
+import asyncio
 import unittest
 
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -16,6 +17,7 @@
     TaskStatus,
     TaskStatusUpdateEvent,
 )
+from a2a.utils.push_url_validator import validate_push_notification_url
 from google.protobuf.json_format import MessageToDict
 
 
@@ -228,3 +230,82 @@ async def test_send_notification_artifact_update_event(self) -> None:
             json=MessageToDict(StreamResponse(artifact_update=event)),
             headers=None,
         )
+
+
+def _gai_result(ip: str, port: int = 80):
+    return [(2, 1, 6, '', (ip, port))]
+
+
+class TestPushUrlValidation(unittest.IsolatedAsyncioTestCase):
+    """SSRF hardening: when validate_push_notification_url is installed, client
+    push URLs must not reach non-public destinations."""
+
+    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,
+            push_url_validator=validate_push_notification_url,
+        )
+
+    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)
+
+    def _patch_gai(self, *, return_value=None, side_effect=None):
+        loop = asyncio.get_running_loop()
+        mock_gai = AsyncMock(return_value=return_value, side_effect=side_effect)
+        return patch.object(loop, 'getaddrinfo', mock_gai)
+
+    async def test_metadata_endpoint_blocked(self) -> None:
+        with self._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 self._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 self._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_invalid_port_blocked(self) -> None:
+        await self._dispatch('http://example.com:99999/hook')
+        self.mock_httpx_client.post.assert_not_called()
+
+    async def test_unresolvable_host_blocked_fail_closed(self) -> None:
+        with self._patch_gai(side_effect=OSError('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 self._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_default_hook_none_skips_validation(self) -> None:
+        sender = BasePushNotificationSender(
+            httpx_client=self.mock_httpx_client,
+            config_store=self.mock_config_store,
+        )
+        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()