feat(cloud-sdk): authenticate client requests

openspec/changes/cloud-control-plane-integration/tasks.md

@@ -64,7 +64,7 @@
 ## 8. Public SDK And Operational Delivery
 
 - [x] 8.1 Extend public task status models/routes with attempt count, lease expiry metadata, and terminal failure details without exposing lease credentials.
-- [ ] 8.2 Add bearer authentication and typed authorization errors to `CloudClient` while preserving injectable HTTP clients for tests.
+- [x] 8.2 Add bearer authentication and typed authorization errors to `CloudClient` while preserving injectable HTTP clients for tests.
 - [ ] 8.3 Add container definitions and example environment configuration for the cloud API, PostgreSQL, and Host Agent without committing secrets.
 - [ ] 8.4 Document local SQLite startup, deployed PostgreSQL migration/startup, credential/scopes setup, Runtime AI Planner configuration, and shutdown/rollback procedures.
 - [ ] 8.5 Document the single scheduler-enabled control-plane limitation and the at-least-once device-side-effect trade-off.

packages/cloud-platform/cloud/sdk/client.py

@@ -13,6 +13,21 @@ from typing import Any
 import httpx
 
 
+class CloudAPIError(httpx.HTTPStatusError):
+    def __init__(self, response: httpx.Response, detail: str) -> None:
+        super().__init__(
+            f"cloud API request failed with status {response.status_code}: {detail}",
+            request=response.request,
+            response=response,
+        )
+        self.status_code = response.status_code
+        self.detail = detail
+
+
+class CloudAuthorizationError(CloudAPIError):
+    pass
+
+
 class CloudClient:
     """A minimal Python wrapper for the platform SDK's ``/v1`` routes."""
 
@@ -22,9 +37,15 @@ class CloudClient:
         *,
         http_client: httpx.Client | Any = None,
         api_prefix: str = "/v1",
+        token: str | None = None,
+        auth: httpx.Auth | None = None,
     ) -> None:
+        if token is not None and auth is not None:
+            raise ValueError("token and auth are mutually exclusive")
         self._base_url = base_url.rstrip("/")
         self._api_prefix = api_prefix.rstrip("/")
+        self._headers = {"Authorization": f"Bearer {token}"} if token else None
+        self._auth = auth
         if http_client is None:
             self._http = httpx.Client(base_url=self._base_url)
             self._owns_client = True
@@ -61,32 +82,27 @@ class CloudClient:
                 "driver_type": driver_type,
                 "capability_tags": list(capability_tags or []),
             }
-        resp = self._http.post(self._url("/tasks"), json=payload)
+        resp = self._request("POST", "/tasks", json=payload)
-        resp.raise_for_status()
         return resp.json()
 
     def get_task_status(self, task_id: str) -> dict[str, Any]:
-        resp = self._http.get(self._url(f"/tasks/{task_id}"))
+        resp = self._request("GET", f"/tasks/{task_id}")
-        resp.raise_for_status()
         return resp.json()
 
     # ----------------------------------------------------------------- devices
 
     def list_devices(self) -> list[dict[str, Any]]:
-        resp = self._http.get(self._url("/devices"))
+        resp = self._request("GET", "/devices")
-        resp.raise_for_status()
         return resp.json()
 
     def list_hosts(self) -> list[dict[str, Any]]:
-        resp = self._http.get(self._url("/hosts"))
+        resp = self._request("GET", "/hosts")
-        resp.raise_for_status()
         return resp.json()
 
     # ----------------------------------------------------------------- plugins
 
     def list_plugins(self) -> list[dict[str, Any]]:
-        resp = self._http.get(self._url("/plugins"))
+        resp = self._request("GET", "/plugins")
-        resp.raise_for_status()
         return resp.json()
 
     def register_plugin(
@@ -103,11 +119,38 @@ class CloudClient:
             "entry_point_kind": entry_point_kind,
             "target": target,
         }
-        resp = self._http.post(self._url("/plugins"), json=payload)
+        resp = self._request("POST", "/plugins", json=payload)
-        resp.raise_for_status()
         return resp.json()
 
     # ------------------------------------------------------------------ helpers
 
     def _url(self, path: str) -> str:
         return f"{self._base_url}{self._api_prefix}{path}"
+
+    def _request(
+        self,
+        method: str,
+        path: str,
+        *,
+        json: dict[str, Any] | None = None,
+    ) -> httpx.Response:
+        response = self._http.request(
+            method,
+            self._url(path),
+            json=json,
+            headers=self._headers,
+            auth=self._auth,
+        )
+        if response.is_success:
+            return response
+        try:
+            payload = response.json()
+        except ValueError:
+            payload = {}
+        detail = payload.get("detail") or "request rejected"
+        error_type = (
+            CloudAuthorizationError
+            if response.status_code in {401, 403}
+            else CloudAPIError
+        )
+        raise error_type(response, str(detail))

tests/test_cloud_client.py

@@ -3,12 +3,14 @@
 from __future__ import annotations
 
 import pytest
+import httpx
 
+from cloud.auth import BearerCredential, ConfiguredBearerAuthProvider
 from cloud.config import CloudConfig
 from cloud.plugins import PluginRegistry
 from cloud.pool import DevicePool
 from cloud.sdk.api import create_cloud_router
-from cloud.sdk.client import CloudClient
+from cloud.sdk.client import CloudAuthorizationError, CloudClient
 from cloud.scheduler import TaskScheduler
 from cloud.store import CloudStore
 from core.models import Device
@@ -30,7 +32,7 @@ def _config() -> CloudConfig:
     )
 
 
-def _client_and_pool(tmp_path):
+def _client_and_pool(tmp_path, *, auth_provider=None, token: str | None = None):
     store = CloudStore(tmp_path / "cloud.sqlite3")
     pool = DevicePool(store, _config())
     scheduler = TaskScheduler(pool, store, _config())
@@ -41,10 +43,15 @@ def _client_and_pool(tmp_path):
             pool=pool,
             scheduler=scheduler,
             plugin_registry=plugin_registry,
+            auth_provider=auth_provider,
         )
     )
     test_client = TestClient(app)
-    cloud_client = CloudClient("http://testserver", http_client=test_client)
+    cloud_client = CloudClient(
+        "http://testserver",
+        http_client=test_client,
+        token=token,
+    )
     return cloud_client, pool
 
 
@@ -111,8 +118,107 @@ def test_client_submit_with_constraints(tmp_path) -> None:
 
 
 def test_client_unknown_task_raises(tmp_path) -> None:
-    import httpx
-
     client, _ = _client_and_pool(tmp_path)
     with pytest.raises(httpx.HTTPStatusError):
         client.get_task_status("does-not-exist")
+
+
+def test_client_applies_bearer_token_to_every_public_method(tmp_path) -> None:
+    token = "sdk-secret"
+    provider = ConfiguredBearerAuthProvider(
+        [
+            BearerCredential(
+                principal_id="sdk",
+                token=token,
+                scopes=frozenset(
+                    {
+                        "tasks:submit",
+                        "tasks:read",
+                        "pool:read",
+                        "plugins:read",
+                        "plugins:admin",
+                    }
+                ),
+            )
+        ]
+    )
+    client, _ = _client_and_pool(
+        tmp_path,
+        auth_provider=provider,
+        token=token,
+    )
+
+    task_id = client.submit_task(goal="authenticated")["task_id"]
+    assert client.get_task_status(task_id)["status"] == "queued"
+    assert client.list_devices() == []
+    assert client.list_hosts() == []
+    assert client.list_plugins() == []
+    assert client.register_plugin(
+        name="authenticated-plugin",
+        version="1.0.0",
+        entry_point_kind="tool",
+        target="cloud.store:CloudStore",
+    )["name"] == "authenticated-plugin"
+
+
+def test_client_raises_typed_authorization_error_without_exposing_token(
+    tmp_path,
+) -> None:
+    provider = ConfiguredBearerAuthProvider(
+        [BearerCredential(principal_id="sdk", token="valid-token")]
+    )
+    client, _ = _client_and_pool(
+        tmp_path,
+        auth_provider=provider,
+        token="invalid-secret-token",
+    )
+
+    with pytest.raises(CloudAuthorizationError) as error:
+        client.list_devices()
+
+    assert error.value.status_code == 401
+    assert "invalid-secret-token" not in str(error.value)
+
+
+def test_client_supports_injected_httpx_auth(tmp_path) -> None:
+    class StaticAuth(httpx.Auth):
+        def auth_flow(self, request):
+            request.headers["Authorization"] = "Bearer injected-token"
+            yield request
+
+    provider = ConfiguredBearerAuthProvider(
+        [
+            BearerCredential(
+                principal_id="sdk",
+                token="injected-token",
+                scopes=frozenset({"pool:read"}),
+            )
+        ]
+    )
+    store = CloudStore(tmp_path / "auth.sqlite3")
+    pool = DevicePool(store, _config())
+    app = FastAPI()
+    app.include_router(
+        create_cloud_router(
+            pool=pool,
+            scheduler=TaskScheduler(pool, store, _config()),
+            plugin_registry=PluginRegistry(store),
+            auth_provider=provider,
+        )
+    )
+    client = CloudClient(
+        "http://testserver",
+        http_client=TestClient(app),
+        auth=StaticAuth(),
+    )
+
+    assert client.list_devices() == []
+
+
+def test_client_rejects_token_and_auth_together() -> None:
+    with pytest.raises(ValueError, match="mutually exclusive"):
+        CloudClient(
+            "https://cloud.example",
+            token="secret",
+            auth=httpx.BasicAuth("user", "password"),
+        )