feat(agents): canvas routes join the agent-token allowlist (#1800 slice 2)

tests/test_auth_middleware.py

@@ -9,6 +9,7 @@
 
 from tinyagentos.auth_middleware import (
     AuthMiddleware,
+    _is_agent_canvas_path,
     _is_exempt,
     _is_loopback_client,
 )
@@ -265,5 +266,125 @@ async def test_prepare_shutdown_denied_from_remote(self):
 
         resp = await middleware.dispatch(req, call_next)
 
+        assert resp.status_code == 401
+        call_next.assert_not_awaited()
+
+
+class TestIsAgentCanvasPath:
+    def test_list_elements_get_allowed(self):
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements") is True
+
+    def test_create_element_post_allowed(self):
+        assert _is_agent_canvas_path("POST", "/api/projects/proj-1/canvas/elements") is True
+
+    def test_delete_element_allowed(self):
+        assert _is_agent_canvas_path("DELETE", "/api/projects/proj-1/canvas/elements/el-1") is True
+
+    def test_snapshot_png_allowed(self):
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshot.png") is True
+
+    def test_snapshot_tldr_allowed(self):
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshot.tldr") is True
+
+    def test_stream_allowed(self):
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/stream") is True
+
+    def test_update_element_patch_not_allowed(self):
+        assert _is_agent_canvas_path("PATCH", "/api/projects/proj-1/canvas/elements/el-1") is False
+
+    def test_permissions_patch_not_allowed(self):
+        assert _is_agent_canvas_path("PATCH", "/api/projects/proj-1/canvas/permissions/agent-1") is False
+
+    def test_extra_path_segment_not_allowed(self):
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements/el-1/extra") is False
+
+    def test_wrong_method_not_allowed(self):
+        assert _is_agent_canvas_path("POST", "/api/projects/proj-1/canvas/elements/el-1") is False
+
+    def test_nested_element_path_not_allowed(self):
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements/x/y") is False
+
+    def test_single_element_get_not_allowed(self):
+        # There is no GET /elements/{id} route in the allowlist; only the
+        # collection GET and the DELETE of a single element are permitted.
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/elements/el-1") is False
+
+    def test_snapshot_dot_is_literal_not_wildcard(self):
+        # The dot in snapshot.png / snapshot.tldr must be a literal, not a regex
+        # wildcard: a near-miss like snapshotXpng must NOT slip through the
+        # agent-token allowlist onto a session-only surface.
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshotXpng") is False
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshot_png") is False
+        assert _is_agent_canvas_path("GET", "/api/projects/proj-1/canvas/snapshotXtldr") is False
+
+
+class TestCanvasAgentTokenDispatch:
+    @pytest.mark.asyncio
+    async def test_canvas_list_elements_bearer_passes(self):
+        middleware = AuthMiddleware(app=MagicMock())
+        auth_mgr = _default_auth_mgr()
+        auth_mgr.validate_local_token.return_value = False
+        req = _request(
+            method="GET",
+            path="/api/projects/proj-1/canvas/elements",
+            headers={"authorization": "Bearer registry-jwt"},
+            auth_mgr=auth_mgr,
+        )
+        call_next = AsyncMock(return_value=JSONResponse({"elements": []}))
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 200
+        assert req.state.via == "registry_jwt_candidate"
+        call_next.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_canvas_delete_element_bearer_passes(self):
+        middleware = AuthMiddleware(app=MagicMock())
+        auth_mgr = _default_auth_mgr()
+        auth_mgr.validate_local_token.return_value = False
+        req = _request(
+            method="DELETE",
+            path="/api/projects/proj-1/canvas/elements/el-1",
+            headers={"authorization": "Bearer registry-jwt"},
+            auth_mgr=auth_mgr,
+        )
+        call_next = AsyncMock(return_value=JSONResponse({"ok": True}))
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 200
+        assert req.state.via == "registry_jwt_candidate"
+        call_next.assert_awaited_once()
+
+    @pytest.mark.asyncio
+    async def test_canvas_permissions_patch_requires_session(self):
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            method="PATCH",
+            path="/api/projects/proj-1/canvas/permissions/agent-1",
+            headers={"authorization": "Bearer registry-jwt"},
+            auth_mgr=_default_auth_mgr(),
+        )
+        call_next = AsyncMock()
+
+        resp = await middleware.dispatch(req, call_next)
+
+        assert resp.status_code == 401
+        call_next.assert_not_awaited()
+
+    @pytest.mark.asyncio
+    async def test_canvas_extra_segment_requires_session(self):
+        middleware = AuthMiddleware(app=MagicMock())
+        req = _request(
+            method="GET",
+            path="/api/projects/proj-1/canvas/elements/el-1/extra",
+            headers={"authorization": "Bearer registry-jwt"},
+            auth_mgr=_default_auth_mgr(),
+        )
+        call_next = AsyncMock()
+
+        resp = await middleware.dispatch(req, call_next)
+
         assert resp.status_code == 401
         call_next.assert_not_awaited()

tinyagentos/auth_middleware.py

@@ -58,11 +58,27 @@
     ("POST", re.compile(rf"^/api/projects/{_SEG}/tasks/{_SEG}/(claim|release|close|reopen)$")),
 )
 
+_AGENT_CANVAS_ROUTES = (
+    ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/elements$")),
+    ("POST", re.compile(rf"^/api/projects/{_SEG}/canvas/elements$")),
+    ("DELETE", re.compile(rf"^/api/projects/{_SEG}/canvas/elements/{_SEG}$")),
+    ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/snapshot\.png$")),
+    ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/snapshot\.tldr$")),
+    ("GET", re.compile(rf"^/api/projects/{_SEG}/canvas/stream$")),
+)
+
 
 def _is_agent_task_path(method: str, path: str) -> bool:
     """True only for the exact subset of task routes a project_tasks token may
     reach.  Strict method + anchored-regex match; everything else is excluded."""
     return any(m == method and rx.match(path) for m, rx in _AGENT_TASK_ROUTES)
+
+
+def _is_agent_canvas_path(method: str, path: str) -> bool:
+    """True only for the exact subset of canvas routes a canvas_read/canvas_write
+    token may reach.  Strict method + anchored-regex match; the PATCH elements
+    and PATCH permissions routes stay session-only."""
+    return any(m == method and rx.match(path) for m, rx in _AGENT_CANVAS_ROUTES)
 # Bundle assets and the SPA shell HTML must be reachable without auth so:
 #   1. The browser can install and cache the shell for offline / PWA use.
 #   2. After a backend restart the cached shell loads immediately without
@@ -255,6 +271,7 @@ async def dispatch(self, request: Request, call_next):
         if (
             path in _AGENT_TOKEN_PATHS
             or _is_agent_task_path(request.method, path)
+            or _is_agent_canvas_path(request.method, path)
         ) and auth_header.lower().startswith("bearer "):
             request.state.user_id = None
             request.state.is_admin = False