diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 3cff820..0346635 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -14,12 +14,27 @@ repos: - id: debug-statements - id: check-merge-conflict - - repo: https://github.com/astral-sh/ruff-pre-commit - rev: v0.8.6 + # Ruff runs from the project's own lockfile rather than a separately + # pinned hook environment. The two used to drift: the hook pinned v0.8.6 + # while the dev dependency resolved to 0.15, and they disagreed on 34 lint + # findings and 9 files' formatting. Every PR that touched one of those + # files failed the gate on code it had not written. Sourcing ruff from + # `uv run` — as the mypy hook below already does — makes that impossible: + # there is one ruff, the one in uv.lock. + - repo: local hooks: - id: ruff - args: [--fix] + name: ruff + entry: uv run ruff check --fix + language: system + types_or: [python, pyi] + require_serial: true - id: ruff-format + name: ruff-format + entry: uv run ruff format + language: system + types_or: [python, pyi] + require_serial: true - repo: local hooks: diff --git a/docs/architecture/api-reference.md b/docs/architecture/api-reference.md index 4d03bb4..e41d8e4 100644 --- a/docs/architecture/api-reference.md +++ b/docs/architecture/api-reference.md @@ -478,6 +478,11 @@ curl -X POST http://localhost:8000/api/v1/my_agent/a2a/tasks \ Get A2A task status. +These per-agent A2A routes — status, `events` and `cancel` — only accept tasks +submitted to *this* agent. A task belonging to another agent answers `404`, the +same as an id that does not exist. The shared task board at +`/api/v1/tasks/{id}` is cross-agent by design and is unaffected. + ```bash curl http://localhost:8000/api/v1/my_agent/a2a/tasks/task_a1b2c3d4e5f6 ``` diff --git a/pyproject.toml b/pyproject.toml index d29512d..61c1817 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -69,7 +69,7 @@ dev = [ "pytest-asyncio>=0.24", "pytest-cov>=6.0", "pytest-mock>=3.14", - "ruff>=0.8", + "ruff>=0.15", "mypy>=1.13", "pre-commit>=4.0", "build>=1.0", @@ -104,7 +104,7 @@ dev = [ "pytest-asyncio>=0.24", "pytest-cov>=6.0", "pytest-mock>=3.14", - "ruff>=0.8", + "ruff>=0.15", "mypy>=1.13", "pre-commit>=4.0", "build>=1.0", diff --git a/src/agentomatic/core/router_factory.py b/src/agentomatic/core/router_factory.py index ea39789..530c255 100644 --- a/src/agentomatic/core/router_factory.py +++ b/src/agentomatic/core/router_factory.py @@ -1306,6 +1306,33 @@ def _a2a_view(record: Any) -> dict[str, Any]: view["error"] = record.error return view + async def _owned_a2a_task(task_id: str) -> Any: + """Fetch a task, insisting it belongs to *this* agent. + + These routes are namespaced per agent but used to accept any task id, + so one agent's route would read — and cancel — another agent's task. + Zero-trust derives its policy from the agent segment of the path, so + that let a caller reach work the policy engine had denied them. + + A task owned by another agent answers 404, exactly as a nonexistent + one does: a distinct status would confirm the id exists. + + Args: + task_id: Task being addressed. + + Returns: + The task record. + + Raises: + HTTPException: 404 when the task is missing or owned elsewhere. + """ + if task_manager is None: + raise HTTPException(501, "A2A task tracking requires a task manager") + record = await task_manager.get(task_id) + if record is None or record.target != agent_name: + raise HTTPException(404, f"Task '{task_id}' not found") + return record + def _a2a_event_view(evt: Any, record: Any) -> dict[str, Any]: """Render one recorded event as an A2A-shaped task object. @@ -1404,16 +1431,13 @@ async def submit_a2a_task(request: A2ATaskRequest) -> dict[str, Any]: @router.get("/a2a/tasks/{task_id}") async def get_a2a_task(task_id: str) -> dict[str, Any]: """Get A2A task status.""" - if task_manager is not None: - record = await task_manager.get(task_id) - if record is None: - raise HTTPException(404, f"Task '{task_id}' not found") - return _a2a_view(record) - return { - "task_id": task_id, - "status": "completed", - "message": "Task tracking requires a task manager", - } + if task_manager is None: + return { + "task_id": task_id, + "status": "completed", + "message": "Task tracking requires a task manager", + } + return _a2a_view(await _owned_a2a_task(task_id)) @router.get("/a2a/tasks/{task_id}/events") async def stream_a2a_task(task_id: str, request: Request, since: int = Query(default=0, ge=0)): @@ -1429,9 +1453,7 @@ async def stream_a2a_task(task_id: str, request: Request, since: int = Query(def if task_manager is None: raise HTTPException(501, "A2A task streaming requires a task manager") - record = await task_manager.get(task_id) - if record is None: - raise HTTPException(404, f"Task '{task_id}' not found") + await _owned_a2a_task(task_id) resume_from = since or parse_last_event_id(request.headers.get("last-event-id")) @@ -1504,6 +1526,9 @@ async def cancel_a2a_task(task_id: str) -> dict[str, Any]: """Cancel an in-flight A2A task.""" if task_manager is None: raise HTTPException(501, "Task cancellation requires a task manager") + # Cancelling is a write, so the ownership check matters more here than + # on the read routes: without it any agent could stop any task. + await _owned_a2a_task(task_id) cancelled = await task_manager.cancel(task_id) if not cancelled: raise HTTPException(409, f"Task '{task_id}' not found or already terminal") diff --git a/tests/test_a2a_streaming.py b/tests/test_a2a_streaming.py index da75e35..f3e239e 100644 --- a/tests/test_a2a_streaming.py +++ b/tests/test_a2a_streaming.py @@ -10,7 +10,9 @@ from __future__ import annotations +import asyncio import json +import time from dataclasses import dataclass, field from typing import Any @@ -244,3 +246,99 @@ async def counting_get(self, tid: str): # One lookup to 404-check, one for the snapshot, at most one for the # terminal payload — never one per frame. assert len(calls) <= 3, f"{len(calls)} store reads for {frames} frames" + + +class TestA2ATasksAreScopedToTheirAgent: + """The per-agent A2A routes are namespaced by agent but used to accept any + task id. Zero-trust derives its policy from that agent segment, so one + agent's route could read — and cancel — work the policy engine had denied + the caller on the owning agent. + + The shared task board at ``/api/v1/tasks/{id}`` is deliberately untouched: + it is cross-agent by design. + """ + + @pytest.fixture + def two_agents(self) -> Any: + async def secret(state: dict[str, Any]) -> dict[str, Any]: + return {"response": "CONFIDENTIAL-PAYROLL"} + + async def slow(state: dict[str, Any]) -> dict[str, Any]: + await asyncio.sleep(5) + return {"response": "done"} + + platform = AgentPlatform( + agents_dir="/tmp/agentomatic_a2a_scope_test", + title="A2A Scope", + version="0.0.1", + enable_tasks=True, + ) + platform.register_agent( + manifest=AgentManifest(name="payroll", slug="p", description="p"), node_fn=secret + ) + platform.register_agent( + manifest=AgentManifest(name="slowpoke", slug="s", description="s"), node_fn=slow + ) + platform.register_agent( + manifest=AgentManifest(name="faq", slug="f", description="f"), node_fn=_echo_fn + ) + with TestClient(platform.build()) as test_client: + yield test_client + + def _submit(self, client: Any, agent: str) -> str: + return str( + client.post(f"{BASE}/{agent}/a2a/tasks", json={"message": {"content": "x"}}).json()[ + "task_id" + ] + ) + + def test_another_agent_cannot_read_the_task(self, two_agents: Any) -> None: + task_id = self._submit(two_agents, "payroll") + + leaked = two_agents.get(f"{BASE}/faq/a2a/tasks/{task_id}") + + assert leaked.status_code == 404 + assert "CONFIDENTIAL-PAYROLL" not in leaked.text + + def test_another_agent_cannot_stream_the_task(self, two_agents: Any) -> None: + task_id = self._submit(two_agents, "payroll") + + leaked = two_agents.get(f"{BASE}/faq/a2a/tasks/{task_id}/events") + + assert leaked.status_code == 404 + assert "CONFIDENTIAL-PAYROLL" not in leaked.text + + def test_another_agent_cannot_cancel_a_running_task(self, two_agents: Any) -> None: + """The write is the sharper end of this: without the check any agent + could stop any task.""" + task_id = self._submit(two_agents, "slowpoke") + time.sleep(0.3) # let it reach running + + assert two_agents.post(f"{BASE}/faq/a2a/tasks/{task_id}/cancel").status_code == 404 + + def test_the_owning_agent_still_reads_and_streams(self, two_agents: Any) -> None: + task_id = self._submit(two_agents, "payroll") + + assert two_agents.get(f"{BASE}/payroll/a2a/tasks/{task_id}").status_code == 200 + assert two_agents.get(f"{BASE}/payroll/a2a/tasks/{task_id}/events").status_code == 200 + + def test_the_owning_agent_can_still_cancel(self, two_agents: Any) -> None: + task_id = self._submit(two_agents, "slowpoke") + time.sleep(0.3) + + assert two_agents.post(f"{BASE}/slowpoke/a2a/tasks/{task_id}/cancel").status_code == 200 + + def test_a_foreign_task_looks_exactly_like_a_missing_one(self, two_agents: Any) -> None: + """A distinct status would confirm the id exists.""" + task_id = self._submit(two_agents, "payroll") + + foreign = two_agents.get(f"{BASE}/faq/a2a/tasks/{task_id}") + missing = two_agents.get(f"{BASE}/faq/a2a/tasks/task_deadbeefdeadbeef") + + assert foreign.status_code == missing.status_code == 404 + + def test_the_shared_task_board_is_unchanged(self, two_agents: Any) -> None: + """This fix is deliberately local to the per-agent A2A routes.""" + task_id = self._submit(two_agents, "payroll") + + assert two_agents.get(f"{BASE}/tasks/{task_id}").status_code == 200 diff --git a/uv.lock b/uv.lock index bf51ff9..e0fc167 100644 --- a/uv.lock +++ b/uv.lock @@ -212,7 +212,7 @@ requires-dist = [ { name = "pyyaml", specifier = ">=6.0" }, { name = "questionary", marker = "extra == 'cli'", specifier = ">=2.0" }, { name = "rich", marker = "extra == 'cli'", specifier = ">=13.0" }, - { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.8" }, + { name = "ruff", marker = "extra == 'dev'", specifier = ">=0.15" }, { name = "sqlalchemy", extras = ["asyncio"], marker = "extra == 'db'", specifier = ">=2.0" }, { name = "sqlalchemy", extras = ["asyncio"], marker = "extra == 'db-postgres'", specifier = ">=2.0" }, { name = "twine", marker = "extra == 'dev'", specifier = ">=6.0" }, @@ -235,7 +235,7 @@ dev = [ { name = "pytest-cov", specifier = ">=6.0" }, { name = "pytest-mock", specifier = ">=3.14" }, { name = "python-semantic-release", specifier = ">=10.0,<11" }, - { name = "ruff", specifier = ">=0.8" }, + { name = "ruff", specifier = ">=0.15" }, { name = "twine", specifier = ">=6.0" }, ] docs = [