diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index ff9ca3ce5..ee925567f 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/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,68 @@ 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. + + IPv4-mapped IPv6 forms (e.g. ``::ffff:127.0.0.1``) are covered: + ``ipaddress`` maps them to the underlying IPv4 address, so the + ``is_private``/``is_loopback`` checks apply to the mapped value. + + Known limitations: + * Validation covers the initial URL only. Redirect responses are + not re-validated, so this check is only sound with + ``follow_redirects=False`` (the httpx default, and the value + ``BasePushNotificationSender`` now asserts on its client). + * DNS rebinding (TOCTOU): validation and the actual connection + resolve the hostname separately, so a hostile DNS server can + answer the validation query with a public address and the + connection query with a private one. Fully closing this would + require pinning the validated address in the HTTP transport; + until then, operators should treat this as defense-in-depth + and keep network-level egress controls in place. + """ + 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 +93,8 @@ def __init__( httpx_client: httpx.AsyncClient, config_store: PushNotificationConfigStore, context: ServerCallContext | None = None, + *, + allow_private_push_urls: bool = False, ) -> None: """Initializes the BasePushNotificationSender. @@ -41,6 +108,21 @@ def __init__( 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). + + Note: + URL validation covers the initial request URL only. If the + client follows redirects, a validated public URL can + redirect to an internal address unchecked, so + ``follow_redirects`` must stay disabled (the httpx + default). This constructor rejects clients configured + otherwise. """ if context is not None: logger.warning( @@ -52,8 +134,17 @@ def __init__( 'caller identity is not carried into dispatch. Drop the ' 'context argument from the constructor call.' ) + if httpx_client.follow_redirects: + raise ValueError( + 'BasePushNotificationSender validates the initial push URL ' + 'only; a client with follow_redirects=True would dispatch ' + 'redirect targets without re-validation (redirect-based ' + 'SSRF). Construct the client with follow_redirects=False ' + '(the default).' + ) 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 +172,15 @@ async def _dispatch_notification( 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: diff --git a/tests/integration/push_notifications/agent_app.py b/tests/integration/push_notifications/agent_app.py index e704c2be9..99eec7fdb 100644 --- a/tests/integration/push_notifications/agent_app.py +++ b/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, ), ) diff --git a/tests/server/tasks/test_inmemory_push_notifications.py b/tests/server/tasks/test_inmemory_push_notifications.py index f204e2181..fac679b32 100644 --- a/tests/server/tasks/test_inmemory_push_notifications.py +++ b/tests/server/tasks/test_inmemory_push_notifications.py @@ -66,7 +66,16 @@ def user_name(self) -> str: class TestInMemoryPushNotifier(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient) + self.mock_httpx_client.follow_redirects = False 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, @@ -440,12 +449,21 @@ class TestPushNotificationDispatchAcrossOwners( def setUp(self) -> None: self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient) + self.mock_httpx_client.follow_redirects = False mock_response = AsyncMock(spec=httpx.Response) mock_response.status_code = 200 self.mock_httpx_client.post.return_value = mock_response 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, diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index 990f6c7f5..518aa77e9 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -41,7 +41,16 @@ def _create_sample_push_config( class TestBasePushNotificationSender(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient) + # The sender rejects clients with follow_redirects enabled. + self.mock_httpx_client.follow_redirects = False 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, @@ -51,6 +60,18 @@ def test_constructor_stores_client_and_config_store(self) -> None: self.assertEqual(self.sender._client, self.mock_httpx_client) self.assertEqual(self.sender._config_store, self.mock_config_store) + def test_constructor_rejects_redirect_following_client(self) -> None: + # Redirect targets are dispatched without re-validation, so a + # redirect-following client reopens the SSRF hole the URL + # validation closes. + redirecting_client = AsyncMock(spec=httpx.AsyncClient) + redirecting_client.follow_redirects = True + with self.assertRaises(ValueError): + BasePushNotificationSender( + httpx_client=redirecting_client, + config_store=self.mock_config_store, + ) + async def test_send_notification_success(self) -> None: task_id = 'task_send_success' task_data = _create_sample_task(task_id=task_id) @@ -228,3 +249,80 @@ 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) + # The sender rejects clients with follow_redirects enabled. + self.mock_httpx_client.follow_redirects = False + 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()