[F003,F004,F014] enforce HMAC agent-token gate on do routes + WebSocket streams

F003/F014: /api/v1/do/* only required X-Agent-ID (UUID) — no token check,
unlike the flow routers' role guards. A forged X-Agent-ID passed. Added
require_any_authenticated_agent (token-only; do router serves all roles)
and applied it as a router-level dependency. Binds X-Agent-ID to a verified
HMAC token when ROBOCO_AGENT_AUTH_REQUIRED=true; rejects a forged token
even in dev mode.

F004: /ws/* per-agent streams (channels/agents/sessions/notifications)
never read the nginx-injected X-Agent-Token, so in strict mode an agent on
the Docker network could subscribe to another agent's notifications with
no auth. Added _require_panel_token verifying the CEO panel token against
the CEO identity; wired into all four per-agent streams (system stream
stays operator-only per its docstring). Same strict/dev contract.

TDD: RED tests watched fail (no gate -> 200/accept), then GREEN. ruff+mypy
clean; 399 api/mcp + 29 WS tests green, no regressions.

roboco/api/routes/v1/_role_dep.py

@@ -61,6 +61,41 @@ require_auditor = _require_roles(frozenset({Role.AUDITOR}))
 require_pr_reviewer = _require_roles(frozenset({Role.PR_REVIEWER}))
 
 
+def _require_authenticated_agent() -> params.Depends:
+    """Token-only guard for the content-tool (do) router (F003/F014).
+
+    The do router serves every role — content tools are role-uniform, with
+    per-role removal handled in the spawn manifest — so, unlike the flow
+    routers, there is no single role to assert. But it must still bind the
+    presented ``X-Agent-ID`` to a verified HMAC token when
+    ``ROBOCO_AGENT_AUTH_REQUIRED=true`` and reject a forged token even in
+    dev mode, exactly as the flow role guards do. Without this the
+    ``/api/v1/do/*`` endpoints were the one agent-gateway path that
+    accepted a forged ``X-Agent-ID`` with no token check — a weaker gate
+    than ``/api/v1/flow/*``. The role/team headers are optional (the do
+    MCP server sends role but not team); they only feed the HMAC payload,
+    so a missing team is the empty-string team the token was issued with.
+    """
+
+    def _check(
+        x_agent_id: Annotated[str, Header(alias="X-Agent-ID")],
+        x_agent_role: Annotated[str | None, Header(alias="X-Agent-Role")] = None,
+        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:
+        from roboco.api.deps import _check_agent_auth_token
+
+        _check_agent_auth_token(
+            x_agent_id, x_agent_role or "", x_agent_team, x_agent_token
+        )
+
+    return cast("params.Depends", Depends(_check))
+
+
+# The do router serves all roles, so this is token-only (no role assertion).
+require_any_authenticated_agent = _require_authenticated_agent()
+
+
 def envelope_to_response(env: Envelope, request: Request) -> dict[str, Any]:
     """Stamp the request's correlation_id onto the envelope and return wire-dict.
 

roboco/api/routes/v1/do.py

@@ -6,7 +6,10 @@ from uuid import UUID
 from fastapi import APIRouter, Depends, Header, Request
 
 from roboco.api.deps import get_content_actions
-from roboco.api.routes.v1._role_dep import envelope_to_response
+from roboco.api.routes.v1._role_dep import (
+    envelope_to_response,
+    require_any_authenticated_agent,
+)
 from roboco.api.schemas.v1.do import (
     ApprovePlaybookRequest,
     ArchivePlaybookRequest,
@@ -31,7 +34,14 @@ from roboco.api.schemas.v1.do import (
 )
 from roboco.services.gateway.content_actions import ContentActions
 
-router = APIRouter(prefix="/api/v1/do", tags=["v1-do"])
+router = APIRouter(
+    prefix="/api/v1/do",
+    tags=["v1-do"],
+    # F003/F014: bind X-Agent-ID to a verified HMAC token — same gate the
+    # flow routers enforce via their role guards. The do router serves all
+    # roles, so this is token-only (no role assertion).
+    dependencies=[require_any_authenticated_agent],
+)
 
 _AgentIdHeader = Annotated[UUID, Header(alias="X-Agent-ID")]
 _ContentActionsDep = Annotated[ContentActions, Depends(get_content_actions)]

roboco/api/websocket.py

@@ -21,6 +21,8 @@ from uuid import UUID
 import httpx
 from fastapi import APIRouter, WebSocket, WebSocketDisconnect, status
 
+from roboco.agents_config import CEO_AGENT_ID, verify_agent_token
+from roboco.api.deps import _auth_required
 from roboco.config import settings
 from roboco.db.base import get_db
 from roboco.services.repositories import resolve_agent_uuid
@@ -28,6 +30,29 @@ from roboco.services.repositories import resolve_agent_uuid
 router = APIRouter()
 
 
+async def _require_panel_token(websocket: WebSocket) -> bool:
+    """F004: bind a per-agent WS upgrade to the panel/CEO HMAC token.
+
+    The /ws/* streams are operator-only — the control panel is the sole WS
+    client (agents use MCP verbs, not WS), and nginx injects the CEO panel
+    token as ``X-Agent-Token`` on /ws/ upgrades. Without verifying it the
+    per-agent endpoints (channels/agents/sessions/notifications) accepted a
+    bare ``agent_id`` query param with no auth, so in strict mode
+    (``ROBOCO_AGENT_AUTH_REQUIRED=true``) an agent on the Docker network
+    could hit e.g. ``/ws/notifications/{id}`` directly and subscribe to
+    another agent's notifications. This gate requires + verifies the token
+    against the CEO identity in strict mode, and rejects a presented-but-
+    forged token even in dev mode — the same contract as the HTTP
+    ``_check_agent_auth_token`` role gates. Returns True to proceed, False
+    to close with a policy violation (caller closes the socket).
+    """
+    token = websocket.headers.get("x-agent-token")
+    if _auth_required() and not token:
+        return False
+    # A missing token in dev mode proceeds; a presented token must verify.
+    return not (token and not verify_agent_token(token, CEO_AGENT_ID, "ceo", ""))
+
+
 # =============================================================================
 # Connection Manager
 # =============================================================================
@@ -262,6 +287,10 @@ async def channel_stream(
 
     Clients receive real-time messages for the channel.
     """
+    # F004: verify the panel/CEO token before any subject lookup.
+    if not await _require_panel_token(websocket):
+        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+        return
     # Get agent ID from query params (or auth in production)
     agent_id_str = websocket.query_params.get("agent_id")
     if not agent_id_str:
@@ -318,6 +347,10 @@ async def agent_stream(
 
     Clients receive real-time LLM output from the agent.
     """
+    # F004: verify the panel/CEO token before any subject lookup.
+    if not await _require_panel_token(websocket):
+        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+        return
     # Get viewer agent ID
     viewer_id_str = websocket.query_params.get("viewer_id")
     if not viewer_id_str:
@@ -365,6 +398,10 @@ async def session_stream(
 
     Clients receive real-time messages for a specific session.
     """
+    # F004: verify the panel/CEO token before any subject lookup.
+    if not await _require_panel_token(websocket):
+        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+        return
     agent_id_str = websocket.query_params.get("agent_id")
     if not agent_id_str:
         await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
@@ -410,6 +447,10 @@ async def notification_stream(
 
     Agents receive real-time notifications via this stream.
     """
+    # F004: verify the panel/CEO token before any subject lookup.
+    if not await _require_panel_token(websocket):
+        await websocket.close(code=status.WS_1008_POLICY_VIOLATION)
+        return
     # Validate agent exists in database
     if not await validate_agent_exists(agent_id):
         await websocket.close(code=status.WS_1008_POLICY_VIOLATION)

tests/unit/api/routes/v1/test_do_auth.py

@@ -0,0 +1,97 @@
+"""F003/F014: /api/v1/do/* must enforce the same HMAC agent-token gate as
+the /api/v1/flow/* routers.
+
+The do router serves every role (content tools are role-uniform), so it has
+no single role to assert — but it must still bind the presented X-Agent-ID
+to a verified token when ROBOCO_AGENT_AUTH_REQUIRED=true and reject a forged
+token even in dev mode. Without this guard the content-tool endpoints were
+the one agent-gateway path that accepted a forged X-Agent-ID with no token
+check — a weaker gate than the flow routers' role guards.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock, MagicMock
+
+from fastapi import FastAPI
+from fastapi.testclient import TestClient
+from roboco.agents_config import issue_agent_token
+from roboco.api.deps import get_content_actions
+from roboco.api.routes.v1.do import router
+from roboco.services.gateway.content_actions import ContentActions
+
+if TYPE_CHECKING:
+    import pytest
+
+_HTTP_200 = 200
+_HTTP_401 = 401
+_AGENT_ID = "00000000-0000-0000-0000-000000000001"
+_SECRET = "test-secret-for-do-auth"
+
+
+def _build_app() -> FastAPI:
+    app = FastAPI()
+    app.include_router(router)
+    mock_actions = MagicMock(spec=ContentActions)
+    mock_env = MagicMock()
+    mock_env.as_dict.return_value = {"status": "ok", "next": "continue"}
+    mock_actions.commit = AsyncMock(return_value=mock_env)
+    app.dependency_overrides[get_content_actions] = lambda: mock_actions
+    return app
+
+
+def _commit_body() -> dict:
+    return {"message": "add user authentication endpoint"}
+
+
+def test_do_route_401_when_auth_required_and_no_token(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Strict mode: a do endpoint must require the token, not just X-Agent-ID."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true")
+    client = TestClient(_build_app())
+    r = client.post(
+        "/api/v1/do/commit",
+        json=_commit_body(),
+        headers={"X-Agent-ID": _AGENT_ID, "X-Agent-Role": "developer"},
+    )
+    assert r.status_code == _HTTP_401
+
+
+def test_do_route_rejects_forged_token_even_in_dev(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Even in header-trust mode, a presented-but-forged token is rejected."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False)
+    client = TestClient(_build_app())
+    r = client.post(
+        "/api/v1/do/commit",
+        json=_commit_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
+
+
+def test_do_route_accepts_valid_token(monkeypatch: pytest.MonkeyPatch) -> None:
+    """The good path: a valid HMAC token passes the guard and reaches the handler."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true")
+    token = issue_agent_token(_AGENT_ID, "developer")
+    client = TestClient(_build_app())
+    r = client.post(
+        "/api/v1/do/commit",
+        json=_commit_body(),
+        headers={
+            "X-Agent-ID": _AGENT_ID,
+            "X-Agent-Role": "developer",
+            "X-Agent-Token": token,
+        },
+    )
+    assert r.status_code == _HTTP_200

tests/unit/api/test_websocket_auth.py

@@ -0,0 +1,121 @@
+"""F004: WebSocket streams must enforce the HMAC panel/CEO token gate when
+ROBOCO_AGENT_AUTH_REQUIRED=true.
+
+The /ws/* streams are operator-only (the panel is the sole WS client; agents
+use MCP verbs, not WS). nginx injects the CEO panel token as X-Agent-Token on
+/ws/ upgrades, but the endpoints never read or verified it — so in strict mode
+an agent on the Docker network could hit /ws/notifications/{id} directly and
+subscribe to another agent's notifications with no auth. The fix binds each
+per-agent WS upgrade to the CEO token: require + verify it in strict mode, and
+reject a forged token even in dev mode (same contract as the HTTP role gates).
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from fastapi import WebSocketDisconnect, status
+from roboco.agents_config import CEO_AGENT_ID, issue_agent_token
+from roboco.api.websocket import agent_stream, notification_stream
+
+if TYPE_CHECKING:
+    import pytest as _pytest  # noqa: F401
+
+_SECRET = "test-secret-for-ws-auth"
+
+
+def _mock_ws(headers: dict[str, str] | None, query: dict[str, str] | None) -> MagicMock:
+    ws = MagicMock()
+    ws.accept = AsyncMock()
+    ws.close = AsyncMock()
+    ws.send_json = AsyncMock()
+    ws.send_text = AsyncMock()
+    # One pong then disconnect so the receive loop exits after a successful gate.
+    ws.receive_text = AsyncMock(side_effect=["ping", WebSocketDisconnect()])
+    ws.headers = headers or {}
+    ws.query_params = query or {}
+    return ws
+
+
+@pytest.mark.asyncio
+async def test_notification_stream_rejects_missing_token_when_required(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Strict mode + no X-Agent-Token => policy-violation close, never accepted."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true")
+    agent_id = uuid4()
+    ws = _mock_ws(headers={}, query={})
+    with pytest.MonkeyPatch.context() as mp:
+        mp.setattr(
+            "roboco.api.websocket.validate_agent_exists",
+            AsyncMock(return_value=True),
+        )
+        await notification_stream(ws, agent_id)
+    ws.close.assert_awaited_once()
+    assert ws.close.await_args.kwargs["code"] == status.WS_1008_POLICY_VIOLATION
+    ws.accept.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notification_stream_rejects_forged_token_even_in_dev(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Even in dev mode a presented-but-forged token is rejected."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.delenv("ROBOCO_AGENT_AUTH_REQUIRED", raising=False)
+    agent_id = uuid4()
+    ws = _mock_ws(headers={"x-agent-token": "forged-not-a-real-hmac"}, query={})
+    with pytest.MonkeyPatch.context() as mp:
+        mp.setattr(
+            "roboco.api.websocket.validate_agent_exists",
+            AsyncMock(return_value=True),
+        )
+        await notification_stream(ws, agent_id)
+    ws.close.assert_awaited_once()
+    assert ws.close.await_args.kwargs["code"] == status.WS_1008_POLICY_VIOLATION
+    ws.accept.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_notification_stream_accepts_valid_panel_token(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A valid CEO panel token passes the gate and the socket is accepted."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true")
+    token = issue_agent_token(CEO_AGENT_ID, "ceo", "")
+    agent_id = uuid4()
+    ws = _mock_ws(headers={"x-agent-token": token}, query={})
+    with pytest.MonkeyPatch.context() as mp:
+        mp.setattr(
+            "roboco.api.websocket.validate_agent_exists",
+            AsyncMock(return_value=True),
+        )
+        await notification_stream(ws, agent_id)
+    ws.accept.assert_awaited_once()
+    ws.close.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_agent_stream_rejects_missing_token_when_required(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """The gate is wired into agent_stream too (viewer_id query param path)."""
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_SECRET", _SECRET)
+    monkeypatch.setenv("ROBOCO_AGENT_AUTH_REQUIRED", "true")
+    target_id = uuid4()
+    viewer_id = uuid4()
+    ws = _mock_ws(headers={}, query={"viewer_id": str(viewer_id)})
+    with pytest.MonkeyPatch.context() as mp:
+        mp.setattr(
+            "roboco.api.websocket.validate_agent_exists",
+            AsyncMock(return_value=True),
+        )
+        await agent_stream(ws, target_id)
+    ws.close.assert_awaited_once()
+    assert ws.close.await_args.kwargs["code"] == status.WS_1008_POLICY_VIOLATION
+    ws.accept.assert_not_awaited()