[scan] gate A2A/notification/stream agent-id deps under cloud auth (C1)

roboco/api/deps.py

@@ -8,7 +8,7 @@ from __future__ import annotations
 
 import contextlib
 import os
-from typing import TYPE_CHECKING, Annotated, Any
+from typing import TYPE_CHECKING, Annotated, Any, cast
 from uuid import UUID
 
 import structlog
@@ -133,31 +133,33 @@ OrchestratorDep = Annotated[AgentOrchestrator, Depends(get_orchestrator)]
 
 async def get_current_agent_id(
     db: DbSession,
+    response: Response,
     x_agent_id: Annotated[str | None, Header()] = None,
+    x_agent_role: Annotated[str | None, Header()] = None,
+    x_agent_team: Annotated[str | None, Header()] = None,
+    x_agent_token: Annotated[str | None, Header()] = None,
+    roboco_session: Annotated[str | None, Cookie(alias=SESSION_COOKIE_NAME)] = None,
 ) -> UUID:
-    """
+    """Resolve the caller's agent id. Dev (header-trust) keeps the historical
-    Get the current agent ID from request headers.
+    slug/UUID resolution unchanged; under cloud auth the request must present a
-
+    valid agent HMAC token or a CEO session cookie — a bare X-Agent-ID is a
-    Accepts either a UUID string or agent slug (e.g., "be-dev-1").
+    spoof and is rejected (see _cloud_auth_agent_context)."""
-    In production, this would validate a JWT token and extract the agent ID.
+    if settings.cloud_auth_enabled:
-    For now, we use a simple header-based approach for development.
+        ctx = await _cloud_auth_agent_context(
-
+            db,
-    Args:
+            response,
-        x_agent_id: Agent ID (UUID or slug) from X-Agent-ID header
+            x_agent_id,
-        db: Database session for slug resolution
+            x_agent_role,
-
+            x_agent_team,
-    Returns:
+            x_agent_token,
-        UUID of the current agent
+            roboco_session,
-
+        )
-    Raises:
+        return ctx.agent_id
-        HTTPException: If agent ID is missing or invalid/not found
-    """
     if not x_agent_id:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,
             detail="Missing X-Agent-ID header",
         )
-
     return await resolve_agent_id(x_agent_id, db)
 
 
@@ -166,23 +168,28 @@ CurrentAgentId = Annotated[UUID, Depends(get_current_agent_id)]
 
 
 async def get_current_agent_slug(
+    db: DbSession,
+    response: Response,
     x_agent_id: Annotated[str | None, Header()] = None,
+    x_agent_role: Annotated[str | None, Header()] = None,
+    x_agent_team: Annotated[str | None, Header()] = None,
+    x_agent_token: Annotated[str | None, Header()] = None,
+    roboco_session: Annotated[str | None, Cookie(alias=SESSION_COOKIE_NAME)] = None,
 ) -> str:
-    """
+    """Return the caller's agent slug. Dev: the header verbatim. Cloud auth:
-    Get the current agent slug from request headers.
+    the slug from the dual-path gate (a verified agent token resolves the real
-
+    slug; a CEO cookie resolves to 'ceo')."""
-    Unlike get_current_agent_id, this returns the slug directly without
+    if settings.cloud_auth_enabled:
-    resolving to UUID. Useful for A2A where we work with agent slugs.
+        ctx = await _cloud_auth_agent_context(
-
+            db,
-    Args:
+            response,
-        x_agent_id: Agent slug from X-Agent-ID header
+            x_agent_id,
-
+            x_agent_role,
-    Returns:
+            x_agent_team,
-        Agent slug string
+            x_agent_token,
-
+            roboco_session,
-    Raises:
+        )
-        HTTPException: If agent ID header is missing
+        return cast("str", ctx.slug)
-    """
     if not x_agent_id:
         raise HTTPException(
             status_code=status.HTTP_401_UNAUTHORIZED,

tests/unit/api/test_deps_cloud_auth_gate.py

@@ -0,0 +1,81 @@
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from fastapi import HTTPException, status
+from roboco.api import deps as d
+from roboco.api.deps import get_current_agent_slug
+
+
+async def _run(dep, headers, settings_on, monkeypatch):
+    monkeypatch.setattr(d.settings, "cloud_auth_enabled", settings_on)
+    db = AsyncMock()
+    response = AsyncMock()
+    # _cloud_auth_agent_context is the gate; stub it to assert it's reached
+    with patch.object(d, "_cloud_auth_agent_context", new=AsyncMock()) as m:
+        m.return_value = type(
+            "Ctx",
+            (),
+            {"agent_id": "00000000-0000-0000-0000-000000000000", "slug": "be-dev-1"},
+        )()
+        return await dep(
+            db=db,
+            response=response,
+            x_agent_id=headers.get("X-Agent-ID"),
+            x_agent_role=headers.get("X-Agent-Role"),
+            x_agent_team=headers.get("X-Agent-Team"),
+            x_agent_token=headers.get("X-Agent-Token"),
+            roboco_session=headers.get("roboco_session"),
+        ), m.called
+
+
+@pytest.mark.asyncio
+async def test_cloud_auth_spoof_bare_agent_id_rejected(monkeypatch):
+    # A bare X-Agent-ID with no token/cookie must not reach the gate body.
+    monkeypatch.setattr(d.settings, "cloud_auth_enabled", True)
+    db = AsyncMock()
+    response = AsyncMock()
+    with patch.object(d, "_cloud_auth_agent_context", new=AsyncMock()) as m:
+        m.side_effect = HTTPException(
+            status.HTTP_401_UNAUTHORIZED,
+            "Cloud auth is enabled — agent requests require a valid token.",
+        )
+        with pytest.raises(HTTPException) as exc:
+            await d.get_current_agent_slug(
+                db=db,
+                response=response,
+                x_agent_id="be-dev-1",
+                x_agent_role=None,
+                x_agent_team=None,
+                x_agent_token=None,
+                roboco_session=None,
+            )
+        assert exc.value.status_code == status.HTTP_401_UNAUTHORIZED
+
+
+@pytest.mark.asyncio
+async def test_cloud_auth_routes_through_dual_path(monkeypatch):
+    _, called = await _run(
+        get_current_agent_slug, {"X-Agent-ID": "be-dev-1"}, True, monkeypatch
+    )
+    assert called  # cloud mode delegates to _cloud_auth_agent_context
+
+
+@pytest.mark.asyncio
+async def test_dev_mode_unchanged_slug_returns_header(monkeypatch):
+    monkeypatch.setattr(d.settings, "cloud_auth_enabled", False)
+    db = AsyncMock()
+    response = AsyncMock()
+    with patch.object(d, "_cloud_auth_agent_context", new=AsyncMock()) as m:
+        slug = await d.get_current_agent_slug(
+            db=db,
+            response=response,
+            x_agent_id="be-dev-1",
+            x_agent_role=None,
+            x_agent_team=None,
+            x_agent_token=None,
+            roboco_session=None,
+        )
+    assert slug == "be-dev-1"
+    assert not m.called  # dev path does not invoke the cloud gate