diff --git a/src/a2a/server/request_handlers/default_request_handler.py b/src/a2a/server/request_handlers/default_request_handler.py index eb179d0ec..384fd5e85 100644 --- a/src/a2a/server/request_handlers/default_request_handler.py +++ b/src/a2a/server/request_handlers/default_request_handler.py @@ -103,8 +103,7 @@ def __init__( # noqa: PLR0913 [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. @@ -119,11 +118,11 @@ def __init__( # noqa: PLR0913 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 @@ -151,11 +150,8 @@ 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}' - ) + if not await self._push_url_validator(url): + raise InvalidParamsError(message='Invalid push notification URL') @validate_request_params async def on_get_task( diff --git a/src/a2a/server/request_handlers/default_request_handler_v2.py b/src/a2a/server/request_handlers/default_request_handler_v2.py index 2f25e4f38..4942ecfef 100644 --- a/src/a2a/server/request_handlers/default_request_handler_v2.py +++ b/src/a2a/server/request_handlers/default_request_handler_v2.py @@ -98,8 +98,7 @@ def __init__( # noqa: PLR0913 [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 = ( @@ -137,11 +136,8 @@ 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}' - ) + 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. diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index 304128bac..c489581e9 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/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: diff --git a/src/a2a/utils/push_url_validator.py b/src/a2a/utils/push_url_validator.py new file mode 100644 index 000000000..7eb3072fa --- /dev/null +++ b/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 diff --git a/tests/server/request_handlers/test_default_request_handler.py b/tests/server/request_handlers/test_default_request_handler.py index 5377895f3..f9a2a0fb3 100644 --- a/tests/server/request_handlers/test_default_request_handler.py +++ b/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( diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index 990f6c7f5..c77dce0cc 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/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()