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

src/a2a/server/request_handlers/default_request_handler.py

@@ -29,25 +29,28 @@
     PushNotificationSender,
     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,
@@ -94,35 +97,40 @@
         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,
+        allow_private_push_urls: bool = False,
     ) -> 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.
+            allow_private_push_urls: Skip SSRF screening of push-notification URLs at config creation. For local development and tests only. Defaults to False.
         """
         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
+        # Opt-in for local development/tests only: skips SSRF screening of
+        # push-notification URLs at config creation.
+        self._allow_private_push_urls = allow_private_push_urls
         self.extended_agent_card = extended_agent_card
         self.extended_card_modifier = extended_card_modifier
         self._request_context_builder = (
@@ -519,30 +527,37 @@
 
         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
 
+        if not self._allow_private_push_urls and (
+            url_error := push_url_validation_error(params.url)
+        ):
+            raise InvalidParamsError(
+                message=f'Invalid push notification URL: {url_error}'
+            )
+
         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:

src/a2a/server/request_handlers/default_request_handler_v2.py

@@ -19,25 +19,28 @@
 from a2a.server.request_handlers.request_handler import (
     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,
@@ -83,20 +86,24 @@
         agent_card: AgentCard,
         queue_manager: Any
         | None = None,  # Kept for backward compat in signature
         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,
+        allow_private_push_urls: bool = False,
     ) -> 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
+        # Opt-in for local development/tests only: skips SSRF screening of
+        # push-notification URLs at config creation.
+        self._allow_private_push_urls = allow_private_push_urls
         self.extended_agent_card = extended_agent_card
         self.extended_card_modifier = extended_card_modifier
         self._request_context_builder = (
@@ -337,29 +344,36 @@
         params: TaskPushNotificationConfig,
         context: ServerCallContext,
     ) -> TaskPushNotificationConfig:
         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
 
+        if not self._allow_private_push_urls and (
+            url_error := push_url_validation_error(params.url)
+        ):
+            raise InvalidParamsError(
+                message=f'Invalid push notification URL: {url_error}'
+            )
+
         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(  # noqa: D102
         self,
         params: GetTaskPushNotificationConfigRequest,
         context: ServerCallContext,

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 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).
+        allow_private_push_urls=True,
     )
     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).
+        allow_private_push_urls=True,
     )
 
     rest_routes = create_rest_routes(

tests/server/request_handlers/test_default_request_handler.py

@@ -555,7 +555,7 @@ async def test_on_message_send_with_push_notification(agent_card):
         agent_card=agent_card,
     )
 
-    push_config = TaskPushNotificationConfig(url='http://callback.com/push')
+    push_config = TaskPushNotificationConfig(url='http://example.com/push')
     message_config = SendMessageConfiguration(
         task_push_notification_config=push_config,
         accepted_output_modes=['text/plain'],  # Added required field
@@ -663,7 +663,7 @@ async def test_on_message_send_with_push_notification_in_non_blocking_request(
     )
 
     # Configure push notification
-    push_config = TaskPushNotificationConfig(url='http://callback.com/push')
+    push_config = TaskPushNotificationConfig(url='http://example.com/push')
     message_config = SendMessageConfiguration(
         task_push_notification_config=push_config,
         accepted_output_modes=['text/plain'],
@@ -790,7 +790,7 @@ async def test_on_message_send_with_push_notification_no_existing_Task(
         agent_card=agent_card,
     )
 
-    push_config = TaskPushNotificationConfig(url='http://callback.com/push')
+    push_config = TaskPushNotificationConfig(url='http://example.com/push')
     message_config = SendMessageConfiguration(
         task_push_notification_config=push_config,
         accepted_output_modes=['text/plain'],  # Added required field
@@ -1233,9 +1233,7 @@ async def test_on_message_send_stream_with_push_notification(agent_card):
         agent_card=agent_card,
     )
 
-    push_config = TaskPushNotificationConfig(
-        url='http://callback.stream.com/push'
-    )
+    push_config = TaskPushNotificationConfig(url='http://example.com/push')
     message_config = SendMessageConfiguration(
         task_push_notification_config=push_config,
         accepted_output_modes=['text/plain'],  # Added required field
@@ -2028,7 +2026,7 @@ async def test_get_task_push_notification_config_info_with_config(agent_card):
     )
 
     set_config_params = TaskPushNotificationConfig(
-        task_id='task_1', id='config_id', url='http://1.example.com'
+        task_id='task_1', id='config_id', url='http://example.com'
     )
     context = create_server_call_context()
     await request_handler.on_create_task_push_notification_config(
@@ -2070,7 +2068,7 @@ async def test_get_task_push_notification_config_info_with_config_no_id(
 
     set_config_params = TaskPushNotificationConfig(
         task_id='task_1',
-        url='http://1.example.com',
+        url='http://example.com',
     )
     await request_handler.on_create_task_push_notification_config(
         set_config_params, create_server_call_context()
@@ -2304,15 +2302,15 @@ async def test_list_task_push_notification_config_info_with_config_and_no_id(
     # multiple calls without config id should replace the existing
     set_config_params1 = TaskPushNotificationConfig(
         task_id='task_1',
-        url='http://1.example.com',
+        url='http://example.com',
     )
     await request_handler.on_create_task_push_notification_config(
         set_config_params1, create_server_call_context()
     )
 
     set_config_params2 = TaskPushNotificationConfig(
         task_id='task_1',
-        url='http://2.example.com',
+        url='http://example.com',
     )
     await request_handler.on_create_task_push_notification_config(
         set_config_params2, create_server_call_context()
@@ -2947,7 +2945,7 @@ async def test_on_create_task_push_notification_config_unsupported(agent_card):
         agent_card=agent_card,
     )
 
-    params = TaskPushNotificationConfig(url='http://callback.com/push')
+    params = TaskPushNotificationConfig(url='http://example.com/push')
 
     context = create_server_call_context()
 
@@ -3143,3 +3141,40 @@ async def test_on_get_task_push_notification_config_is_owner_scoped(
             ),
             _ctx('bob'),
         )
+
+
+@pytest.mark.asyncio
+async def test_on_create_task_push_notification_config_rejects_invalid_url(
+    agent_card,
+):
+    """Test on_create_task_push_notification_config rejects non-public URLs."""
+    mock_task_store = AsyncMock(spec=TaskStore)
+    mock_task_store.get.return_value = Task(id='task_1', context_id='ctx_1')
+    push_store = InMemoryPushNotificationConfigStore()
+    request_handler = DefaultRequestHandler(
+        agent_executor=MockAgentExecutor(),
+        task_store=mock_task_store,
+        push_config_store=push_store,
+        agent_card=agent_card,
+    )
+    context = create_server_call_context()
+
+    set_config_params = TaskPushNotificationConfig(
+        task_id='task_1', id='config_id', url='http://127.0.0.1/hook'
+    )
+    with pytest.raises(
+        InvalidParamsError, match='Invalid push notification URL'
+    ):
+        await request_handler.on_create_task_push_notification_config(
+            set_config_params, context
+        )
+
+    set_config_params = TaskPushNotificationConfig(
+        task_id='task_1', id='config_id', url='file:///etc/passwd'
+    )
+    with pytest.raises(
+        InvalidParamsError, match='Invalid push notification URL'
+    ):
+        await request_handler.on_create_task_push_notification_config(
+            set_config_params, context
+        )

tests/server/request_handlers/test_default_request_handler_v2.py

@@ -524,7 +524,7 @@ async def test_get_task_push_notification_config_info_with_config():
         agent_card=create_default_agent_card(),
     )
     set_config_params = TaskPushNotificationConfig(
-        task_id='task_1', id='config_id', url='http://1.example.com'
+        task_id='task_1', id='config_id', url='http://example.com'
     )
     context = create_server_call_context()
     await request_handler.on_create_task_push_notification_config(
@@ -557,7 +557,7 @@ async def test_get_task_push_notification_config_info_with_config_no_id():
         agent_card=create_default_agent_card(),
     )
     set_config_params = TaskPushNotificationConfig(
-        task_id='task_1', url='http://1.example.com'
+        task_id='task_1', url='http://example.com'
     )
     await request_handler.on_create_task_push_notification_config(
         set_config_params, create_server_call_context()
@@ -740,13 +740,13 @@ async def test_list_task_push_notification_config_info_with_config_and_no_id():
         agent_card=create_default_agent_card(),
     )
     set_config_params1 = TaskPushNotificationConfig(
-        task_id='task_1', url='http://1.example.com'
+        task_id='task_1', url='http://example.com'
     )
     await request_handler.on_create_task_push_notification_config(
         set_config_params1, create_server_call_context()
     )
     set_config_params2 = TaskPushNotificationConfig(
-        task_id='task_1', url='http://2.example.com'
+        task_id='task_1', url='http://example.com'
     )
     await request_handler.on_create_task_push_notification_config(
         set_config_params2, create_server_call_context()