diff --git a/bot_bottle/orchestrator/__init__.py b/bot_bottle/orchestrator/__init__.py index b04bab4..02c0693 100644 --- a/bot_bottle/orchestrator/__init__.py +++ b/bot_bottle/orchestrator/__init__.py @@ -7,23 +7,45 @@ backend-neutral "consolidation core" that needs no VM packaging: * `registry` — the SQLite runtime-state store + fail-closed attribution (source IP + per-bottle identity token). - * `control_plane` — the HTTP control-plane RPC (register / deregister / - list / attribute / health), with live reload. + * `broker` — the signed, structured launch-request contract + a + `LaunchBroker` (stub for the harness) that verifies + provenance before acting. + * `service` — the `Orchestrator`: owns the registry, brokers the + launch lifecycle (launch/teardown), attributes. + * `control_plane` — the HTTP control-plane RPC (launch / teardown / + list / attribute / health). -Launch/teardown, the launch broker, and the actual egress/git/supervise -data plane land in later slices once this core is proven (see the PRD's -sequencing: plain-process dev-harness -> docker orchestrator -> firecracker). +The actual backend-native launch (a real docker/firecracker broker) and +the egress/git/supervise data plane land in later slices once this core is +proven (see the PRD sequencing: plain-process dev-harness -> docker +orchestrator -> firecracker). """ from __future__ import annotations from .registry import BottleRecord, RegistryStore, new_identity_token +from .broker import ( + BrokerAuthError, + LaunchBroker, + LaunchRequest, + StubBroker, + sign_request, + verify_request, +) +from .service import Orchestrator from .control_plane import ControlPlaneServer, dispatch, make_server __all__ = [ "BottleRecord", "RegistryStore", "new_identity_token", + "BrokerAuthError", + "LaunchBroker", + "LaunchRequest", + "StubBroker", + "sign_request", + "verify_request", + "Orchestrator", "ControlPlaneServer", "dispatch", "make_server", diff --git a/bot_bottle/orchestrator/__main__.py b/bot_bottle/orchestrator/__main__.py index 7a2fd9a..cb1417a 100644 --- a/bot_bottle/orchestrator/__main__.py +++ b/bot_bottle/orchestrator/__main__.py @@ -12,11 +12,14 @@ container packaging. Wrapping this exact service in a backend-native unit from __future__ import annotations import argparse +import secrets from pathlib import Path from .. import log +from .broker import StubBroker from .control_plane import make_server from .registry import RegistryStore, default_db_path +from .service import Orchestrator def main(argv: list[str] | None = None) -> int: @@ -33,7 +36,13 @@ def main(argv: list[str] | None = None) -> int: registry = RegistryStore(args.db) registry.migrate() - server = make_server(registry, host=args.host, port=args.port) + # Dev-harness wiring: an ephemeral signing secret + a stub broker that + # records launches instead of starting anything. A real backend broker + # (docker, then firecracker) drops in here later. + secret = secrets.token_bytes(32) + orchestrator = Orchestrator(registry, StubBroker(secret), secret) + + server = make_server(orchestrator, host=args.host, port=args.port) bound_host, bound_port = server.server_address[0], server.server_address[1] log.info( "orchestrator control plane listening", diff --git a/bot_bottle/orchestrator/broker.py b/bot_bottle/orchestrator/broker.py new file mode 100644 index 0000000..bd25d4d --- /dev/null +++ b/bot_bottle/orchestrator/broker.py @@ -0,0 +1,176 @@ +"""Launch broker contract (PRD 0070). + +A VM/container can't spawn its own host-network siblings, so the +orchestrator brokers agent launches through a small privileged component. +The request it sends is: + + * **structured** — static flags + ids only (bottle id, pool slot, a + content-addressed image ref), never a free-form path or argv, so a + request can't be coerced into launching an arbitrary payload; and + * **signed** — wrapped as a compact JWS/JWT the broker verifies before + acting, so a *compromised co-located component* (an agent-facing + sidecar, say) can't forge a launch. This is the concrete form of the + "structured requests only" rule in the PRD security review. + +Signing is **HS256** over a secret shared by the orchestrator (signer) and +the broker (verifier) — appropriate here because both are trusted host +components provisioned together; the secret is exactly what an +agent-facing component does not have. (Stdlib only — the project takes no +runtime deps; a cross-host deployment could swap in an asymmetric alg.) +""" + +from __future__ import annotations + +import abc +import base64 +import hashlib +import hmac +import json +import secrets +import time +from dataclasses import dataclass + +_JWT_HEADER = {"alg": "HS256", "typ": "JWT"} +_ALLOWED_OPS = ("launch", "teardown") + + +class BrokerAuthError(Exception): + """A broker request failed provenance or schema verification — + bad/absent signature, malformed token, or a payload that doesn't match + the fixed launch-request shape. Fail-closed: the broker must not act.""" + + +@dataclass(frozen=True) +class LaunchRequest: + """The structured, un-coercible launch/teardown request. Ids + flags + only — `image_ref` is a content-addressed id, never a path/argv.""" + + op: str # one of _ALLOWED_OPS + bottle_id: str + source_ip: str = "" + image_ref: str = "" + slot: int | None = None + + +# --- minimal JWS/JWT (HS256), stdlib only ---------------------------------- + +def _b64url(data: bytes) -> str: + return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii") + + +def _b64url_decode(text: str) -> bytes: + return base64.urlsafe_b64decode(text + "=" * (-len(text) % 4)) + + +def _mac(signing_input: str, secret: bytes) -> str: + digest = hmac.new(secret, signing_input.encode("ascii"), hashlib.sha256).digest() + return _b64url(digest) + + +def sign_request(req: LaunchRequest, secret: bytes) -> str: + """Serialize `req` to signed-JWT form. Adds a per-signature `jti` + (replay id) and `iat` (issued-at).""" + claims: dict[str, object] = { + "op": req.op, + "bottle_id": req.bottle_id, + "source_ip": req.source_ip, + "image_ref": req.image_ref, + "slot": req.slot, + "jti": secrets.token_hex(8), + "iat": int(time.time()), + } + header = _b64url(json.dumps(_JWT_HEADER, sort_keys=True, separators=(",", ":")).encode()) + payload = _b64url(json.dumps(claims, sort_keys=True, separators=(",", ":")).encode()) + signing_input = f"{header}.{payload}" + return f"{signing_input}.{_mac(signing_input, secret)}" + + +def verify_request(token: str, secret: bytes) -> LaunchRequest: + """Verify the signature and the request shape; return the parsed + request. Raises `BrokerAuthError` on any failure (fail-closed).""" + parts = token.split(".") + if len(parts) != 3: + raise BrokerAuthError("malformed token") + header_b, payload_b, sig = parts + if not hmac.compare_digest(_mac(f"{header_b}.{payload_b}", secret), sig): + raise BrokerAuthError("bad signature") + try: + header = json.loads(_b64url_decode(header_b)) + claims = json.loads(_b64url_decode(payload_b)) + except (ValueError, json.JSONDecodeError) as e: + raise BrokerAuthError("malformed token payload") from e + if not isinstance(header, dict) or header.get("alg") != "HS256": + raise BrokerAuthError("unexpected header/alg") + if not isinstance(claims, dict): + raise BrokerAuthError("claims are not an object") + + op = claims.get("op") + bottle_id = claims.get("bottle_id") + source_ip = claims.get("source_ip", "") + image_ref = claims.get("image_ref", "") + slot = claims.get("slot") + if ( + not isinstance(op, str) or op not in _ALLOWED_OPS + or not isinstance(bottle_id, str) or not bottle_id + or not isinstance(source_ip, str) or not isinstance(image_ref, str) + or not (slot is None or isinstance(slot, int)) + ): + raise BrokerAuthError("request does not match the launch-request schema") + return LaunchRequest( + op=op, bottle_id=bottle_id, source_ip=source_ip, image_ref=image_ref, slot=slot + ) + + +# --- the broker itself ------------------------------------------------------ + +class LaunchBroker(abc.ABC): + """Verifies a signed request came from the orchestrator, then performs + the backend-native launch/teardown. Subclasses implement `_launch` / + `_teardown`; verification is shared and fail-closed.""" + + def __init__(self, secret: bytes) -> None: + self._secret = secret + + def submit(self, token: str) -> LaunchRequest: + """Verify `token` and perform its op. Returns the verified request; + raises `BrokerAuthError` if provenance/shape fails.""" + req = verify_request(token, self._secret) + if req.op == "launch": + self._launch(req) + else: + self._teardown(req) + return req + + @abc.abstractmethod + def _launch(self, req: LaunchRequest) -> None: + ... + + @abc.abstractmethod + def _teardown(self, req: LaunchRequest) -> None: + ... + + +class StubBroker(LaunchBroker): + """Dev-harness broker: records verified requests without launching + anything. Exercises the full sign -> verify -> act contract in-process.""" + + def __init__(self, secret: bytes) -> None: + super().__init__(secret) + self.launched: list[LaunchRequest] = [] + self.torn_down: list[LaunchRequest] = [] + + def _launch(self, req: LaunchRequest) -> None: + self.launched.append(req) + + def _teardown(self, req: LaunchRequest) -> None: + self.torn_down.append(req) + + +__all__ = [ + "BrokerAuthError", + "LaunchRequest", + "LaunchBroker", + "StubBroker", + "sign_request", + "verify_request", +] diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index 772eae2..6c2ca2c 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/bot_bottle/orchestrator/control_plane.py @@ -2,23 +2,25 @@ The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over **HTTP** — the universal transport chosen in 0070 (works on every host; no -vsock / unix-socket portability caveats). Endpoints mutate the live -registry, so register/deregister are the *live-reload* path — no restart: +vsock / unix-socket portability caveats): GET /health -> 200 {"status": "ok"} GET /bottles -> 200 {"bottles": [ , ... ]} - POST /bottles -> 201 {"bottle_id", "identity_token"} - body: {"source_ip", ["bottle_id"], ["metadata"]} - DELETE /bottles/ -> 200 {"deregistered": true} | 404 + POST /bottles -> 201 {"bottle_id", "identity_token"} (launch) + body: {"source_ip", ["image_ref"], ["metadata"]} + DELETE /bottles/ -> 200 {"torn_down": true} | 404 (teardown) POST /attribute -> 200 {"bottle_id"} | 403 body: {"source_ip", "identity_token"} +`POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or +tear down) the bottle in the registry AND broker the backend-native launch +via the orchestrator. Register/deregister without a launch are internal to +`Orchestrator`, not exposed here. + Routing/handling is the pure function `dispatch()` so it is unit-testable without a socket; `Handler` / `ControlPlaneServer` / `make_server` are a -thin stdlib adapter around it. - -Note the listing redacts identity tokens — they are never returned except -once, to the caller that registers the bottle. +thin stdlib adapter around it. Listing redacts identity tokens — they are +returned only once, to the caller that launches the bottle. """ from __future__ import annotations @@ -30,7 +32,7 @@ import socketserver import typing from urllib.parse import urlsplit -from .registry import RegistryStore +from .service import Orchestrator # JSON body payload type (parsed request / rendered response). Json = dict[str, object] @@ -47,17 +49,17 @@ def _parse_json_object(body: bytes) -> Json: def dispatch( # pylint: disable=too-many-return-statements - registry: RegistryStore, method: str, path: str, body: bytes + orch: Orchestrator, method: str, path: str, body: bytes ) -> tuple[int, Json]: """Route one control-plane request to a (status, payload) pair. Pure — - no I/O beyond the registry — so it is fully testable without a socket.""" + no I/O beyond the orchestrator — so it is fully testable without a socket.""" route = urlsplit(path).path.rstrip("/") or "/" if method == "GET" and route == "/health": return 200, {"status": "ok"} if method == "GET" and route == "/bottles": - return 200, {"bottles": [r.redacted() for r in registry.all()]} + return 200, {"bottles": [r.redacted() for r in orch.registry.all()]} if method == "POST" and route == "/bottles": try: @@ -67,19 +69,19 @@ def dispatch( # pylint: disable=too-many-return-statements source_ip = data.get("source_ip") if not isinstance(source_ip, str) or not source_ip: return 400, {"error": "source_ip (string) is required"} - bottle_id = data.get("bottle_id") + image_ref = data.get("image_ref") metadata = data.get("metadata") - rec = registry.register( + rec = orch.launch_bottle( source_ip, - bottle_id=bottle_id if isinstance(bottle_id, str) else None, + image_ref=image_ref if isinstance(image_ref, str) else "", metadata=metadata if isinstance(metadata, str) else "", ) return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token} if method == "DELETE" and route.startswith("/bottles/"): bottle_id = route[len("/bottles/"):] - if registry.deregister(bottle_id): - return 200, {"deregistered": True} + if orch.teardown_bottle(bottle_id): + return 200, {"torn_down": True} return 404, {"error": "no such bottle"} if method == "POST" and route == "/attribute": @@ -91,7 +93,7 @@ def dispatch( # pylint: disable=too-many-return-statements token = data.get("identity_token") if not isinstance(source_ip, str) or not isinstance(token, str): return 400, {"error": "source_ip and identity_token (strings) required"} - rec = registry.attribute(source_ip, token) + rec = orch.attribute(source_ip, token) if rec is None: return 403, {"error": "unattributed"} return 200, {"bottle_id": rec.bottle_id} @@ -114,7 +116,7 @@ class Handler(http.server.BaseHTTPRequestHandler): assert isinstance(server, ControlPlaneServer) length = int(self.headers.get("Content-Length") or 0) body = self.rfile.read(length) if length > 0 else b"" - status, payload = dispatch(server.registry, method, self.path, body) + status, payload = dispatch(server.orchestrator, method, self.path, body) data = json.dumps(payload).encode() self.send_response(status) self.send_header("Content-Type", "application/json") @@ -133,22 +135,22 @@ class Handler(http.server.BaseHTTPRequestHandler): class ControlPlaneServer(socketserver.ThreadingMixIn, http.server.HTTPServer): - """Threading HTTP server that carries the registry for its handlers.""" + """Threading HTTP server that carries the orchestrator for its handlers.""" daemon_threads = True allow_reuse_address = True - def __init__(self, address: tuple[str, int], registry: RegistryStore) -> None: - self.registry = registry + def __init__(self, address: tuple[str, int], orchestrator: Orchestrator) -> None: + self.orchestrator = orchestrator super().__init__(address, Handler) def make_server( - registry: RegistryStore, host: str = "127.0.0.1", port: int = 0 + orchestrator: Orchestrator, host: str = "127.0.0.1", port: int = 0 ) -> ControlPlaneServer: """Build (but do not start) a control-plane server. `port=0` binds an ephemeral port — read `server.server_address` for the actual one.""" - return ControlPlaneServer((host, port), registry) + return ControlPlaneServer((host, port), orchestrator) __all__ = ["dispatch", "Handler", "ControlPlaneServer", "make_server", "Json"] diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py new file mode 100644 index 0000000..5b43e17 --- /dev/null +++ b/bot_bottle/orchestrator/service.py @@ -0,0 +1,76 @@ +"""The per-host orchestrator core (PRD 0070). + +`Orchestrator` is the single backend-neutral object the control plane talks +to: it owns the registry (runtime state) and brokers agent launches. It +never branches on backend — the `LaunchBroker` abstracts the backend-native +launch, so this same object drives docker / firecracker / apple once a real +broker is wired in. + +Launch lifecycle: + + * `launch_bottle` mints the bottle (registry: source IP + identity + token), sends a *signed, structured* launch request through the broker, + and returns the record. If the broker rejects/fails, the registry entry + is rolled back so a failed launch leaves no orphan. + * `teardown_bottle` sends a signed teardown request, then deregisters. +""" + +from __future__ import annotations + +from .broker import LaunchBroker, LaunchRequest, sign_request +from .registry import BottleRecord, RegistryStore + + +class Orchestrator: + """Owns the registry + brokers launches. Backend-neutral.""" + + def __init__( + self, registry: RegistryStore, broker: LaunchBroker, sign_secret: bytes + ) -> None: + self.registry = registry + self._broker = broker + self._secret = sign_secret + + def launch_bottle( + self, + source_ip: str, + *, + image_ref: str = "", + slot: int | None = None, + metadata: str = "", + ) -> BottleRecord: + """Register a bottle and broker its launch. Rolls the registry entry + back if the launch doesn't take, so a failure leaves no orphan.""" + rec = self.registry.register(source_ip, metadata=metadata) + req = LaunchRequest( + op="launch", + bottle_id=rec.bottle_id, + source_ip=source_ip, + image_ref=image_ref, + slot=slot, + ) + launched = False + try: + self._broker.submit(sign_request(req, self._secret)) + launched = True + finally: + if not launched: + self.registry.deregister(rec.bottle_id) + return rec + + def teardown_bottle(self, bottle_id: str) -> bool: + """Broker teardown then deregister. False if the bottle is unknown.""" + rec = self.registry.get(bottle_id) + if rec is None: + return False + req = LaunchRequest(op="teardown", bottle_id=bottle_id, source_ip=rec.source_ip) + self._broker.submit(sign_request(req, self._secret)) + self.registry.deregister(bottle_id) + return True + + def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None: + """Fail-closed attribution (delegates to the registry).""" + return self.registry.attribute(source_ip, identity_token) + + +__all__ = ["Orchestrator"] diff --git a/tests/unit/test_orchestrator_broker.py b/tests/unit/test_orchestrator_broker.py new file mode 100644 index 0000000..e7588b0 --- /dev/null +++ b/tests/unit/test_orchestrator_broker.py @@ -0,0 +1,86 @@ +"""Unit tests for the orchestrator launch broker (PRD 0070).""" + +from __future__ import annotations + +import secrets +import unittest + +from bot_bottle.orchestrator.broker import ( + BrokerAuthError, + LaunchRequest, + StubBroker, + sign_request, + verify_request, +) + + +class TestSignVerify(unittest.TestCase): + def setUp(self) -> None: + self.secret = secrets.token_bytes(16) + + def test_launch_round_trip(self) -> None: + req = LaunchRequest( + op="launch", bottle_id="b1", source_ip="10.243.0.1", + image_ref="sha256:abc", slot=3, + ) + self.assertEqual(req, verify_request(sign_request(req, self.secret), self.secret)) + + def test_teardown_round_trip(self) -> None: + req = LaunchRequest(op="teardown", bottle_id="b1") + got = verify_request(sign_request(req, self.secret), self.secret) + self.assertEqual(("teardown", "b1"), (got.op, got.bottle_id)) + + def test_wrong_secret_rejected(self) -> None: + token = sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret) + with self.assertRaises(BrokerAuthError): + verify_request(token, secrets.token_bytes(16)) + + def test_tampered_payload_rejected(self) -> None: + token = sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret) + header, payload, sig = token.split(".") + flipped = payload[:-1] + ("A" if payload[-1] != "A" else "B") + with self.assertRaises(BrokerAuthError): + verify_request(f"{header}.{flipped}.{sig}", self.secret) + + def test_malformed_tokens_rejected(self) -> None: + for bad in ("", "a.b", "a.b.c.d", "not-a-token"): + with self.assertRaises(BrokerAuthError): + verify_request(bad, self.secret) + + def test_unknown_op_rejected_despite_valid_signature(self) -> None: + # A correctly-signed token whose op isn't in the allow-list must + # still be refused — the schema guard is independent of provenance. + token = sign_request(LaunchRequest(op="rm-rf", bottle_id="b1"), self.secret) + with self.assertRaises(BrokerAuthError): + verify_request(token, self.secret) + + +class TestStubBroker(unittest.TestCase): + def setUp(self) -> None: + self.secret = secrets.token_bytes(16) + self.broker = StubBroker(self.secret) + + def test_submit_launch_records_verified_request(self) -> None: + req = LaunchRequest(op="launch", bottle_id="b1", source_ip="10.243.0.1") + got = self.broker.submit(sign_request(req, self.secret)) + self.assertEqual(req, got) + self.assertEqual(["b1"], [r.bottle_id for r in self.broker.launched]) + self.assertEqual([], self.broker.torn_down) + + def test_submit_teardown_records(self) -> None: + self.broker.submit( + sign_request(LaunchRequest(op="teardown", bottle_id="b1"), self.secret) + ) + self.assertEqual(["b1"], [r.bottle_id for r in self.broker.torn_down]) + + def test_submit_forged_token_is_fail_closed(self) -> None: + forged = sign_request( + LaunchRequest(op="launch", bottle_id="b1"), secrets.token_bytes(16) + ) + with self.assertRaises(BrokerAuthError): + self.broker.submit(forged) + self.assertEqual([], self.broker.launched) # nothing acted on + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index 8310df1..3d1a29c 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -7,53 +7,62 @@ server tests), plus one real-socket round-trip to prove the handler wiring. from __future__ import annotations import json +import secrets import tempfile import threading import unittest import urllib.request from pathlib import Path +from bot_bottle.orchestrator.broker import StubBroker from bot_bottle.orchestrator.control_plane import dispatch, make_server from bot_bottle.orchestrator.registry import RegistryStore +from bot_bottle.orchestrator.service import Orchestrator def _body(obj: object) -> bytes: return json.dumps(obj).encode() +def _orchestrator(db_path: Path) -> Orchestrator: + store = RegistryStore(db_path) + store.migrate() + secret = secrets.token_bytes(16) + return Orchestrator(store, StubBroker(secret), secret) + + class TestDispatch(unittest.TestCase): def setUp(self) -> None: self._tmp = tempfile.TemporaryDirectory() - self.store = RegistryStore(Path(self._tmp.name) / "r.db") - self.store.migrate() + self.orch = _orchestrator(Path(self._tmp.name) / "r.db") def tearDown(self) -> None: self._tmp.cleanup() def test_health(self) -> None: - status, payload = dispatch(self.store, "GET", "/health", b"") + status, payload = dispatch(self.orch, "GET", "/health", b"") self.assertEqual(200, status) self.assertEqual("ok", payload["status"]) def test_register_returns_id_and_token(self) -> None: status, payload = dispatch( - self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}) + self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}) ) self.assertEqual(201, status) self.assertTrue(payload["bottle_id"]) self.assertTrue(payload["identity_token"]) def test_register_requires_source_ip(self) -> None: - status, _ = dispatch(self.store, "POST", "/bottles", _body({})) + status, _ = dispatch(self.orch, "POST", "/bottles", _body({})) self.assertEqual(400, status) def test_register_rejects_bad_json(self) -> None: - status, _ = dispatch(self.store, "POST", "/bottles", b"{not json") + status, _ = dispatch(self.orch, "POST", "/bottles", b"{not json") self.assertEqual(400, status) def test_list_redacts_identity_token(self) -> None: - dispatch(self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})) - status, payload = dispatch(self.store, "GET", "/bottles", b"") + dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})) + status, payload = dispatch(self.orch, "GET", "/bottles", b"") self.assertEqual(200, status) bottles = payload["bottles"] assert isinstance(bottles, list) @@ -65,48 +74,48 @@ class TestDispatch(unittest.TestCase): def test_attribute_ok_and_forbidden(self) -> None: _, reg = dispatch( - self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.5"}) + self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.5"}) ) token = reg["identity_token"] ok_status, ok = dispatch( - self.store, "POST", "/attribute", + self.orch, "POST", "/attribute", _body({"source_ip": "10.243.0.5", "identity_token": token}), ) self.assertEqual(200, ok_status) self.assertEqual(reg["bottle_id"], ok["bottle_id"]) bad_status, _ = dispatch( - self.store, "POST", "/attribute", + self.orch, "POST", "/attribute", _body({"source_ip": "10.243.0.5", "identity_token": "nope"}), ) self.assertEqual(403, bad_status) def test_attribute_requires_both_fields(self) -> None: status, _ = dispatch( - self.store, "POST", "/attribute", _body({"source_ip": "10.243.0.5"}) + self.orch, "POST", "/attribute", _body({"source_ip": "10.243.0.5"}) ) self.assertEqual(400, status) def test_delete(self) -> None: _, reg = dispatch( - self.store, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}) + self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"}) ) bid = reg["bottle_id"] - status, payload = dispatch(self.store, "DELETE", f"/bottles/{bid}", b"") + status, payload = dispatch(self.orch, "DELETE", f"/bottles/{bid}", b"") self.assertEqual(200, status) - self.assertEqual(True, payload["deregistered"]) - self.assertEqual([], self.store.all()) + self.assertEqual(True, payload["torn_down"]) + self.assertEqual([], self.orch.registry.all()) def test_delete_missing_404(self) -> None: - status, _ = dispatch(self.store, "DELETE", "/bottles/ghost", b"") + status, _ = dispatch(self.orch, "DELETE", "/bottles/ghost", b"") self.assertEqual(404, status) def test_unknown_route_404(self) -> None: - status, _ = dispatch(self.store, "GET", "/nope", b"") + status, _ = dispatch(self.orch, "GET", "/nope", b"") self.assertEqual(404, status) def test_trailing_slash_normalized(self) -> None: - status, _ = dispatch(self.store, "GET", "/health/", b"") + status, _ = dispatch(self.orch, "GET", "/health/", b"") self.assertEqual(200, status) @@ -114,9 +123,8 @@ class TestServerRoundTrip(unittest.TestCase): def test_http_register_health_attribute(self) -> None: tmp = tempfile.TemporaryDirectory() self.addCleanup(tmp.cleanup) - store = RegistryStore(Path(tmp.name) / "r.db") - store.migrate() - server = make_server(store, "127.0.0.1", 0) + orch = _orchestrator(Path(tmp.name) / "r.db") + server = make_server(orch, "127.0.0.1", 0) self.addCleanup(server.server_close) thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py new file mode 100644 index 0000000..7d3ff22 --- /dev/null +++ b/tests/unit/test_orchestrator_service.py @@ -0,0 +1,73 @@ +"""Unit tests for the Orchestrator launch lifecycle (PRD 0070).""" + +from __future__ import annotations + +import secrets +import tempfile +import unittest +from pathlib import Path + +from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker +from bot_bottle.orchestrator.registry import RegistryStore +from bot_bottle.orchestrator.service import Orchestrator + + +class _FailingBroker(LaunchBroker): + """Verifies the token like any broker, then fails the launch — to + exercise the orchestrator's registry rollback.""" + + def _launch(self, req: LaunchRequest) -> None: + raise RuntimeError("launch failed") + + def _teardown(self, req: LaunchRequest) -> None: + pass + + +class TestOrchestrator(unittest.TestCase): + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + self.secret = secrets.token_bytes(16) + self.store = RegistryStore(Path(self._tmp.name) / "r.db") + self.store.migrate() + self.broker = StubBroker(self.secret) + self.orch = Orchestrator(self.store, self.broker, self.secret) + + def tearDown(self) -> None: + self._tmp.cleanup() + + def test_launch_registers_and_brokers_signed_request(self) -> None: + rec = self.orch.launch_bottle("10.243.0.1", image_ref="sha256:abc", slot=2) + self.assertIsNotNone(self.store.get(rec.bottle_id)) + self.assertEqual(1, len(self.broker.launched)) + req = self.broker.launched[0] + self.assertEqual("launch", req.op) + self.assertEqual(rec.bottle_id, req.bottle_id) + self.assertEqual("10.243.0.1", req.source_ip) + self.assertEqual("sha256:abc", req.image_ref) + self.assertEqual(2, req.slot) + + def test_launch_then_attribute(self) -> None: + rec = self.orch.launch_bottle("10.243.0.3") + got = self.orch.attribute("10.243.0.3", rec.identity_token) + assert got is not None + self.assertEqual(rec.bottle_id, got.bottle_id) + self.assertIsNone(self.orch.attribute("10.243.0.3", "wrong-token")) + + def test_teardown_brokers_and_deregisters(self) -> None: + rec = self.orch.launch_bottle("10.243.0.1") + self.assertTrue(self.orch.teardown_bottle(rec.bottle_id)) + self.assertIsNone(self.store.get(rec.bottle_id)) + self.assertEqual(["teardown"], [r.op for r in self.broker.torn_down]) + + def test_teardown_unknown_is_false(self) -> None: + self.assertFalse(self.orch.teardown_bottle("ghost")) + + def test_launch_rolls_back_registry_on_broker_failure(self) -> None: + orch = Orchestrator(self.store, _FailingBroker(self.secret), self.secret) + with self.assertRaises(RuntimeError): + orch.launch_bottle("10.243.0.9") + self.assertEqual([], self.store.all()) # no orphan + + +if __name__ == "__main__": + unittest.main()