feat(cloud-auth): bind host principals

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

@@ -29,7 +29,7 @@
 
 - [x] 4.1 Extend authenticated principals with scopes and implement constant-time configured bearer-token verification without logging credentials.
 - [x] 4.2 Add public scopes for task submission/read, pool read, plugin read, and plugin administration and enforce them on every `/v1` route.
-- [ ] 4.3 Add host principals bound to one `host_id` and reject cross-host heartbeat, claim, renewal, or result operations.
+- [x] 4.3 Add host principals bound to one `host_id` and reject cross-host heartbeat, claim, renewal, or result operations.
 - [ ] 4.4 Make missing production credentials a startup/readiness failure and permit anonymous mode only through the explicit non-production override.
 - [ ] 4.5 Add authentication tests covering invalid tokens, missing scopes, host impersonation, plugin administration, and secret redaction.
 

packages/cloud-platform/cloud/auth.py

@@ -17,10 +17,31 @@ PLUGINS_ADMIN_SCOPE = "plugins:admin"
 class Principal:
     id: str = "anonymous"
     scopes: frozenset[str] = field(default_factory=frozenset)
+    host_id: str | None = None
 
     def has_scope(self, scope: str) -> bool:
         return "*" in self.scopes or scope in self.scopes
 
+    def require_host(self, requested_host_id: str) -> None:
+        if self.host_id is None:
+            raise HostPrincipalRequiredError("a host-bound principal is required")
+        if self.host_id != requested_host_id:
+            raise HostIdentityMismatchError(
+                "authenticated host cannot act for the requested host"
+            )
+
+
+class HostAuthorizationError(PermissionError):
+    pass
+
+
+class HostPrincipalRequiredError(HostAuthorizationError):
+    pass
+
+
+class HostIdentityMismatchError(HostAuthorizationError):
+    pass
+
 
 @runtime_checkable
 class AuthProvider(Protocol):
@@ -39,12 +60,15 @@ class BearerCredential:
     principal_id: str
     token: str = field(repr=False)
     scopes: frozenset[str] = field(default_factory=frozenset)
+    host_id: str | None = None
 
     def __post_init__(self) -> None:
         if not self.principal_id.strip():
             raise ValueError("bearer credential principal_id must not be empty")
         if not self.token:
             raise ValueError("bearer credential token must not be empty")
+        if self.host_id is not None and not self.host_id.strip():
+            raise ValueError("bearer credential host_id must not be empty")
 
 
 @dataclass(frozen=True)
@@ -62,6 +86,7 @@ class ConfiguredBearerAuthProvider:
                 principal=Principal(
                     id=credential.principal_id,
                     scopes=frozenset(credential.scopes),
+                    host_id=credential.host_id,
                 ),
                 token_digest=_token_digest(credential.token),
             )

tests/test_cloud_auth.py

@@ -3,9 +3,12 @@ from __future__ import annotations
 from dataclasses import dataclass
 
 import cloud.auth as auth_module
+import pytest
 from cloud.auth import (
     BearerCredential,
     ConfiguredBearerAuthProvider,
+    HostIdentityMismatchError,
+    HostPrincipalRequiredError,
     NullAuthProvider,
 )
 
@@ -91,3 +94,56 @@ def test_null_auth_provider_is_explicitly_unrestricted() -> None:
     assert principal is not None
     assert principal.id == "anonymous"
     assert principal.scopes == frozenset({"*"})
+
+
+def test_host_credential_authenticates_as_one_bound_host() -> None:
+    provider = ConfiguredBearerAuthProvider(
+        [
+            BearerCredential(
+                principal_id="host-agent-a",
+                token="host-secret",
+                scopes=frozenset({"host:agent"}),
+                host_id="host-a",
+            )
+        ]
+    )
+
+    principal = provider.authenticate(
+        _Request(headers={"authorization": "Bearer host-secret"})
+    )
+
+    assert principal is not None
+    assert principal.host_id == "host-a"
+    principal.require_host("host-a")
+
+
+def test_host_principal_rejects_cross_host_operation() -> None:
+    provider = ConfiguredBearerAuthProvider(
+        [
+            BearerCredential(
+                principal_id="host-agent-a",
+                token="host-secret",
+                host_id="host-a",
+            )
+        ]
+    )
+    principal = provider.authenticate(
+        _Request(headers={"authorization": "Bearer host-secret"})
+    )
+
+    assert principal is not None
+    with pytest.raises(HostIdentityMismatchError):
+        principal.require_host("host-b")
+
+
+def test_public_principal_cannot_act_as_host() -> None:
+    provider = ConfiguredBearerAuthProvider(
+        [BearerCredential(principal_id="integrator", token="public-secret")]
+    )
+    principal = provider.authenticate(
+        _Request(headers={"authorization": "Bearer public-secret"})
+    )
+
+    assert principal is not None
+    with pytest.raises(HostPrincipalRequiredError):
+        principal.require_host("host-a")