From 5f804518b4f88f40ff446feaf9455298415d3c61 Mon Sep 17 00:00:00 2001 From: SashaMIT Date: Thu, 6 Aug 2026 04:54:47 +0700 Subject: [PATCH 1/6] 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/6] 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, From abbb878d9db6abd585bf1d54d59526269baa1780 Mon Sep 17 00:00:00 2001 From: Sasha Mitchell Date: Thu, 27 Aug 2026 18:55:14 +0100 Subject: [PATCH 3/6] fix: clear spelling and type-check failures in URL validation --- .../server/tasks/base_push_notification_sender.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index ee925567f..144d37747 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -48,16 +48,16 @@ def push_url_validation_error(url: str) -> str | None: 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. + 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. 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-then-connect race: 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 @@ -77,10 +77,10 @@ def push_url_validation_error(url: str) -> str | None: port = parsed.port or (443 if parsed.scheme == 'https' else 80) try: infos = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) - except socket.gaierror: + except OSError: return f"host '{host}' could not be resolved" for info in infos: - if _ip_is_blocked(info[4][0]): + if _ip_is_blocked(str(info[4][0])): return f"host '{host}' resolves to a non-public address" return None From 3d67ca3bf1af0e29a34d9698c303deeb95ba3309 Mon Sep 17 00:00:00 2001 From: Sasha Mitchell Date: Wed, 2 Sep 2026 12:30:52 +0100 Subject: [PATCH 4/6] fix: treat invalid push URL ports as unparseable Drop the sender re-export so push_url_validation_error is imported from utils only. --- src/a2a/server/tasks/base_push_notification_sender.py | 2 -- src/a2a/utils/push_url_validator.py | 3 ++- tests/server/tasks/test_push_notification_sender.py | 4 ++++ 3 files changed, 6 insertions(+), 3 deletions(-) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index 199197255..f72c42b23 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -17,14 +17,12 @@ ) from a2a.types.a2a_pb2 import TaskPushNotificationConfig from a2a.utils.proto_utils import to_stream_response -from a2a.utils.push_url_validator import push_url_validation_error logger = logging.getLogger(__name__) __all__ = [ 'BasePushNotificationSender', - 'push_url_validation_error', ] diff --git a/src/a2a/utils/push_url_validator.py b/src/a2a/utils/push_url_validator.py index 82d23bde1..5a36ced49 100644 --- a/src/a2a/utils/push_url_validator.py +++ b/src/a2a/utils/push_url_validator.py @@ -44,6 +44,7 @@ async def push_url_validation_error(url: str) -> str | None: """ try: parsed = urllib.parse.urlparse(url) + explicit_port = parsed.port except ValueError: return 'unparseable URL' if parsed.scheme not in ('http', 'https'): @@ -51,7 +52,7 @@ async def push_url_validation_error(url: str) -> str | None: host = parsed.hostname if not host: return 'no hostname' - port = parsed.port or (443 if parsed.scheme == 'https' else 80) + 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) diff --git a/tests/server/tasks/test_push_notification_sender.py b/tests/server/tasks/test_push_notification_sender.py index aed3b6fb6..dff4f4e56 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -282,6 +282,10 @@ 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/') From d6e14154eb71f769b20855a97e1274a9fedf8abb Mon Sep 17 00:00:00 2001 From: Sasha Mitchell Date: Wed, 2 Sep 2026 20:01:44 +0100 Subject: [PATCH 5/6] fix: return bool from push URL validator and log rejects in place Rename to validate_push_notification_url. Drop the sender __all__ re-export. --- .../default_request_handler.py | 20 +++++------ .../default_request_handler_v2.py | 10 ++---- .../tasks/base_push_notification_sender.py | 25 ++++--------- src/a2a/utils/push_url_validator.py | 35 ++++++++++++++----- .../test_default_request_handler.py | 6 ++-- .../tasks/test_inmemory_push_notifications.py | 1 + .../tasks/test_push_notification_sender.py | 6 ++-- 7 files changed, 52 insertions(+), 51 deletions(-) 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 f72c42b23..81f9b6be8 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -21,10 +21,6 @@ logger = logging.getLogger(__name__) -__all__ = [ - 'BasePushNotificationSender', -] - class BasePushNotificationSender(PushNotificationSender): """Base implementation of PushNotificationSender interface.""" @@ -35,8 +31,7 @@ def __init__( config_store: PushNotificationConfigStore, context: ServerCallContext | None = None, *, - push_url_validator: Callable[[str], Awaitable[str | None]] - | None = None, + push_url_validator: Callable[[str], Awaitable[bool]] | None = None, ) -> None: """Initializes the BasePushNotificationSender. @@ -50,11 +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 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``. """ if context is not None: logger.warning( @@ -97,13 +92,7 @@ async def _dispatch_notification( ) -> bool: url = push_info.url if self._push_url_validator is not None: - validation_error = await self._push_url_validator(url) - if validation_error: - logger.warning( - 'Push-notification URL for task_id=%s rejected: %s', - task_id, - validation_error, - ) + if not await self._push_url_validator(url): return False try: headers = None diff --git a/src/a2a/utils/push_url_validator.py b/src/a2a/utils/push_url_validator.py index 5a36ced49..7eb3072fa 100644 --- a/src/a2a/utils/push_url_validator.py +++ b/src/a2a/utils/push_url_validator.py @@ -2,10 +2,14 @@ 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: @@ -22,8 +26,8 @@ def _ip_is_blocked(ip_str: str) -> bool: ) -async def push_url_validation_error(url: str) -> str | None: - """Return an error string if a push-notification URL is not safe. +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 @@ -46,19 +50,34 @@ async def push_url_validation_error(url: str) -> str | None: parsed = urllib.parse.urlparse(url) explicit_port = parsed.port except ValueError: - return 'unparseable URL' + logger.warning('Push-notification URL is unparseable: %s', url) + return False if parsed.scheme not in ('http', 'https'): - return f"scheme '{parsed.scheme}' is not 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: - return 'no hostname' + 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: - return f"host '{host}' could not be resolved" + 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])): - return f"host '{host}' resolves to a non-public address" - return None + 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 c59dcae33..f9a2a0fb3 100644 --- a/tests/server/request_handlers/test_default_request_handler.py +++ b/tests/server/request_handlers/test_default_request_handler.py @@ -77,7 +77,7 @@ TaskStatusUpdateEvent, ) from a2a.utils.push_url_validator import ( - push_url_validation_error, + validate_push_notification_url, ) @@ -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_inmemory_push_notifications.py b/tests/server/tasks/test_inmemory_push_notifications.py index 0fcf6650f..0a53352f8 100644 --- a/tests/server/tasks/test_inmemory_push_notifications.py +++ b/tests/server/tasks/test_inmemory_push_notifications.py @@ -531,6 +531,7 @@ def setUp(self) -> None: self.mock_httpx_client.post.return_value = mock_response self.config_store = InMemoryPushNotificationConfigStore() + 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 dff4f4e56..c77dce0cc 100644 --- a/tests/server/tasks/test_push_notification_sender.py +++ b/tests/server/tasks/test_push_notification_sender.py @@ -17,7 +17,7 @@ TaskStatus, TaskStatusUpdateEvent, ) -from a2a.utils.push_url_validator import push_url_validation_error +from a2a.utils.push_url_validator import validate_push_notification_url from google.protobuf.json_format import MessageToDict @@ -237,7 +237,7 @@ def _gai_result(ip: str, port: int = 80): class TestPushUrlValidation(unittest.IsolatedAsyncioTestCase): - """SSRF hardening: when push_url_validation_error is installed, client + """SSRF hardening: when validate_push_notification_url is installed, client push URLs must not reach non-public destinations.""" def setUp(self) -> None: @@ -246,7 +246,7 @@ def setUp(self) -> None: self.sender = BasePushNotificationSender( httpx_client=self.mock_httpx_client, config_store=self.mock_config_store, - push_url_validator=push_url_validation_error, + push_url_validator=validate_push_notification_url, ) async def _dispatch(self, url: str) -> None: From 37f6a826e013164d78eab7edf523009f980f8b70 Mon Sep 17 00:00:00 2001 From: Sasha Mitchell Date: Wed, 2 Sep 2026 20:01:55 +0100 Subject: [PATCH 6/6] fix: flatten push URL validator guard for ruff SIM102 --- src/a2a/server/tasks/base_push_notification_sender.py | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/a2a/server/tasks/base_push_notification_sender.py b/src/a2a/server/tasks/base_push_notification_sender.py index 81f9b6be8..c489581e9 100644 --- a/src/a2a/server/tasks/base_push_notification_sender.py +++ b/src/a2a/server/tasks/base_push_notification_sender.py @@ -91,9 +91,11 @@ async def _dispatch_notification( task_id: str, ) -> bool: url = push_info.url - if self._push_url_validator is not None: - if not await self._push_url_validator(url): - return False + 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: