From 5f804518b4f88f40ff446feaf9455298415d3c61 Mon Sep 17 00:00:00 2001 From: SashaMIT Date: Thu, 6 Aug 2026 04:54:47 +0700 Subject: [PATCH 1/2] fix(server): validate push-notification URLs before dispatch (SSRF hardening) A client sets its push-notification webhook URL via tasks/pushNotificationConfig (or inline on message/send), and the server then POSTs task events to that URL. The URL was used exactly as supplied - no scheme check, no destination check - so every deployment of the reference sender exposed a blind server-side request forgery primitive: point a task's push config at http://169.254.169.254/... (cloud metadata), http://localhost:PORT/admin, or any internal service and the agent server POSTs there on every task event. BasePushNotificationSender now validates each URL at dispatch time: scheme must be http/https, the host must resolve, and every resolved address must be public unicast (loopback, link-local, private, reserved, multicast, and unspecified addresses are rejected; unresolvable hosts fail closed since the POST would fail anyway). Operators whose legitimate webhooks live on private networks can opt out with allow_private_push_urls=True. Validation happens at dispatch rather than at config-write so configs registered through any path (create, inline on send, future stores) are covered by the same choke point. Residual risk, documented in the constructor docstring: DNS rebinding between validation and the POST itself remains possible for attacker-controlled domains; static internal targets are fully blocked. Tests: 7 new unit tests (metadata IP, loopback, private range, non-http scheme, unresolvable host fail-closed, public allowed, opt-out); existing suites made DNS-hermetic; push-notification e2e app opts out since its webhooks are real local servers. Signed-off-by: SashaMIT Co-authored-by: Cursor --- .../tasks/base_push_notification_sender.py | 67 +++++++++++++++ .../push_notifications/agent_app.py | 4 + .../tasks/test_inmemory_push_notifications.py | 16 ++++ .../tasks/test_push_notification_sender.py | 82 +++++++++++++++++++ 4 files changed, 169 insertions(+) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index ff9ca3ce5..5545ee56f 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,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 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 +76,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 +91,13 @@ 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). """ if context is not None: logger.warning( @@ -54,6 +111,7 @@ def __init__( ) 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 +139,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..0e277e1ef 100644 --- a/tests/server/tasks/test_inmemory_push_notifications.py +++ b/tests/server/tasks/test_inmemory_push_notifications.py @@ -67,6 +67,14 @@ class TestInMemoryPushNotifier(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient) 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, @@ -446,6 +454,14 @@ def setUp(self) -> None: 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..c77be17ac 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -42,6 +42,13 @@ class TestBasePushNotificationSender(unittest.IsolatedAsyncioTestCase): def setUp(self) -> None: self.mock_httpx_client = AsyncMock(spec=httpx.AsyncClient) 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, @@ -228,3 +235,78 @@ 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) + 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() From 9747ab2c5c916fb494f999b0907027c01951a1dc Mon Sep 17 00:00:00 2001 From: SashaMIT Date: Fri, 7 Aug 2026 00:45:28 +0700 Subject: [PATCH 2/2] Address review: reject redirect-following clients, document residual SSRF risks Per review from @kuangmi-bit: - Constructor now rejects an httpx.AsyncClient configured with follow_redirects=True. URL validation covers the initial URL only; with redirects enabled a validated public URL could 30x to an internal address and be dispatched unchecked. Failing fast at construction turns that misconfiguration into an explicit error. - push_url_validation_error docstring now documents the two residual risks: redirect targets are not re-validated (mitigated by the new guard) and DNS rebinding TOCTOU between validation and connection (documented as defense-in-depth; operators should keep network-level egress controls). - Notes that IPv4-mapped IPv6 forms are covered via ipaddress mapping. - Tests: setUp mocks pin follow_redirects=False explicitly; new test asserts the constructor guard raises on a redirect-following client. Full suite green: 1354 passed, 90 skipped, 3 xfailed. Signed-off-by: SashaMIT --- .../tasks/base_push_notification_sender.py | 33 +++++++++++++++++++ .../tasks/test_inmemory_push_notifications.py | 2 ++ .../tasks/test_push_notification_sender.py | 16 +++++++++ 3 files changed, 51 insertions(+) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index 5545ee56f..ee925567f 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -47,6 +47,23 @@ def push_url_validation_error(url: str) -> str | None: (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) @@ -98,6 +115,14 @@ def __init__( 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( @@ -109,6 +134,14 @@ 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 diff --git a/tests/server/tasks/test_inmemory_push_notifications.py b/tests/server/tasks/test_inmemory_push_notifications.py index 0e277e1ef..fac679b32 100644 --- a/tests/server/tasks/test_inmemory_push_notifications.py +++ b/tests/server/tasks/test_inmemory_push_notifications.py @@ -66,6 +66,7 @@ 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). @@ -448,6 +449,7 @@ 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 diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index c77be17ac..518aa77e9 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -41,6 +41,8 @@ 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( @@ -58,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) @@ -250,6 +264,8 @@ class TestPushUrlValidation(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() self.sender = BasePushNotificationSender( httpx_client=self.mock_httpx_client,