feat(cloud-auth): enforce public scopes

tests/test_cloud_sdk_api.py

@@ -5,15 +5,15 @@ from __future__ import annotations
 import pytest
 
 from cloud.config import CloudConfig
+from cloud.auth import BearerCredential, ConfiguredBearerAuthProvider
 from cloud.plugins import PluginRegistry
 from cloud.pool import DevicePool
 from cloud.sdk.api import (
     AuthProvider,
-    NullAuthProvider,
     Principal,
     create_cloud_router,
 )
-from cloud.scheduler import TaskConstraints, TaskScheduler
+from cloud.scheduler import TaskScheduler
 from cloud.store import CloudStore
 from core.models import Device
 
@@ -177,12 +177,18 @@ def test_rejecting_auth_provider_blocks_every_route(tmp_path) -> None:
     assert client.get("/v1/devices").status_code == 401
     assert client.get("/v1/hosts").status_code == 401
     assert client.get("/v1/plugins").status_code == 401
-    assert client.post("/v1/plugins", json={
-        "name": "x",
-        "version": "1",
-        "entry_point_kind": "tool",
-        "target": "cloud.store:CloudStore",
-    }).status_code == 401
+    assert (
+        client.post(
+            "/v1/plugins",
+            json={
+                "name": "x",
+                "version": "1",
+                "entry_point_kind": "tool",
+                "target": "cloud.store:CloudStore",
+            },
+        ).status_code
+        == 401
+    )
 
 
 def test_default_null_auth_provider_is_used_when_omitted(tmp_path) -> None:
@@ -216,3 +222,68 @@ def test_submit_with_constraints(tmp_path) -> None:
     # And the device should match if we run assign() manually via the scheduler.
     status = client.get(f"/v1/tasks/{task_id}").json()
     assert status["status"] == "queued"
+
+
+@pytest.mark.parametrize(
+    ("method", "path", "payload", "required_scope"),
+    [
+        ("post", "/v1/tasks", {"goal": "x"}, "tasks:submit"),
+        ("get", "/v1/tasks/missing", None, "tasks:read"),
+        ("get", "/v1/devices", None, "pool:read"),
+        ("get", "/v1/hosts", None, "pool:read"),
+        ("get", "/v1/plugins", None, "plugins:read"),
+        (
+            "post",
+            "/v1/plugins",
+            {
+                "name": "scope-test",
+                "version": "1.0.0",
+                "entry_point_kind": "tool",
+                "target": "cloud.store:CloudStore",
+            },
+            "plugins:admin",
+        ),
+    ],
+)
+def test_every_public_route_enforces_its_scope(
+    tmp_path,
+    method: str,
+    path: str,
+    payload: dict[str, object] | None,
+    required_scope: str,
+) -> None:
+    provider = ConfiguredBearerAuthProvider(
+        [
+            BearerCredential(
+                principal_id="integrator",
+                token="scoped-token",
+                scopes=frozenset({required_scope}),
+            ),
+            BearerCredential(
+                principal_id="under-scoped",
+                token="wrong-token",
+                scopes=frozenset(),
+            ),
+        ]
+    )
+    app, _, _, _ = _build_app(tmp_path, auth_provider=provider)
+    client = _client_for(app)
+
+    unauthorized = client.request(method, path, json=payload)
+    forbidden = client.request(
+        method,
+        path,
+        json=payload,
+        headers={"Authorization": "Bearer wrong-token"},
+    )
+    authorized = client.request(
+        method,
+        path,
+        json=payload,
+        headers={"Authorization": "Bearer scoped-token"},
+    )
+
+    assert unauthorized.status_code == 401
+    assert unauthorized.headers["www-authenticate"] == "Bearer"
+    assert forbidden.status_code == 403
+    assert authorized.status_code not in {401, 403}