fix(server): validate push-notification URLs at config creation

src/a2a/server/request_handlers/default_request_handler.py

@@ -30,25 +30,28 @@
     ResultAggregator,
     TaskManager,
     TaskStore,
 )
+from a2a.server.tasks.base_push_notification_sender import (
+    push_url_validation_error,
+)
 from a2a.types.a2a_pb2 import (
     AgentCard,
     CancelTaskRequest,
     DeleteTaskPushNotificationConfigRequest,
     GetExtendedAgentCardRequest,
     GetTaskPushNotificationConfigRequest,
     GetTaskRequest,
     ListTaskPushNotificationConfigsRequest,
     ListTaskPushNotificationConfigsResponse,
     ListTasksRequest,
     ListTasksResponse,
     Message,
     SendMessageRequest,
     SubscribeToTaskRequest,
     Task,
     TaskPushNotificationConfig,
     TaskState,
 )
 from a2a.utils.errors import (
     ExtendedAgentCardNotConfiguredError,
     InternalError,
@@ -95,34 +98,42 @@
         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 = push_url_validation_error,
     ) -> 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
+              ``push_url_validation_error``. Pass None to skip library
+              screening (the spec lists these checks as SHOULD, so the
+              policy is deployment-specific).
         """
         self.agent_executor = agent_executor
         self.task_store = task_store
         self._agent_card = agent_card
         self._queue_manager = queue_manager or InMemoryQueueManager()
         self._push_config_store = push_config_store
         self._push_sender = push_sender
+        self._push_url_validator = push_url_validator
         self.extended_agent_card = extended_agent_card
         self.extended_card_modifier = extended_card_modifier
         self._request_context_builder = (
@@ -137,9 +148,19 @@
         # Tracks background tasks (e.g., deferred cleanups) to avoid orphaning
         # asyncio tasks and to surface unexpected exceptions.
         self._background_tasks = set()
 
+    async def _reject_unsafe_push_url(self, url: str) -> None:
+        """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}'
+            )
+
     @validate_request_params
     async def on_get_task(
         self,
         params: GetTaskRequest,
         context: ServerCallContext,
@@ -304,6 +325,9 @@
         if self._push_config_store and params.configuration.HasField(
             'task_push_notification_config'
         ):
+            await self._reject_unsafe_push_url(
+                params.configuration.task_push_notification_config.url
+            )
             await self._push_config_store.set_info(
                 task_id,
                 params.configuration.task_push_notification_config,
@@ -520,30 +544,32 @@
         Requires a `PushNotifier` to be configured.
         """
         if not self._push_config_store:
             raise PushNotificationNotSupportedError
 
         task_id = params.task_id
         task: Task | None = await self.task_store.get(task_id, context)
         if not task:
             raise TaskNotFoundError
 
+        await self._reject_unsafe_push_url(params.url)
+
         await self._push_config_store.set_info(
             task_id,
             params,
             context,
         )
 
         return params
 
     @validate_request_params
     @validate(
         lambda self: self._agent_card.capabilities.push_notifications,
         error_message='Push notifications are not supported by the agent',
         error_type=PushNotificationNotSupportedError,
     )
     async def on_get_task_push_notification_config(
         self,
         params: GetTaskPushNotificationConfigRequest,
         context: ServerCallContext,
     ) -> TaskPushNotificationConfig:
         """Default handler for 'tasks/pushNotificationConfig/get'.

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -20,25 +20,28 @@
     RequestHandler,
     validate,
     validate_request_params,
 )
+from a2a.server.tasks.base_push_notification_sender import (
+    push_url_validation_error,
+)
 from a2a.types.a2a_pb2 import (
     AgentCard,
     CancelTaskRequest,
     DeleteTaskPushNotificationConfigRequest,
     GetExtendedAgentCardRequest,
     GetTaskPushNotificationConfigRequest,
     GetTaskRequest,
     ListTaskPushNotificationConfigsRequest,
     ListTaskPushNotificationConfigsResponse,
     ListTasksRequest,
     ListTasksResponse,
     Message,
     SendMessageRequest,
     SubscribeToTaskRequest,
     Task,
     TaskPushNotificationConfig,
 )
 from a2a.utils.errors import (
     ExtendedAgentCardNotConfiguredError,
     InternalError,
     InvalidParamsError,
@@ -90,15 +93,18 @@
         extended_card_modifier: Callable[
             [AgentCard, ServerCallContext], Awaitable[AgentCard]
         ]
         | None = None,
+        push_url_validator: Callable[[str], Awaitable[str | None]]
+        | None = push_url_validation_error,
     ) -> None:
         self.agent_executor = agent_executor
         self.task_store = task_store
         self._agent_card = agent_card
         self._push_config_store = push_config_store
         self._push_sender = push_sender
+        self._push_url_validator = push_url_validator
         self.extended_agent_card = extended_agent_card
         self.extended_card_modifier = extended_card_modifier
         self._request_context_builder = (
             request_context_builder
             or SimpleRequestContextBuilder(
@@ -112,6 +118,16 @@
         )
         self._background_tasks = set()
 
+    async def _reject_unsafe_push_url(self, url: str) -> None:
+        """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}'
+            )
+
     async def aclose(self) -> None:
         """Shuts down the handler, draining all active tasks.
 
@@ -220,6 +236,9 @@
         if self._push_config_store and params.configuration.HasField(
             'task_push_notification_config'
         ):
+            await self._reject_unsafe_push_url(
+                params.configuration.task_push_notification_config.url
+            )
             await self._push_config_store.set_info(
                 task_id,
                 params.configuration.task_push_notification_config,
@@ -345,6 +364,8 @@
         if not task:
             raise TaskNotFoundError
 
+        await self._reject_unsafe_push_url(params.url)
+
         await self._push_config_store.set_info(
             task_id,
             params,

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,56 @@
 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."""
 

tests/integration/cross_version/client_server/client_1_0.py

@@ -227,7 +227,7 @@ async def test_push_notification_lifecycle(client, task_id, server_name):
 
     # 1. Create
     task_push_cfg = TaskPushNotificationConfig(
-        task_id=task_id, id=config_id, url='http://127.0.0.1:9999/webhook'
+        task_id=task_id, id=config_id, url='http://example.com/webhook'
     )
 
     created = await client.create_task_push_notification_config(

tests/integration/push_notifications/agent_app.py

@@ -155,6 +155,8 @@ def create_agent_app(
             httpx_client=notification_client,
             config_store=push_config_store,
         ),
+        # e2e webhooks are real local test servers (loopback).
+        push_url_validator=None,
     )
     rest_routes = create_rest_routes(request_handler=handler)
     agent_card_routes = create_agent_card_routes(
@@ -226,6 +228,8 @@ def create_multi_user_agent_app(
             httpx_client=notification_client,
             config_store=push_config_store,
         ),
+        # e2e webhooks are real local test servers (loopback).
+        push_url_validator=None,
     )
 
     rest_routes = create_rest_routes(