Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 72286f188d |
@@ -0,0 +1,140 @@
|
||||
"""Host-side control-plane client (PRD 0070).
|
||||
|
||||
The launch path talks to the orchestrator over its HTTP control plane to
|
||||
register, re-policy, and tear down bottles — the counterpart to the
|
||||
gateway-side `PolicyResolver` (which only reads `/resolve`). Where
|
||||
`PolicyResolver` is fail-closed and lives in the untrusted data plane, this
|
||||
is the trusted control-plane caller: a non-success response is an error the
|
||||
launch path must surface, not silently swallow.
|
||||
|
||||
Stdlib-only, so the CLI can drive the orchestrator without any dependency.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import urllib.error
|
||||
import urllib.request
|
||||
from dataclasses import dataclass
|
||||
|
||||
DEFAULT_TIMEOUT_SECONDS = 5.0
|
||||
|
||||
|
||||
class OrchestratorClientError(RuntimeError):
|
||||
"""A control-plane call failed (unreachable, or an unexpected status)."""
|
||||
|
||||
|
||||
@dataclass(frozen=True)
|
||||
class RegisteredBottle:
|
||||
"""What `POST /bottles` returns: the minted bottle id and the per-bottle
|
||||
identity token the agent presents for app-layer attribution."""
|
||||
|
||||
bottle_id: str
|
||||
identity_token: str
|
||||
|
||||
|
||||
class OrchestratorClient:
|
||||
"""Trusted host-side client for the orchestrator control plane."""
|
||||
|
||||
def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT_SECONDS) -> None:
|
||||
self._base = base_url.rstrip("/")
|
||||
self._timeout = timeout
|
||||
|
||||
def _request(
|
||||
self, method: str, path: str, body: dict[str, object] | None = None,
|
||||
) -> tuple[int, dict[str, object]]:
|
||||
"""Send one request; return `(status, payload)`. Raises
|
||||
`OrchestratorClientError` only when the orchestrator can't be reached
|
||||
or returns malformed data — HTTP *status* codes are returned so
|
||||
callers can treat 404 as a meaningful "no such bottle"."""
|
||||
data = json.dumps(body).encode() if body is not None else None
|
||||
headers = {"Content-Type": "application/json"} if data is not None else {}
|
||||
req = urllib.request.Request(
|
||||
f"{self._base}{path}", data=data, method=method, headers=headers,
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
|
||||
raw = resp.read()
|
||||
payload = json.loads(raw) if raw else {}
|
||||
return resp.status, payload if isinstance(payload, dict) else {}
|
||||
except urllib.error.HTTPError as e:
|
||||
# A structured error response still carries a usable status.
|
||||
try:
|
||||
payload = json.loads(e.read() or b"{}")
|
||||
except (ValueError, OSError):
|
||||
payload = {}
|
||||
return e.code, payload if isinstance(payload, dict) else {}
|
||||
except (urllib.error.URLError, TimeoutError, OSError, ValueError) as e:
|
||||
raise OrchestratorClientError(f"{method} {path}: {e}") from e
|
||||
|
||||
def _ok(self, method: str, path: str, body: dict[str, object] | None = None) -> dict[str, object]:
|
||||
"""`_request` that requires a 2xx, raising otherwise."""
|
||||
status, payload = self._request(method, path, body)
|
||||
if not 200 <= status < 300:
|
||||
detail = payload.get("error", "")
|
||||
raise OrchestratorClientError(f"{method} {path}: HTTP {status} {detail}".rstrip())
|
||||
return payload
|
||||
|
||||
def health(self) -> bool:
|
||||
"""True iff the control plane answers `GET /health` with 200."""
|
||||
try:
|
||||
status, _ = self._request("GET", "/health")
|
||||
except OrchestratorClientError:
|
||||
return False
|
||||
return status == 200
|
||||
|
||||
def register_bottle(
|
||||
self,
|
||||
source_ip: str,
|
||||
*,
|
||||
image_ref: str = "",
|
||||
metadata: str = "",
|
||||
policy: str = "",
|
||||
) -> RegisteredBottle:
|
||||
"""Register a bottle and broker its launch (`POST /bottles`). Returns
|
||||
its minted id + identity token."""
|
||||
payload = self._ok("POST", "/bottles", {
|
||||
"source_ip": source_ip,
|
||||
"image_ref": image_ref,
|
||||
"metadata": metadata,
|
||||
"policy": policy,
|
||||
})
|
||||
bottle_id = payload.get("bottle_id")
|
||||
token = payload.get("identity_token")
|
||||
if not isinstance(bottle_id, str) or not isinstance(token, str):
|
||||
raise OrchestratorClientError("register: response missing bottle_id/identity_token")
|
||||
return RegisteredBottle(bottle_id=bottle_id, identity_token=token)
|
||||
|
||||
def teardown_bottle(self, bottle_id: str) -> bool:
|
||||
"""Tear a bottle down (`DELETE /bottles/<id>`). False if the
|
||||
orchestrator didn't know it (404) — idempotent for cleanup paths."""
|
||||
status, _ = self._request("DELETE", f"/bottles/{bottle_id}")
|
||||
if status == 404:
|
||||
return False
|
||||
if not 200 <= status < 300:
|
||||
raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}")
|
||||
return True
|
||||
|
||||
def set_policy(self, bottle_id: str, policy: str) -> bool:
|
||||
"""Live-reload a bottle's policy (`PUT /bottles/<id>/policy`). False on
|
||||
404 (unknown bottle)."""
|
||||
status, _ = self._request("PUT", f"/bottles/{bottle_id}/policy", {"policy": policy})
|
||||
if status == 404:
|
||||
return False
|
||||
if not 200 <= status < 300:
|
||||
raise OrchestratorClientError(f"set_policy {bottle_id}: HTTP {status}")
|
||||
return True
|
||||
|
||||
def list_bottles(self) -> list[dict[str, object]]:
|
||||
"""Every registered bottle's redacted record (`GET /bottles`)."""
|
||||
payload = self._ok("GET", "/bottles")
|
||||
bottles = payload.get("bottles")
|
||||
return bottles if isinstance(bottles, list) else []
|
||||
|
||||
|
||||
__all__ = [
|
||||
"OrchestratorClient",
|
||||
"OrchestratorClientError",
|
||||
"RegisteredBottle",
|
||||
"DEFAULT_TIMEOUT_SECONDS",
|
||||
]
|
||||
@@ -0,0 +1,106 @@
|
||||
"""Unit: host-side orchestrator control-plane client (PRD 0070). HTTP mocked."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import unittest
|
||||
import urllib.error
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from bot_bottle.orchestrator.client import (
|
||||
OrchestratorClient,
|
||||
OrchestratorClientError,
|
||||
RegisteredBottle,
|
||||
)
|
||||
|
||||
_URLOPEN = "bot_bottle.orchestrator.client.urllib.request.urlopen"
|
||||
|
||||
|
||||
def _resp(status: int, payload: object) -> MagicMock:
|
||||
m = MagicMock()
|
||||
inner = m.__enter__.return_value
|
||||
inner.status = status
|
||||
inner.read.return_value = json.dumps(payload).encode()
|
||||
return m
|
||||
|
||||
|
||||
def _http_error(code: int, payload: object = None) -> urllib.error.HTTPError:
|
||||
del payload # the client tolerates an empty error body; keep the signature
|
||||
return urllib.error.HTTPError("http://x", code, "err", {}, None) # type: ignore[arg-type]
|
||||
|
||||
|
||||
class TestRegister(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.c = OrchestratorClient("http://orch:8080")
|
||||
|
||||
def test_register_returns_id_and_token(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1", "identity_token": "tok"})):
|
||||
got = self.c.register_bottle("10.0.0.2", policy="routes: []\n", metadata="{}")
|
||||
self.assertEqual(RegisteredBottle("b1", "tok"), got)
|
||||
|
||||
def test_register_posts_source_ip_and_policy(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b", "identity_token": "t"})) as m:
|
||||
self.c.register_bottle("10.0.0.9", policy="P", metadata="M", image_ref="img")
|
||||
sent = json.loads(m.call_args.args[0].data)
|
||||
self.assertEqual("10.0.0.9", sent["source_ip"])
|
||||
self.assertEqual("P", sent["policy"])
|
||||
self.assertEqual("img", sent["image_ref"])
|
||||
|
||||
def test_register_missing_fields_raises(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(201, {"bottle_id": "b1"})):
|
||||
with self.assertRaises(OrchestratorClientError):
|
||||
self.c.register_bottle("10.0.0.2")
|
||||
|
||||
def test_register_non_2xx_raises(self) -> None:
|
||||
with patch(_URLOPEN, side_effect=_http_error(400, {"error": "bad"})):
|
||||
with self.assertRaises(OrchestratorClientError):
|
||||
self.c.register_bottle("")
|
||||
|
||||
|
||||
class TestTeardown(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.c = OrchestratorClient("http://orch:8080")
|
||||
|
||||
def test_teardown_true_on_success(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})):
|
||||
self.assertTrue(self.c.teardown_bottle("b1"))
|
||||
|
||||
def test_teardown_false_on_404(self) -> None:
|
||||
# Idempotent: an already-gone bottle is a clean no-op.
|
||||
with patch(_URLOPEN, side_effect=_http_error(404)):
|
||||
self.assertFalse(self.c.teardown_bottle("gone"))
|
||||
|
||||
def test_teardown_uses_delete(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(200, {"torn_down": True})) as m:
|
||||
self.c.teardown_bottle("b1")
|
||||
self.assertEqual("DELETE", m.call_args.args[0].get_method())
|
||||
|
||||
|
||||
class TestHealthAndPolicy(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
self.c = OrchestratorClient("http://orch:8080")
|
||||
|
||||
def test_health_true_on_200(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(200, {"status": "ok"})):
|
||||
self.assertTrue(self.c.health())
|
||||
|
||||
def test_health_false_when_unreachable(self) -> None:
|
||||
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
||||
self.assertFalse(self.c.health())
|
||||
|
||||
def test_set_policy_false_on_404(self) -> None:
|
||||
with patch(_URLOPEN, side_effect=_http_error(404)):
|
||||
self.assertFalse(self.c.set_policy("gone", "P"))
|
||||
|
||||
def test_set_policy_true_on_success(self) -> None:
|
||||
with patch(_URLOPEN, return_value=_resp(200, {"updated": True})):
|
||||
self.assertTrue(self.c.set_policy("b1", "P"))
|
||||
|
||||
def test_unreachable_raises(self) -> None:
|
||||
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
|
||||
with self.assertRaises(OrchestratorClientError):
|
||||
self.c.register_bottle("10.0.0.2")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user