diff --git a/openspec/changes/cloud-control-plane-integration/tasks.md b/openspec/changes/cloud-control-plane-integration/tasks.md index 3666662..e2c125b 100644 --- a/openspec/changes/cloud-control-plane-integration/tasks.md +++ b/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. diff --git a/packages/cloud-platform/cloud/sdk/client.py b/packages/cloud-platform/cloud/sdk/client.py index e435ba9..f050fce 100644 --- a/packages/cloud-platform/cloud/sdk/client.py +++ b/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.raise_for_status() + resp = self._request("POST", "/tasks", json=payload) 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.raise_for_status() + resp = self._request("GET", f"/tasks/{task_id}") return resp.json() # ----------------------------------------------------------------- devices def list_devices(self) -> list[dict[str, Any]]: - resp = self._http.get(self._url("/devices")) - resp.raise_for_status() + resp = self._request("GET", "/devices") return resp.json() def list_hosts(self) -> list[dict[str, Any]]: - resp = self._http.get(self._url("/hosts")) - resp.raise_for_status() + resp = self._request("GET", "/hosts") return resp.json() # ----------------------------------------------------------------- plugins def list_plugins(self) -> list[dict[str, Any]]: - resp = self._http.get(self._url("/plugins")) - resp.raise_for_status() + resp = self._request("GET", "/plugins") 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.raise_for_status() + resp = self._request("POST", "/plugins", json=payload) 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)) diff --git a/tests/test_cloud_client.py b/tests/test_cloud_client.py index 2ae9d12..a2fd75e 100644 --- a/tests/test_cloud_client.py +++ b/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"), + )