fix(security): sanitize REST path parameters to prevent injection attacks

src/a2a/client/transports/rest.py

@@ -8,33 +8,34 @@
 
 from google.protobuf.json_format import MessageToDict, Parse, ParseDict
 
 from a2a.client.client import ClientCallContext
 from a2a.client.errors import A2AClientError
 from a2a.client.transports.base import ClientTransport
 from a2a.client.transports.http_helpers import (
     get_http_args,
     send_http_request,
     send_http_stream_request,
 )
 from a2a.types.a2a_pb2 import (
     AgentCard,
     CancelTaskRequest,
     DeleteTaskPushNotificationConfigRequest,
     GetExtendedAgentCardRequest,
     GetTaskPushNotificationConfigRequest,
     GetTaskRequest,
     ListTaskPushNotificationConfigsRequest,
     ListTaskPushNotificationConfigsResponse,
     ListTasksRequest,
     ListTasksResponse,
     SendMessageRequest,
     SendMessageResponse,
     StreamResponse,
     SubscribeToTaskRequest,
     Task,
     TaskPushNotificationConfig,
 )
 from a2a.utils.errors import A2A_REASON_TO_ERROR, MethodNotFoundError
+from a2a.utils.sanitizers import sanitize_path_id
 from a2a.utils.telemetry import SpanKind, trace_class
 
 
@@ -149,7 +150,7 @@
 
         response_data = await self._execute_request(
             'GET',
-            f'/tasks/{request.id}',
+            f'/tasks/{sanitize_path_id(request.id)}',
             request.tenant,
             context=context,
             params=params,
@@ -189,7 +190,7 @@
         """Requests the agent to cancel a specific task."""
         response_data = await self._execute_request(
             'POST',
-            f'/tasks/{request.id}:cancel',
+            f'/tasks/{sanitize_path_id(request.id)}:cancel',
             request.tenant,
             context=context,
             json=MessageToDict(request),
@@ -206,7 +207,7 @@
         """Sets or updates the push notification configuration for a specific task."""
         response_data = await self._execute_request(
             'POST',
-            f'/tasks/{request.task_id}/pushNotificationConfigs',
+            f'/tasks/{sanitize_path_id(request.task_id)}/pushNotificationConfigs',
             request.tenant,
             context=context,
             json=MessageToDict(request),
@@ -233,7 +234,7 @@
 
         response_data = await self._execute_request(
             'GET',
-            f'/tasks/{request.task_id}/pushNotificationConfigs/{request.id}',
+            f'/tasks/{sanitize_path_id(request.task_id)}/pushNotificationConfigs/{sanitize_path_id(request.id, "push_id")}',
             request.tenant,
             context=context,
             params=params,
@@ -258,7 +259,7 @@
 
         response_data = await self._execute_request(
             'GET',
-            f'/tasks/{request.task_id}/pushNotificationConfigs',
+            f'/tasks/{sanitize_path_id(request.task_id)}/pushNotificationConfigs',
             request.tenant,
             context=context,
             params=params,
@@ -285,7 +286,7 @@
 
         await self._execute_request(
             'DELETE',
-            f'/tasks/{request.task_id}/pushNotificationConfigs/{request.id}',
+            f'/tasks/{sanitize_path_id(request.task_id)}/pushNotificationConfigs/{sanitize_path_id(request.id, "push_id")}',
             request.tenant,
             context=context,
             params=params,
@@ -300,7 +301,7 @@
         """Reconnects to get task updates."""
         async for event in self._send_stream_request(
             'POST',
-            f'/tasks/{request.id}:subscribe',
+            f'/tasks/{sanitize_path_id(request.id)}:subscribe',
             request.tenant,
             context=context,
         ):

src/a2a/server/routes/rest_dispatcher.py

@@ -27,6 +27,7 @@
     InvalidRequestError,
     TaskNotFoundError,
 )
+from a2a.utils.sanitizers import sanitize_path_id
 from a2a.utils.telemetry import SpanKind, trace_class
 from a2a.utils.version_validator import validate_version
 
@@ -200,7 +201,7 @@
 
         @validate_version(constants.PROTOCOL_VERSION_1_0)
         async def _handler(context: ServerCallContext) -> a2a_pb2.Task:
-            task_id = request.path_params['id']
+            task_id = sanitize_path_id(request.path_params['id'])
             task = await self.request_handler.on_cancel_task(
                 CancelTaskRequest(id=task_id), context
             )
@@ -216,7 +217,7 @@
         self, request: Request
     ) -> EventSourceResponse:
         """Handles the 'SubscribeToTask' REST method."""
-        task_id = request.path_params['id']
+        task_id = sanitize_path_id(request.path_params['id'])
 
         @validate_version(constants.PROTOCOL_VERSION_1_0)
         async def _handler(
@@ -238,7 +239,7 @@
         async def _handler(context: ServerCallContext) -> a2a_pb2.Task:
             params = a2a_pb2.GetTaskRequest()
             proto_utils.parse_params(request.query_params, params)
-            params.id = request.path_params['id']
+            params.id = sanitize_path_id(request.path_params['id'])
             task = await self.request_handler.on_get_task(params, context)
             if task:
                 return task
@@ -255,8 +256,10 @@
         async def _handler(
             context: ServerCallContext,
         ) -> a2a_pb2.TaskPushNotificationConfig:
-            task_id = request.path_params['id']
-            push_id = request.path_params['push_id']
+            task_id = sanitize_path_id(request.path_params['id'])
+            push_id = sanitize_path_id(
+                request.path_params['push_id'], 'push_id'
+            )
             params = GetTaskPushNotificationConfigRequest(
                 task_id=task_id, id=push_id
             )
@@ -275,8 +278,10 @@
 
         @validate_version(constants.PROTOCOL_VERSION_1_0)
         async def _handler(context: ServerCallContext) -> None:
-            task_id = request.path_params['id']
-            push_id = request.path_params['push_id']
+            task_id = sanitize_path_id(request.path_params['id'])
+            push_id = sanitize_path_id(
+                request.path_params['push_id'], 'push_id'
+            )
             params = a2a_pb2.DeleteTaskPushNotificationConfigRequest(
                 task_id=task_id, id=push_id
             )
@@ -298,7 +303,7 @@
             body = await request.body()
             params = a2a_pb2.TaskPushNotificationConfig()
             Parse(body, params)
-            params.task_id = request.path_params['id']
+            params.task_id = sanitize_path_id(request.path_params['id'])
             return await self.request_handler.on_create_task_push_notification_config(
                 params, context
             )
@@ -316,7 +321,7 @@
         ) -> a2a_pb2.ListTaskPushNotificationConfigsResponse:
             params = a2a_pb2.ListTaskPushNotificationConfigsRequest()
             proto_utils.parse_params(request.query_params, params)
-            params.task_id = request.path_params['id']
+            params.task_id = sanitize_path_id(request.path_params['id'])
             return await self.request_handler.on_list_task_push_notification_configs(
                 params, context
             )

src/a2a/utils/__init__.py

@@ -7,12 +7,14 @@
     TransportProtocol,
 )
 from a2a.utils.proto_utils import to_stream_response
+from a2a.utils.sanitizers import sanitize_path_id
 
 
 __all__ = [
     'AGENT_CARD_WELL_KNOWN_PATH',
     'DEFAULT_RPC_URL',
     'TransportProtocol',
     'proto_utils',
+    'sanitize_path_id',
     'to_stream_response',
 ]

src/a2a/utils/sanitizers.py

@@ -0,0 +1,56 @@
+"""Input sanitization utilities for A2A path parameters.
+
+This module provides validation functions for resource IDs used in
+REST URL paths, preventing injection of control characters, path
+traversal sequences, and other unsafe inputs.
+"""
+
+import re
+
+from a2a.utils.errors import InvalidRequestError
+
+
+# Allowed characters for path resource IDs: alphanumeric, hyphen,
+# underscore, and period.  This matches the character set typically
+# used for UUIDs, task IDs, and push-notification config IDs.
+_PATH_ID_PATTERN = re.compile(r'^[A-Za-z0-9._-]+$')
+
+# ASCII control character boundaries.
+_MAX_PRINTABLE_ASCII = 0x20  # First non-control character (space)
+_DEL_ASCII = 0x7F  # DEL control character
+
+
+def sanitize_path_id(value: str, param_name: str = 'id') -> str:
+    """Validate and sanitize a path parameter used as a resource ID.
+
+    Rejects values containing null bytes, newlines, other control
+    characters, or any characters outside the safe set
+    ``[A-Za-z0-9._-]``.
+
+    Args:
+        value: The raw path parameter value.
+        param_name: Name of the parameter (for error messages).
+
+    Returns:
+        The validated value unchanged.
+
+    Raises:
+        InvalidRequestError: If the value contains disallowed characters
+            or is empty.
+    """
+    if not value:
+        raise InvalidRequestError(
+            message=f'{param_name} must not be empty',
+        )
+    # Reject null bytes and other control characters (0x00-0x1F, 0x7F).
+    if any(
+        ord(c) < _MAX_PRINTABLE_ASCII or ord(c) == _DEL_ASCII for c in value
+    ):
+        raise InvalidRequestError(
+            message=f'{param_name} contains control characters',
+        )
+    if not _PATH_ID_PATTERN.match(value):
+        raise InvalidRequestError(
+            message=f'{param_name} contains invalid characters: {value!r}',
+        )
+    return value

tests/server/routes/test_rest_dispatcher.py

@@ -323,3 +323,91 @@ async def stream_with_non_ascii(
                 payload = payload.decode('utf-8')
             assert non_ascii_text in payload
             assert '\\u4f60\\u597d' not in payload
+
+
+@pytest.mark.asyncio
+class TestPathIdSanitization:
+    """Regression tests for https://github.com/a2aproject/a2a-python/issues/805.
+
+    Verify that malicious path parameters (null bytes, control characters,
+    path traversal) are rejected at the REST dispatcher layer.
+    """
+
+    @pytest.fixture
+    def dispatcher(self, mock_handler: AsyncMock) -> RestDispatcher:
+        return RestDispatcher(request_handler=mock_handler)
+
+    async def test_get_task_rejects_null_byte_id(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Null byte in task ID must be rejected."""
+        req = make_mock_request(method='GET', path_params={'id': 'abc\x00def'})
+        response = await dispatcher.on_get_task(req)
+        assert response.status_code == 400
+
+    async def test_get_task_rejects_newline_id(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Newline in task ID must be rejected."""
+        req = make_mock_request(method='GET', path_params={'id': 'abc\ndef'})
+        response = await dispatcher.on_get_task(req)
+        assert response.status_code == 400
+
+    async def test_get_task_rejects_path_traversal_id(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Path traversal in task ID must be rejected."""
+        req = make_mock_request(
+            method='GET', path_params={'id': '../../etc/passwd'}
+        )
+        response = await dispatcher.on_get_task(req)
+        assert response.status_code == 400
+
+    async def test_get_task_rejects_space_id(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Space in task ID must be rejected."""
+        req = make_mock_request(method='GET', path_params={'id': 'abc def'})
+        response = await dispatcher.on_get_task(req)
+        assert response.status_code == 400
+
+    async def test_cancel_task_rejects_null_byte_id(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Null byte in task ID on cancel must be rejected."""
+        req = make_mock_request(method='POST', path_params={'id': 'abc\x00def'})
+        response = await dispatcher.on_cancel_task(req)
+        assert response.status_code == 400
+
+    async def test_get_push_notification_rejects_malicious_ids(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Malicious task_id and push_id must be rejected."""
+        req = make_mock_request(
+            method='GET',
+            path_params={'id': 'abc\x00def', 'push_id': 'valid-id'},
+        )
+        response = await dispatcher.get_push_notification(req)
+        assert response.status_code == 400
+
+    async def test_get_push_notification_rejects_malicious_push_id(
+        self, dispatcher: RestDispatcher
+    ) -> None:
+        """Malicious push_id must be rejected even with valid task_id."""
+        req = make_mock_request(
+            method='GET',
+            path_params={'id': 'valid-task-id', 'push_id': '../evil'},
+        )
+        response = await dispatcher.get_push_notification(req)
+        assert response.status_code == 400
+
+    async def test_valid_id_passes(
+        self, dispatcher: RestDispatcher, mock_handler: AsyncMock
+    ) -> None:
+        """Valid task ID passes sanitization and reaches the handler."""
+        req = make_mock_request(
+            method='GET', path_params={'id': 'valid-task-123'}
+        )
+        response = await dispatcher.on_get_task(req)
+        # Handler returns a task, so status should be 200
+        assert response.status_code == 200