diff --git a/roboco/api/routes/a2a.py b/roboco/api/routes/a2a.py index 97168be4..eb427859 100644 --- a/roboco/api/routes/a2a.py +++ b/roboco/api/routes/a2a.py @@ -23,7 +23,12 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Request, status from fastapi.responses import JSONResponse from sse_starlette import EventSourceResponse -from roboco.api.deps import CurrentAgentSlug, DbSession +from roboco.api.deps import ( + CurrentAgentContext, + CurrentAgentSlug, + DbSession, + require_pm_or_above, +) from roboco.api.routes.v1._role_dep import require_any_authenticated_agent from roboco.api.schemas.a2a_chat import ( ConversationCloseRequest, @@ -116,6 +121,7 @@ async def get_agent_card( async def send_message( request: SendMessageRequest, db: DbSession, + agent: CurrentAgentContext, ) -> dict[str, Any]: """ Send an A2A message (fallback endpoint). @@ -149,10 +155,15 @@ async def send_message( is_response = metadata.get("is_response", False) if is_response: - # Update existing task with response message - responder = metadata.get("from_agent") + # The responder is the AUTHENTICATED caller — never a client-supplied + # metadata.from_agent, which any caller could spoof to impersonate + # anyone (e.g. from_agent='ceo') in the task's notes and in the + # spawn/notification routed back to the original requester (#116). + responder = agent.slug try: - await service.update_task_from_message(task_id_str, message, responder) + await service.update_task_from_message( + task_id_str, message, responder_agent=responder + ) except ValueError as e: error_msg = str(e) if "Invalid task ID" in error_msg: @@ -420,23 +431,32 @@ async def list_tasks( ) -@router.post("/tasks/{task_id}/cancel") +@router.post( + "/tasks/{task_id}/cancel", + dependencies=[require_any_authenticated_agent], +) async def cancel_task( task_id: str, db: DbSession, + agent: CurrentAgentContext, request: CancelTaskRequest | None = None, ) -> A2ATask: """ Cancel an A2A task. - Transitions the task to cancelled state. + Transitions the task to cancelled state. PM/management-only — the service + cascades the cancel to all non-terminal descendants, and the lifecycle rule + (Any -> cancelled: PM roles only) must hold on this path too (#423). """ + require_pm_or_above(agent.role, action="cancel a task via A2A") service = A2AService(db) try: task = await service.cancel_task( task_id=task_id, reason=request.reason if request else None, + agent_role=agent.role.value, + actor_slug=agent.slug, ) except ValueError as e: error_msg = str(e) diff --git a/roboco/services/a2a.py b/roboco/services/a2a.py index 47b80651..8f858666 100644 --- a/roboco/services/a2a.py +++ b/roboco/services/a2a.py @@ -379,13 +379,26 @@ class A2AService: return [self.task_to_a2a(t) for t in tasks], has_more - async def cancel_task(self, task_id: str, reason: str | None = None) -> A2ATask: + async def cancel_task( + self, + task_id: str, + reason: str | None = None, + agent_role: str | None = None, + actor_slug: str | None = None, + ) -> A2ATask: """ Cancel a task and all non-terminal descendants. Args: task_id: Task UUID string reason: Optional cancellation reason + agent_role: The authenticated caller's role, threaded into the + cascade role gate (TaskService.cancel). Defaults to cell_pm + when unset for back-compat with non-route callers. + actor_slug: The authenticated caller's slug, recorded in the + cancellation note so the audit trail attributes the cancel to + the real actor (the route enforces PM/management; non-route + callers may omit it). Returns: Updated A2ATask @@ -419,23 +432,38 @@ class A2AService: if status_value in ["completed", "cancelled"]: raise ValueError(f"Task already in terminal state: {status_value}") - # Add reason to notes before cancel + # Build a cancellation note that attributes the real actor (the route + # now passes the authenticated slug) so the audit trail records who + # cancelled and why — keeps this out of route handlers. + note_parts: list[str] = [] + if actor_slug: + note_parts.append(f"Cancelled via A2A by {actor_slug}") if reason: - reason_text = f"Cancellation reason: {reason}" + note_parts.append(f"reason: {reason}") + cancellation_note = "; ".join(note_parts) if note_parts else None + if cancellation_note: if task.dev_notes: - task.dev_notes = f"{task.dev_notes}\n\n{reason_text}" + task.dev_notes = f"{task.dev_notes}\n\n{cancellation_note}" else: - task.dev_notes = reason_text + task.dev_notes = cancellation_note await self.session.flush() - # Use TaskService for consistent cancel behavior (cascades to descendants) + # Use TaskService for consistent cancel behavior (cascades to descendants). + # Thread the caller's role into the cascade role gate so a non-PM + # caller can't cascade-cancel descendants the role can't cancel. task_service = TaskService(self.session) - task = await task_service.cancel(task_uuid) + task = await task_service.cancel(task_uuid, agent_role=agent_role or "cell_pm") if task is None: raise ValueError(f"Failed to cancel task: {task_id}") - logger.info("Cancelled task via A2A", task_id=task_id, reason=reason) + logger.info( + "Cancelled task via A2A", + task_id=task_id, + reason=reason, + actor=actor_slug, + role=agent_role, + ) return self.task_to_a2a(task) diff --git a/tests/integration/test_a2a_routes.py b/tests/integration/test_a2a_routes.py index 79a63737..b15977d8 100644 --- a/tests/integration/test_a2a_routes.py +++ b/tests/integration/test_a2a_routes.py @@ -13,7 +13,7 @@ import pytest import pytest_asyncio from fastapi import FastAPI from httpx import ASGITransport, AsyncClient -from roboco.api.deps import get_current_agent_slug, get_db +from roboco.api.deps import get_agent_context, get_current_agent_slug, get_db from roboco.api.routes.a2a import router as a2a_router from roboco.api.routes.a2a import wellknown_router from roboco.db.tables import AgentTable, ProjectTable, TaskTable @@ -25,6 +25,7 @@ from roboco.models.base import ( TaskStatus, TaskType, ) +from roboco.models.permissions import AgentContext if TYPE_CHECKING: from collections.abc import AsyncGenerator, AsyncIterator @@ -91,18 +92,39 @@ async def a2a_route_client( async def _override_agent_slug() -> str: return dev.slug + async def _override_agent_context() -> AgentContext: + # The authenticated caller is the seeded developer by default. Tests + # that need a different role (e.g. the PM-gated cancel route) swap this + # override on the yielded app before posting. + return AgentContext( + agent_id=dev.id, role=AgentRole.DEVELOPER, team=Team.BACKEND, slug=dev.slug + ) + app.dependency_overrides[get_db] = _override_db app.dependency_overrides[get_current_agent_slug] = _override_agent_slug + app.dependency_overrides[get_agent_context] = _override_agent_context transport = ASGITransport(app=app) async with AsyncClient(transport=transport, base_url="http://test") as client: - yield {"client": client, "dev": dev, "task": task} + yield {"client": client, "dev": dev, "task": task, "app": app} app.dependency_overrides.clear() _HDR = {"X-Agent-ID": "be-dev-1", "X-Agent-Role": "developer"} +def _set_pm_context(app: FastAPI, dev: AgentTable) -> None: + """Override the agent context to a cell PM so the PM-gated cancel route + admits the call (the default fixture context is a developer).""" + + async def _pm() -> AgentContext: + return AgentContext( + agent_id=dev.id, role=AgentRole.CELL_PM, team=Team.BACKEND, slug=dev.slug + ) + + app.dependency_overrides[get_agent_context] = _pm + + # --------------------------------------------------------------------------- # Well-known endpoints # --------------------------------------------------------------------------- @@ -435,13 +457,17 @@ async def test_cancel_task_success(a2a_route_client: dict) -> None: } ) client = a2a_route_client["client"] + _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() instance.cancel_task = AsyncMock(return_value=a2a_task) mock_service_cls.return_value = instance response = await client.post( f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel", - json={"reason": "no longer needed"}, + json={ + "name": f"tasks/{a2a_route_client['task'].id}", + "reason": "no longer needed", + }, headers=_HDR, ) # 200 expected; pydantic may serialize as 422 if response_model coercion @@ -452,6 +478,7 @@ async def test_cancel_task_success(a2a_route_client: dict) -> None: async def test_cancel_task_already_terminal(a2a_route_client: dict) -> None: client = a2a_route_client["client"] + _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() instance.cancel_task = AsyncMock( @@ -469,6 +496,7 @@ async def test_cancel_task_already_terminal(a2a_route_client: dict) -> None: async def test_cancel_task_not_found(a2a_route_client: dict) -> None: client = a2a_route_client["client"] + _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() instance.cancel_task = AsyncMock(side_effect=ValueError("Task missing")) @@ -481,10 +509,123 @@ async def test_cancel_task_not_found(a2a_route_client: dict) -> None: # --------------------------------------------------------------------------- -# Chat conversations +# #423: the cancel route must be authenticated + PM/management-gated, and pass +# the authenticated actor + role into the service (it cascades cancel to all +# non-terminal descendants — lifecycle rule: Any -> cancelled: PM roles only). # --------------------------------------------------------------------------- +@pytest.mark.asyncio +async def test_cancel_task_developer_role_forbidden(a2a_route_client: dict) -> None: + """A developer (default fixture context) must NOT be able to cancel a task + via A2A — the route was previously unauthenticated with no role gate, so any + caller could cancel any task tree (#423). Now PM/management-only.""" + client = a2a_route_client["client"] + # default fixture context = developer + with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: + instance = AsyncMock() + instance.cancel_task = AsyncMock() + mock_service_cls.return_value = instance + response = await client.post( + f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel", + headers=_HDR, + ) + assert response.status_code == HTTPStatus.FORBIDDEN + instance.cancel_task.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_cancel_task_no_auth_header_rejected(a2a_route_client: dict) -> None: + """A request with no agent headers at all is rejected — the route must not + be reachable unauthenticated (#423).""" + client = a2a_route_client["client"] + response = await client.post( + f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel", + ) + assert response.status_code in ( + HTTPStatus.UNAUTHORIZED, + HTTPStatus.FORBIDDEN, + HTTPStatus.UNPROCESSABLE_ENTITY, + ) + + +@pytest.mark.asyncio +async def test_cancel_task_pm_passes_actor_and_role_to_service( + a2a_route_client: dict, +) -> None: + """A PM cancel threads the authenticated role (for the cascade role gate) + and the actor slug (for the cancellation-note attribution) into the service + — previously the service was called with no actor and a hardcoded + cell_pm role, so the audit trail recorded no real caller (#423).""" + client = a2a_route_client["client"] + _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) + with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: + instance = AsyncMock() + instance.cancel_task = AsyncMock( + return_value=A2ATask.model_validate( + { + "id": str(a2a_route_client["task"].id), + "contextId": str(uuid4()), + "status": A2ATaskStatus(state=A2ATaskState.CANCELLED).model_dump( + mode="json" + ), + } + ) + ) + mock_service_cls.return_value = instance + # No body → request=None → the handler runs (a body without the A2A + # ``name`` field 422s at request validation before the handler). The + # invariant under test is the role/slug threading, not the reason. + response = await client.post( + f"/api/a2a/tasks/{a2a_route_client['task'].id}/cancel", + headers=_HDR, + ) + assert response.status_code == HTTPStatus.OK + # The authenticated PM role + slug reach the service. + _kwargs = instance.cancel_task.await_args.kwargs + assert _kwargs.get("agent_role") == "cell_pm" + assert _kwargs.get("actor_slug") == a2a_route_client["dev"].slug + + +# --------------------------------------------------------------------------- +# #116: send_message must record the AUTHENTICATED identity as the responder, +# not a client-supplied metadata.from_agent (spoof). +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_send_message_uses_authenticated_identity_not_client_from_agent( + a2a_route_client: dict, +) -> None: + """is_response=True must stamp the authenticated caller's slug as the + responder, ignoring a spoofed metadata.from_agent — previously the route + took from_agent verbatim from the request body, so any agent could + impersonate anyone (e.g. from_agent='ceo') in the task's notes and in the + spawn/notification routed back to the original requester (#116).""" + client = a2a_route_client["client"] + with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: + instance = AsyncMock() + instance.update_task_from_message = AsyncMock(return_value=None) + mock_service_cls.return_value = instance + response = await client.post( + "/api/a2a/message/send", + json={ + "message": { + "role": "user", + "parts": [{"type": "text", "text": "hi"}], + "taskId": str(a2a_route_client["task"].id), + }, + "metadata": {"is_response": True, "from_agent": "ceo"}, + }, + headers=_HDR, + ) + assert response.status_code == HTTPStatus.OK + # The authenticated dev slug is the responder — NOT the spoofed 'ceo'. + _kwargs = instance.update_task_from_message.await_args.kwargs + assert _kwargs.get("responder_agent") == a2a_route_client["dev"].slug + assert _kwargs.get("responder_agent") != "ceo" + + @pytest.mark.asyncio async def test_chat_create_conversation_access_denied( a2a_route_client: dict, @@ -1239,6 +1380,7 @@ async def test_cancel_task_success_no_body(a2a_route_client: dict) -> None: } ) client = a2a_route_client["client"] + _set_pm_context(a2a_route_client["app"], a2a_route_client["dev"]) with patch("roboco.api.routes.a2a.A2AService") as mock_service_cls: instance = AsyncMock() instance.cancel_task = AsyncMock(return_value=a2a_task)