diff --git a/roboco/api/middleware.py b/roboco/api/middleware.py index eb698854..fbbcc499 100644 --- a/roboco/api/middleware.py +++ b/roboco/api/middleware.py @@ -363,6 +363,47 @@ def _uuid_field_remediation(errors: Sequence[Any]) -> str | None: return None +# Credential-bearing request fields. A 422 on a secret-bearing request +# would otherwise dump the plaintext GitHub PAT / provider API key / bearer +# token into structlog output before the route ever encrypts it. The per-field +# ``errors`` carry only field names and types (never values), so they stay +# logged unchanged. Match by exact key name so a renamed secret field is +# caught by the next audit pass rather than silently leaking. +_SECRET_FIELD_NAMES: frozenset[str] = frozenset( + { + "git_token", + "api_key", + "auth_token", + "token", + "password", + "secret", + "client_secret", + "access_token", + "refresh_token", + } +) + +_REDACTED = "***REDACTED***" + + +def _scrub_secrets(value: Any) -> Any: + """Return a deep copy of ``value`` with known secret fields redacted. + + Recurses into nested dicts and lists so a secret inside ``nested: {...}`` + or a list element is also scrubbed. Non-secret fields are preserved so ops + can still see which field broke. The original ``rve.body`` is not mutated + (the 422 response body echoes the client's own submission unscrubbed). + """ + if isinstance(value, dict): + return { + k: (_REDACTED if k in _SECRET_FIELD_NAMES else _scrub_secrets(v)) + for k, v in value.items() + } + if isinstance(value, list): + return [_scrub_secrets(v) for v in value] + return value + + async def request_validation_handler(request: Request, exc: Exception) -> JSONResponse: """Log the rejected body before returning the standard 422 response. @@ -373,15 +414,21 @@ async def request_validation_handler(request: Request, exc: Exception) -> JSONRe When the failure is a truncated ``task_id`` (the recurring agent mistake), add a ``remediate`` hint so the agent knows to retry with the full UUID. + + F022: the log line scrubs known credential-bearing fields + (``git_token`` / ``api_key`` / ``auth_token`` / …) from the body before + logging. The 422 *response* body is unchanged — the client sent those + values, only the server's own log is redacted. """ rve = cast("RequestValidationError", exc) body = rve.body if isinstance(rve.body, str | bytes | dict | list) else None errors = rve.errors() + body_for_log = _scrub_secrets(body) if isinstance(body, dict | list) else body logger.warning( "Request validation failed", path=request.url.path, method=request.method, - body=body, + body=body_for_log, errors=errors, ) content: dict[str, Any] = {"detail": errors, "body": body} diff --git a/roboco/api/routes/a2a.py b/roboco/api/routes/a2a.py index e5761262..ae5ab111 100644 --- a/roboco/api/routes/a2a.py +++ b/roboco/api/routes/a2a.py @@ -24,6 +24,7 @@ from fastapi.responses import JSONResponse from sse_starlette import EventSourceResponse from roboco.api.deps import CurrentAgentSlug, DbSession +from roboco.api.routes.v1._role_dep import require_any_authenticated_agent from roboco.api.schemas.a2a_chat import ( ConversationCloseRequest, ConversationCreateRequest, @@ -39,6 +40,7 @@ from roboco.api.schemas.a2a_chat import ( PairListResponse, PairResponse, ) +from roboco.db.base import get_session_factory from roboco.enforcement import A2AAccessDeniedError from roboco.models.a2a import ( A2AConversationStatus, @@ -107,7 +109,10 @@ async def get_agent_card( # ============================================================================= -@router.post("/message/send") +@router.post( + "/message/send", + dependencies=[require_any_authenticated_agent], +) async def send_message( request: SendMessageRequest, db: DbSession, @@ -187,7 +192,10 @@ async def send_message( return {"status": "success", "a2a_request": result} -@router.post("/message/stream") +@router.post( + "/message/stream", + dependencies=[require_any_authenticated_agent], +) async def send_message_stream( request: Request, body: SendMessageRequest, @@ -275,22 +283,31 @@ async def send_message_stream( ) -@router.get("/tasks/{task_id}/subscribe") +@router.get( + "/tasks/{task_id}/subscribe", + dependencies=[require_any_authenticated_agent], +) async def subscribe_to_task( request: Request, task_id: str, - db: DbSession, ) -> EventSourceResponse: """ Subscribe to task updates via SSE. Opens a persistent connection that streams task state changes until the task reaches a terminal state or client disconnects. - """ - service = A2AService(db) - # Validate task exists - a2a_task = await service.get_task(task_id) + F024: each poll opens a SHORT-LIVED session via ``get_session_factory`` + and closes it before the next ``asyncio.sleep`` — never holding one + asyncpg connection across the full SSE lifetime (up to 1 hour / 720 + polls), which previously exhausted the pool one connection per connected + client. The route takes no ``db: DbSession`` for the same reason. + """ + session_factory = get_session_factory() + + # Validate task exists with a short-lived session (released immediately). + async with session_factory() as session: + a2a_task = await A2AService(session).get_task(task_id) if a2a_task is None: raise HTTPException( status_code=status.HTTP_404_NOT_FOUND, @@ -298,7 +315,7 @@ async def subscribe_to_task( ) async def generate_updates() -> AsyncGenerator[dict[str, Any]]: - """Stream task updates.""" + """Stream task updates — one short-lived session per poll.""" poll_count = 0 max_polls = 720 # 1 hour at 5s interval last_state = None @@ -307,8 +324,11 @@ async def subscribe_to_task( if await request.is_disconnected(): break - # Refresh task state from DB - task = await service.get_task(task_id) + # F024: refresh task state from a per-poll session that is + # released before the sleep below — never held across the poll + # interval, so the asyncpg pool is free between queries. + async with session_factory() as session: + task = await A2AService(session).get_task(task_id) if task is None: break diff --git a/roboco/api/routes/dashboard.py b/roboco/api/routes/dashboard.py index 492483f4..9e65d592 100644 --- a/roboco/api/routes/dashboard.py +++ b/roboco/api/routes/dashboard.py @@ -10,7 +10,7 @@ from uuid import UUID from fastapi import APIRouter, HTTPException, Query, status -from roboco.api.deps import DbSession +from roboco.api.deps import CurrentAgentContext, DbSession from roboco.api.schemas.dashboard import ( AuditorDashboard, AuditorFlag, @@ -23,6 +23,7 @@ from roboco.api.schemas.dashboard import ( TeamHealth, UsageSummary, ) +from roboco.models import AgentRole from roboco.models.base import Team from roboco.models.dashboard import CreateFlagParams from roboco.services.dashboard import get_dashboard_service @@ -32,6 +33,22 @@ from roboco.services.usage import get_usage_service router = APIRouter() +# The auditor flag/report mutating routes are gated to the Auditor and the +# CEO. The Auditor is the silent-observer role whose flags/reports feed the +# CEO; the CEO overrides. Mirrors ``_require_curator`` in playbooks.py and +# ``_require_ceo`` in release.py. Read-only auditor views (``GET +# /auditor/flags``, ``GET /auditor/reports``, ``GET /auditor``) stay open — +# the dashboard is observable by any authenticated operator. +_AUDITOR_OR_CEO_ROLES = frozenset({AgentRole.AUDITOR, AgentRole.CEO}) + + +def _require_auditor_or_ceo(agent: CurrentAgentContext) -> None: + if agent.role not in _AUDITOR_OR_CEO_ROLES: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only the Auditor or CEO may mutate auditor flags or reports", + ) + # ============================================================================= # AUDITOR DASHBOARD @@ -154,8 +171,10 @@ async def get_auditor_flags( async def create_auditor_flag( data: CreateFlagRequest, db: DbSession, + agent: CurrentAgentContext, ) -> AuditorFlag: """Create a new auditor flag.""" + _require_auditor_or_ceo(agent) service = get_dashboard_service(db) params = CreateFlagParams( severity=data.severity.value, @@ -184,9 +203,11 @@ async def create_auditor_flag( async def resolve_auditor_flag( flag_id: UUID, db: DbSession, + agent: CurrentAgentContext, notes: str | None = None, ) -> dict[str, str]: """Resolve an auditor flag.""" + _require_auditor_or_ceo(agent) service = get_dashboard_service(db) if not service.resolve_flag(flag_id, notes): raise HTTPException( @@ -226,8 +247,10 @@ async def get_auditor_reports( async def create_auditor_report( data: CreateReportRequest, db: DbSession, + agent: CurrentAgentContext, ) -> AuditorReport: """Create a new auditor report.""" + _require_auditor_or_ceo(agent) service = get_dashboard_service(db) report = service.create_report( report_type=data.report_type, @@ -250,8 +273,10 @@ async def create_auditor_report( async def send_auditor_report( report_id: UUID, db: DbSession, + agent: CurrentAgentContext, ) -> dict[str, str]: """Mark a report as sent to CEO.""" + _require_auditor_or_ceo(agent) service = get_dashboard_service(db) if not service.send_report(report_id): raise HTTPException( diff --git a/roboco/api/routes/orchestrator.py b/roboco/api/routes/orchestrator.py index fa2d48d4..0f70b39e 100644 --- a/roboco/api/routes/orchestrator.py +++ b/roboco/api/routes/orchestrator.py @@ -5,10 +5,11 @@ API endpoints for managing the Agent Orchestrator. """ from datetime import datetime +from typing import Annotated -from fastapi import APIRouter, HTTPException, status +from fastapi import APIRouter, Depends, Header, HTTPException, status -from roboco.api.deps import get_orchestrator, set_orchestrator +from roboco.api.deps import _check_agent_auth_token, get_orchestrator, set_orchestrator from roboco.api.schemas.orchestrator import ( AgentStatusResponse, OrchestratorStatusResponse, @@ -16,8 +17,42 @@ from roboco.api.schemas.orchestrator import ( SpawnAgentRequest, WaitingAgentResponse, ) +from roboco.foundation.identity import Role -router = APIRouter() +# Orchestrator control routes (spawn / stop / resolve-wait / mark-waiting, +# plus the read-only status views) are operator/CEO control surfaces — any +# client that could reach the API could previously spawn, stop, or +# manipulate any agent's runtime state. The guard mirrors the panel-token +# approach used by the WebSocket streams (DB-free): it binds the presented +# ``X-Agent-ID`` to a verified HMAC token and asserts the role is CEO. In +# dev (header-trust) mode a missing token is a no-op (the panel/operator +# flow keeps working), but a presented-but-forged token is still rejected — +# the same contract as the v1 flow role guards and the do router. CEO is the +# sole operator role; agents (developers/QA/PMs) drive the orchestrator via +# MCP verbs, not these HTTP routes, so a developer token is correctly 403'd +# here. +_CEO_ROLE = Role.CEO.value + + +def _require_ceo( + x_agent_id: Annotated[str, Header(alias="X-Agent-ID")], + x_agent_role: Annotated[str, Header(alias="X-Agent-Role")], + x_agent_team: Annotated[str | None, Header(alias="X-Agent-Team")] = None, + x_agent_token: Annotated[str | None, Header(alias="X-Agent-Token")] = None, +) -> None: + # Bind the role header to a verified token BEFORE trusting it (same + # defense-in-depth contract as the v1 flow role guards in _role_dep.py). + _check_agent_auth_token(x_agent_id, x_agent_role, x_agent_team, x_agent_token) + # ``Role`` is a StrEnum so the lowercase header string compares equal to + # its matching member. + if x_agent_role.lower() != _CEO_ROLE: + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="Only the CEO/operator may control the orchestrator", + ) + + +router = APIRouter(dependencies=[Depends(_require_ceo)]) # Re-export set_orchestrator for bootstrap code __all__ = ["router", "set_orchestrator"] diff --git a/tests/unit/api/test_a2a_message_auth.py b/tests/unit/api/test_a2a_message_auth.py new file mode 100644 index 00000000..2298e0ef --- /dev/null +++ b/tests/unit/api/test_a2a_message_auth.py @@ -0,0 +1,190 @@ +"""F023: POST /api/a2a/message/send and /message/stream must enforce the same +HMAC agent-token gate as the /api/v1/do/* router (F003). + +Both routes previously took only ``request: SendMessageRequest, db: DbSession`` +— no auth dependency. The sender was self-declared in the request body +(``metadata.from_agent``), so any caller could impersonate any agent and +inject A2A notifications that the orchestrator dispatcher picks up to spawn +target agents. The fix reuses F003's ``require_any_authenticated_agent`` +(token-only, DB-free, no role assertion — the a2a router serves every role). +In dev (header-trust) mode a missing token is a no-op; a presented-but-forged +token is still rejected, exactly as the do router does. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.agents_config import issue_agent_token +from roboco.api.routes.a2a import router as a2a_router + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + +_SECRET = "test-secret-for-a2a-auth" +_AGENT_ID = "00000000-0000-0000-0000-000000000002" +_HTTP_200 = 200 +_HTTP_400 = 400 +_HTTP_401 = 401 + + +def _message_body() -> dict: + """A minimal valid SendMessageRequest body. + + ``message.task_id`` defaults to None, so the send route raises + TASK_ID_REQUIRED (400) AFTER the gate passes — proving the gate let the + request through without touching the DB. The stream route takes the + ``else`` (new-task) branch and returns 200 with no DB access. + """ + return {"message": {"role": "user", "parts": [{"type": "text", "text": "x"}]}} + + +@pytest.fixture +async def a2a_client() -> AsyncIterator[AsyncClient]: + app = FastAPI() + app.include_router(a2a_router, prefix="/api/a2a") + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + yield client + app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# /message/send +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_send_rejects_missing_token_when_required( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Strict mode + no X-Agent-Token => 401, never reaches the handler.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + r = await a2a_client.post( + "/api/a2a/message/send", + json=_message_body(), + headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "developer"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_send_rejects_forged_token_even_in_dev( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A presented-but-forged token is rejected even in header-trust mode.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + r = await a2a_client.post( + "/api/a2a/message/send", + json=_message_body(), + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": "forged-not-a-real-hmac", + }, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_send_accepts_valid_token( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A valid token passes the gate; the route body then raises + TASK_ID_REQUIRED (400) because message.task_id is None — proving the + gate let the request through (401 would mean the gate rejected it).""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + token = issue_agent_token(_AGENT_ID, "developer") + r = await a2a_client.post( + "/api/a2a/message/send", + json=_message_body(), + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": token, + }, + ) + assert r.status_code == _HTTP_400 # TASK_ID_REQUIRED — gate passed + + +@pytest.mark.asyncio +async def test_send_dev_mode_missing_token_still_succeeds_gate( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Dev mode + no token => no-op, route body runs (400 TASK_ID_REQUIRED). + Preserves the agent/panel flow in dev exactly as F003/F004 did.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + r = await a2a_client.post( + "/api/a2a/message/send", + json=_message_body(), + headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "developer"}, + ) + assert r.status_code == _HTTP_400 # gate passed; route raised TASK_ID_REQUIRED + + +# --------------------------------------------------------------------------- +# /message/stream +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_stream_rejects_missing_token_when_required( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Strict mode + no X-Agent-Token => 401 on the stream route too.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + r = await a2a_client.post( + "/api/a2a/message/stream", + json=_message_body(), + headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "developer"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_stream_rejects_forged_token_even_in_dev( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A presented-but-forged token is rejected even in header-trust mode.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + r = await a2a_client.post( + "/api/a2a/message/stream", + json=_message_body(), + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": "forged-not-a-real-hmac", + }, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_stream_accepts_valid_token( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A valid token passes the gate; the stream route returns 200 (SSE) on + the new-task branch (message.task_id is None -> no DB access).""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + token = issue_agent_token(_AGENT_ID, "developer") + r = await a2a_client.post( + "/api/a2a/message/stream", + json=_message_body(), + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": token, + }, + ) + assert r.status_code == _HTTP_200 diff --git a/tests/unit/api/test_a2a_subscribe.py b/tests/unit/api/test_a2a_subscribe.py new file mode 100644 index 00000000..4f3b8728 --- /dev/null +++ b/tests/unit/api/test_a2a_subscribe.py @@ -0,0 +1,226 @@ +"""F024: the SSE ``subscribe_to_task`` endpoint must (a) be authenticated +like the rest of the a2a message surface (F023) and (b) acquire a SHORT-LIVED +DB session per poll iteration instead of holding the request-scoped +``db: DbSession`` for the full SSE lifetime (up to 1 hour / 720 polls), which +exhausted the asyncpg pool one connection per connected client. + +The fix mirrors F003's ``require_any_authenticated_agent`` for auth and uses +``get_session_factory()`` inside the generator so each poll opens, queries, +and closes its own session — no connection is held across ``asyncio.sleep``. +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING, Any, cast +from unittest.mock import AsyncMock, MagicMock + +import pytest +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.agents_config import issue_agent_token +from roboco.api.routes import a2a as a2a_module +from roboco.api.routes.a2a import router as a2a_router +from roboco.db.base import get_db + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + + from fastapi.routing import APIRoute + +_SECRET = "test-secret-for-a2a-subscribe" +_AGENT_ID = "00000000-0000-0000-0000-000000000003" +_HTTP_200 = 200 +_HTTP_401 = 401 +_HTTP_404 = 404 + + +@pytest.fixture +async def a2a_client() -> AsyncIterator[AsyncClient]: + app = FastAPI() + app.include_router(a2a_router, prefix="/api/a2a") + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + yield client + app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# Auth gate (F023 parity) +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_subscribe_rejects_missing_token_when_required( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """Strict mode + no X-Agent-Token => 401, never reaches the generator.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + r = await a2a_client.get( + "/api/a2a/tasks/some-task/subscribe", + headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "developer"}, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_subscribe_rejects_forged_token_even_in_dev( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A presented-but-forged token is rejected even in header-trust mode.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + r = await a2a_client.get( + "/api/a2a/tasks/some-task/subscribe", + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": "forged-not-a-real-hmac", + }, + ) + assert r.status_code == _HTTP_401 + + +@pytest.mark.asyncio +async def test_subscribe_accepts_valid_token_then_404s_unknown_task( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """A valid token passes the gate; the route then 404s on the initial + task-existence check (no DB seeded). 404 (not 401) proves the gate let + the request through.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + token = issue_agent_token(_AGENT_ID, "developer") + # get_task returns None -> 404. Patch A2AService.get_task to return None + # so the route doesn't need a real DB. + monkeypatch.setattr(a2a_module.A2AService, "get_task", AsyncMock(return_value=None)) + r = await a2a_client.get( + "/api/a2a/tasks/some-task/subscribe", + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": token, + }, + ) + assert r.status_code == _HTTP_404 + + +# --------------------------------------------------------------------------- +# Session-per-query: structural + behavioral +# --------------------------------------------------------------------------- + + +def test_subscribe_route_does_not_hold_request_scoped_db() -> None: + """F024: the route must NOT depend on ``get_db`` — the request-scoped + session would be held for the full SSE lifetime (up to 1 hour). Each + poll must open its own short-lived session via ``get_session_factory``. + """ + subscribe_route = cast( + "APIRoute", + next( + r + for r in a2a_router.routes + if getattr(r, "path", "") == "/tasks/{task_id}/subscribe" + ), + ) + # Walk the route's dependency tree; get_db must not appear anywhere. + deps = [subscribe_route.dependant] + seen: set[int] = set() + found_get_db = False + while deps: + d = deps.pop() + if id(d) in seen: + continue + seen.add(id(d)) + if d.call is get_db: + found_get_db = True + deps.extend(d.dependencies) + assert not found_get_db, ( + "subscribe_to_task still depends on get_db — the request-scoped " + "session is held for the full SSE lifetime (pool-exhaustion vector)." + ) + + +@pytest.mark.asyncio +async def test_subscribe_opens_a_short_lived_session_per_poll( + a2a_client: AsyncClient, monkeypatch: pytest.MonkeyPatch +) -> None: + """F024: each poll iteration opens its own session and closes it before + the next ``asyncio.sleep`` — never holding one connection across the full + SSE lifetime. We patch ``get_session_factory`` to count session opens, + patch ``A2AService.get_task`` to return a non-terminal task, patch + ``asyncio.sleep`` to no-op, and make ``request.is_disconnected`` return + True after a few polls to terminate the stream quickly. The count of + session opens must exceed 1 (one per poll, not one for the lifetime).""" + + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + token = issue_agent_token(_AGENT_ID, "developer") + + # Count session opens across the SSE lifetime. + open_count = {"n": 0} + + def _factory() -> Any: + open_count["n"] += 1 + + class _Ctx: + async def __aenter__(self) -> MagicMock: + return MagicMock() + + async def __aexit__(self, *exc: object) -> None: + return None + + return _Ctx() + + monkeypatch.setattr(a2a_module, "get_session_factory", lambda: _factory) + + # Non-terminal fake task so the loop keeps polling. + fake_task = MagicMock() + fake_task.status.state = "in_progress" + fake_task.model_dump_json = MagicMock(return_value="{}") + monkeypatch.setattr( + a2a_module.A2AService, "get_task", AsyncMock(return_value=fake_task) + ) + + # No sleeping — drain the generator as fast as possible. + monkeypatch.setattr(a2a_module.asyncio, "sleep", AsyncMock(return_value=None)) + + # Disconnect after 3 polls so the stream terminates. + disconnect_after = {"remaining": 3} + + async def _fake_is_disconnected() -> bool: + if disconnect_after["remaining"] <= 0: + return True + disconnect_after["remaining"] -= 1 + return False + + # The route reads request.is_disconnected(); patch it on the request via + # the Starlette request. We patch the Request.is_disconnected property. + monkeypatch.setattr( + "fastapi.Request.is_disconnected", + lambda _self: _fake_is_disconnected(), + ) + + r = await a2a_client.get( + "/api/a2a/tasks/some-task/subscribe", + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "developer", + "X-Agent-Token": token, + }, + ) + # Drain the SSE stream so the generator runs to completion. + assert r.status_code == _HTTP_200 + # Consume the body (the SSE stream finishes once is_disconnected returns + # True on the 4th check). + _ = await r.aread() + + # 3 polls + 1 initial validation = 4 session opens (one per query, none + # held across the lifetime). The key assertion: more than one session + # was opened — proving the request-scoped session is gone. + assert open_count["n"] > 1, ( + f"only {open_count['n']} session open(s) — the route is holding a " + "single request-scoped session for the full SSE lifetime (pool " + "exhaustion vector)." + ) diff --git a/tests/unit/api/test_dashboard_auditor_auth.py b/tests/unit/api/test_dashboard_auditor_auth.py new file mode 100644 index 00000000..a96a30eb --- /dev/null +++ b/tests/unit/api/test_dashboard_auditor_auth.py @@ -0,0 +1,200 @@ +"""F025: dashboard auditor flag/report mutating routes must be gated to the +Auditor or CEO. + +``create_auditor_flag`` / ``resolve_auditor_flag`` / ``create_auditor_report`` +/ ``send_auditor_report`` previously took only ``db: DbSession`` — no +``CurrentAgentContext``, no role check — so any unauthenticated caller could +create/resolve flags and mark reports as sent to the CEO. The fix mirrors +``roboco/api/routes/playbooks.py::_require_curator``: a ``CurrentAgentContext`` +dependency plus a coarse role gate that admits only ``AUDITOR`` and ``CEO``. +""" + +from __future__ import annotations + +from http import HTTPStatus +from typing import TYPE_CHECKING +from uuid import uuid4 + +import pytest +import pytest_asyncio +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.api.deps import get_agent_context, get_db +from roboco.api.routes.dashboard import router as dashboard_router +from roboco.models import AgentRole +from roboco.models.permissions import AgentContext +from roboco.services.dashboard import reset_storage + +if TYPE_CHECKING: + from collections.abc import AsyncGenerator, AsyncIterator + + from sqlalchemy.ext.asyncio import AsyncSession + + +def _override_agent(role: AgentRole) -> AgentContext: + return AgentContext(agent_id=uuid4(), role=role, team=None) + + +@pytest_asyncio.fixture +async def auditor_client( + db_session: AsyncSession, +) -> AsyncIterator[AsyncClient]: + """A client authenticated as the Auditor (the legitimate caller).""" + reset_storage() + app = FastAPI() + app.include_router(dashboard_router, prefix="/api/dashboard") + + async def _override_db() -> AsyncGenerator[AsyncSession]: + yield db_session + + app.dependency_overrides[get_db] = _override_db + app.dependency_overrides[get_agent_context] = lambda: _override_agent( + AgentRole.AUDITOR + ) + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + yield client + app.dependency_overrides.clear() + + +@pytest_asyncio.fixture +async def dev_client( + db_session: AsyncSession, +) -> AsyncIterator[AsyncClient]: + """A client authenticated as a Developer — must NOT be able to mutate + auditor flags/reports.""" + reset_storage() + app = FastAPI() + app.include_router(dashboard_router, prefix="/api/dashboard") + + async def _override_db() -> AsyncGenerator[AsyncSession]: + yield db_session + + app.dependency_overrides[get_db] = _override_db + app.dependency_overrides[get_agent_context] = lambda: _override_agent( + AgentRole.DEVELOPER + ) + transport = ASGITransport(app=app) + async with AsyncClient(transport=transport, base_url="http://test") as client: + yield client + app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# Legitimate caller (Auditor) succeeds +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_auditor_can_create_flag(auditor_client: AsyncClient) -> None: + response = await auditor_client.post( + "/api/dashboard/auditor/flags", + json={ + "severity": "warning", + "category": "quality", + "title": "Flag", + "description": "x", + }, + ) + assert response.status_code == HTTPStatus.CREATED + + +@pytest.mark.asyncio +async def test_auditor_can_create_report(auditor_client: AsyncClient) -> None: + response = await auditor_client.post( + "/api/dashboard/auditor/reports", + json={ + "report_type": "weekly", + "title": "T", + "summary": "s", + "sections": [], + }, + ) + assert response.status_code == HTTPStatus.CREATED + + +@pytest.mark.asyncio +async def test_auditor_can_send_report(auditor_client: AsyncClient) -> None: + create = await auditor_client.post( + "/api/dashboard/auditor/reports", + json={ + "report_type": "weekly", + "title": "T", + "summary": "s", + "sections": [], + }, + ) + rid = create.json()["id"] + response = await auditor_client.post(f"/api/dashboard/auditor/reports/{rid}/send") + assert response.status_code == HTTPStatus.OK + + +@pytest.mark.asyncio +async def test_auditor_can_resolve_flag(auditor_client: AsyncClient) -> None: + create = await auditor_client.post( + "/api/dashboard/auditor/flags", + json={ + "severity": "warning", + "category": "quality", + "title": "F", + "description": "x", + }, + ) + flag_id = create.json()["id"] + response = await auditor_client.put( + f"/api/dashboard/auditor/flags/{flag_id}/resolve", + params={"notes": "fixed"}, + ) + assert response.status_code == HTTPStatus.OK + + +# --------------------------------------------------------------------------- +# Forged caller (Developer) is rejected with 403 +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_developer_cannot_create_flag(dev_client: AsyncClient) -> None: + response = await dev_client.post( + "/api/dashboard/auditor/flags", + json={ + "severity": "warning", + "category": "quality", + "title": "F", + "description": "x", + }, + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_developer_cannot_resolve_flag(dev_client: AsyncClient) -> None: + # The role gate fires before the route checks flag existence, so a random + # UUID is enough to prove the dev is rejected at the gate. + response = await dev_client.put( + f"/api/dashboard/auditor/flags/{uuid4()}/resolve", + params={"notes": "fixed"}, + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_developer_cannot_create_report(dev_client: AsyncClient) -> None: + response = await dev_client.post( + "/api/dashboard/auditor/reports", + json={ + "report_type": "weekly", + "title": "T", + "summary": "s", + "sections": [], + }, + ) + assert response.status_code == HTTPStatus.FORBIDDEN + + +@pytest.mark.asyncio +async def test_developer_cannot_send_report(dev_client: AsyncClient) -> None: + response = await dev_client.post( + f"/api/dashboard/auditor/reports/{uuid4()}/send", + ) + assert response.status_code == HTTPStatus.FORBIDDEN diff --git a/tests/unit/api/test_middleware.py b/tests/unit/api/test_middleware.py index 9efcd134..8a304238 100644 --- a/tests/unit/api/test_middleware.py +++ b/tests/unit/api/test_middleware.py @@ -38,6 +38,7 @@ from roboco.services.base import ( from roboco.services.base import ( ValidationError as ServiceValidationError, ) +from structlog.testing import capture_logs # --------------------------------------------------------------------------- # get_status_code @@ -275,3 +276,107 @@ def test_request_validation_handler_returns_422_with_details() -> None: body = response.json() assert "detail" in body assert "body" in body + + +# --------------------------------------------------------------------------- +# F022: secret scrubbing in the 422 log line +# --------------------------------------------------------------------------- + + +class _SecretBody(BaseModel): + """Module-level model so FastAPI can resolve the annotation under + `from __future__ import annotations` (function-local classes with complex + field types aren't resolvable from the function's module globals).""" + + name: str + git_token: str | None = None + api_key: str | None = None + auth_token: str | None = None + nested: dict[str, Any] | None = None + + +def test_request_validation_handler_scrubs_secrets_from_log() -> None: + """F022: a 422 on a secret-bearing request must not dump the plaintext + secret into the log line — only the redacted placeholder. The 422 + response body is unchanged (the client sent those values; the server + only redacts its own log).""" + + app = FastAPI() + setup_middleware(app) + + @app.post("/project") + async def _create(_data: _SecretBody) -> Any: + return {"ok": True} + + secret_pat = "ghp_livesecret_123456" + secret_key = "ollama-key-do-not-log" + secret_token = "bearer-should-not-leak" + payload = { + # Missing required `name` -> 422, but the secret fields are still + # parsed into rve.body and would be logged verbatim without the scrub. + "git_token": secret_pat, + "api_key": secret_key, + "auth_token": secret_token, + "nested": {"git_token": "nested-secret-abc", "safe": "keep"}, + } + + client = TestClient(app, raise_server_exceptions=False) + with capture_logs() as logs: + response = client.post("/project", json=payload) + + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + + # The response body is NOT scrubbed (the client sent these values). + resp_body = response.json() + assert resp_body["body"]["git_token"] == secret_pat + assert resp_body["body"]["api_key"] == secret_key + + # Exactly one "Request validation failed" warning was emitted. + fails = [e for e in logs if e["event"] == "Request validation failed"] + assert len(fails) == 1 + logged_body = fails[0]["body"] + + # The log line must not contain any of the plaintext secrets. + assert secret_pat not in str(logged_body) + assert secret_key not in str(logged_body) + assert secret_token not in str(logged_body) + assert "nested-secret-abc" not in str(logged_body) + + # The redaction placeholder appears for each secret field (so ops can see + # WHICH secret field was present), and the per-field errors are still + # logged (they don't carry secrets). + assert logged_body["git_token"] == "***REDACTED***" + assert logged_body["api_key"] == "***REDACTED***" + assert logged_body["auth_token"] == "***REDACTED***" + assert logged_body["nested"]["git_token"] == "***REDACTED***" + assert logged_body["nested"]["safe"] == "keep" # non-secret preserved + assert "errors" in fails[0] + + +def test_request_validation_handler_log_preserves_non_secret_fields() -> None: + """F022: non-secret fields in the body are still logged in full — only + the known credential-looking field names are redacted.""" + + app = FastAPI() + setup_middleware(app) + + @app.post("/project") + async def _create(_data: _SecretBody) -> Any: + return {"ok": True} + + client = TestClient(app, raise_server_exceptions=False) + # `title` is not a field on _SecretBody -> 422, and `title` is non-secret + # so it should still appear in the log; `git_token` is secret and must be + # redacted. + with capture_logs() as logs: + response = client.post( + "/project", + json={"title": "visible-title", "git_token": "ghp_secret_xyz"}, + ) + assert response.status_code == HTTPStatus.UNPROCESSABLE_ENTITY + fails = [e for e in logs if e["event"] == "Request validation failed"] + assert len(fails) == 1 + logged_body = fails[0]["body"] + assert logged_body["title"] == "visible-title" # non-secret preserved + assert logged_body["git_token"] == "***REDACTED***" # secret redacted + assert "ghp_secret_xyz" not in str(logged_body) diff --git a/tests/unit/api/test_orchestrator_auth.py b/tests/unit/api/test_orchestrator_auth.py new file mode 100644 index 00000000..e93628ec --- /dev/null +++ b/tests/unit/api/test_orchestrator_auth.py @@ -0,0 +1,204 @@ +"""F026: orchestrator control routes (/api/orchestrator/*) must be gated to +the CEO/operator identity. + +``spawn_agent`` / ``stop_agent`` / ``resolve_wait`` / ``mark_waiting`` previously +took no auth dependency at all — any client that could reach the API could +spawn, stop, mark-waiting, or resolve-wait any agent. The fix mirrors the +F004 panel-token guard (DB-free): bind the presented ``X-Agent-ID`` to a +verified HMAC token and assert the role is CEO. In dev (header-trust) mode a +missing token is a no-op (the panel/operator flow keeps working), but a +presented-but-forged token is still rejected — same contract as the v1 flow +role guards and the do router (F003). +""" + +from __future__ import annotations + +from typing import TYPE_CHECKING +from unittest.mock import AsyncMock, MagicMock +from uuid import uuid4 + +import pytest +import pytest_asyncio +from fastapi import FastAPI +from httpx import ASGITransport, AsyncClient +from roboco.agents_config import issue_agent_token +from roboco.api.deps import _ServiceHolder, set_orchestrator +from roboco.api.routes.orchestrator import router as orch_router + +if TYPE_CHECKING: + from collections.abc import AsyncIterator + +_SECRET = "test-secret-for-orch-auth" +_AGENT_ID = "00000000-0000-0000-0000-000000000001" +_HTTP_201 = 201 +_HTTP_204 = 204 +_HTTP_401 = 401 +_HTTP_403 = 403 + + +def _mock_orchestrator() -> MagicMock: + orch = MagicMock() + orch.spawn_agent = AsyncMock( + return_value=MagicMock( + agent_id=_AGENT_ID, + state=MagicMock(value="starting"), + current_task_id=None, + error_count=0, + started_at=None, + waiting_for=None, + ) + ) + orch.stop_agent = AsyncMock(return_value=None) + return orch + + +@pytest_asyncio.fixture +async def orch_client() -> AsyncIterator[tuple[AsyncClient, MagicMock]]: + app = FastAPI() + app.include_router(orch_router, prefix="/api/orchestrator") + orch = _mock_orchestrator() + set_orchestrator(orch) + async with AsyncClient( + transport=ASGITransport(app=app), base_url="http://test" + ) as client: + yield client, orch + _ServiceHolder.orchestrator = None + app.dependency_overrides.clear() + + +# --------------------------------------------------------------------------- +# Strict mode: token required +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_spawn_rejects_missing_token_when_required( + orch_client: tuple[AsyncClient, MagicMock], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Strict mode + no X-Agent-Token => 401, never reaches the orchestrator.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + client, orch = orch_client + r = await client.post( + f"/api/orchestrator/agents/{_AGENT_ID}/spawn", + headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "ceo"}, + ) + assert r.status_code == _HTTP_401 + orch.spawn_agent.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# Dev mode: forged token rejected, missing token is a no-op +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_spawn_rejects_forged_token_even_in_dev( + orch_client: tuple[AsyncClient, MagicMock], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A presented-but-forged token is rejected even in header-trust mode.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + client, orch = orch_client + r = await client.post( + f"/api/orchestrator/agents/{_AGENT_ID}/spawn", + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "ceo", + "X-Agent-Token": "forged-not-a-real-hmac", + }, + ) + assert r.status_code == _HTTP_401 + orch.spawn_agent.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_spawn_rejects_non_ceo_role( + orch_client: tuple[AsyncClient, MagicMock], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A developer (even with a validly-issued token) must not spawn/stop agents.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + client, orch = orch_client + dev_id = str(uuid4()) + token = issue_agent_token(dev_id, "developer") + r = await client.post( + f"/api/orchestrator/agents/{_AGENT_ID}/spawn", + headers={ + "X-Agent-ID": dev_id, + "X-Agent-Role": "developer", + "X-Agent-Token": token, + }, + ) + assert r.status_code == _HTTP_403 + orch.spawn_agent.assert_not_awaited() + + +# --------------------------------------------------------------------------- +# Legitimate CEO caller succeeds +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_spawn_accepts_valid_ceo_token( + orch_client: tuple[AsyncClient, MagicMock], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A valid CEO token passes the gate and reaches the orchestrator.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + client, orch = orch_client + token = issue_agent_token(_AGENT_ID, "ceo") + r = await client.post( + f"/api/orchestrator/agents/{_AGENT_ID}/spawn", + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "ceo", + "X-Agent-Token": token, + }, + ) + assert r.status_code == _HTTP_201 + orch.spawn_agent.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_stop_accepts_valid_ceo_token( + orch_client: tuple[AsyncClient, MagicMock], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The gate is wired into stop_agent too.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true") + client, orch = orch_client + token = issue_agent_token(_AGENT_ID, "ceo") + r = await client.post( + f"/api/orchestrator/agents/{_AGENT_ID}/stop", + headers={ + "X-Agent-ID": _AGENT_ID, + "X-Agent-Role": "ceo", + "X-Agent-Token": token, + }, + ) + assert r.status_code == _HTTP_204 + orch.stop_agent.assert_awaited_once() + + +@pytest.mark.asyncio +async def test_dev_mode_missing_token_still_succeeds( + orch_client: tuple[AsyncClient, MagicMock], + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Dev mode (no ROBOCO_AGENT_AUTH_REQUIRED) + no token => no-op, route runs. + Preserves the panel/operator flow in dev exactly as F003/F004 did.""" + monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET) + monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False) + client, orch = orch_client + r = await client.post( + f"/api/orchestrator/agents/{_AGENT_ID}/spawn", + headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "ceo"}, + ) + assert r.status_code == _HTTP_201 + orch.spawn_agent.assert_awaited_once()