diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index b25adaf..df8d8f9 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/bot_bottle/orchestrator/control_plane.py @@ -4,14 +4,18 @@ 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): - GET /health -> 200 {"status": "ok"} - GET /sidecar -> 200 {"configured", ["name", "running"]} - GET /bottles -> 200 {"bottles": [ , ... ]} - 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"} + GET /health -> 200 {"status": "ok"} + GET /sidecar -> 200 {"configured", ["name","running"]} + GET /bottles -> 200 {"bottles": [ , ...]} + POST /bottles -> 201 {"bottle_id","identity_token"} (launch) + body: {"source_ip", ["image_ref"], + ["metadata"], ["policy"]} + PUT /bottles//policy -> 200 {"updated": true} | 404 (live reload) + body: {"policy"} + DELETE /bottles/ -> 200 {"torn_down": true} | 404 (teardown) + POST /attribute -> 200 {"bottle_id"} | 403 + POST /resolve -> 200 {"bottle_id","policy"} | 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 @@ -49,7 +53,7 @@ def _parse_json_object(body: bytes) -> Json: return obj -def dispatch( # pylint: disable=too-many-return-statements +def dispatch( # pylint: disable=too-many-return-statements,too-many-branches orch: Orchestrator, method: str, path: str, body: bytes ) -> tuple[int, Json]: """Route one control-plane request to a (status, payload) pair. Pure — @@ -75,20 +79,35 @@ def dispatch( # pylint: disable=too-many-return-statements return 400, {"error": "source_ip (string) is required"} image_ref = data.get("image_ref") metadata = data.get("metadata") + policy = data.get("policy") rec = orch.launch_bottle( source_ip, image_ref=image_ref if isinstance(image_ref, str) else "", metadata=metadata if isinstance(metadata, str) else "", + policy=policy if isinstance(policy, str) else "", ) return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token} + if method == "PUT" and route.startswith("/bottles/") and route.endswith("/policy"): + bottle_id = route[len("/bottles/"):-len("/policy")] + try: + data = _parse_json_object(body) + except ValueError as e: + return 400, {"error": f"invalid JSON: {e}"} + policy = data.get("policy") + if not isinstance(policy, str): + return 400, {"error": "policy (string) is required"} + if orch.set_policy(bottle_id, policy): + return 200, {"updated": True} + return 404, {"error": "no such bottle"} + if method == "DELETE" and route.startswith("/bottles/"): bottle_id = route[len("/bottles/"):] if orch.teardown_bottle(bottle_id): return 200, {"torn_down": True} return 404, {"error": "no such bottle"} - if method == "POST" and route == "/attribute": + if method == "POST" and route in ("/attribute", "/resolve"): try: data = _parse_json_object(body) except ValueError as e: @@ -100,6 +119,11 @@ def dispatch( # pylint: disable=too-many-return-statements rec = orch.attribute(source_ip, token) if rec is None: return 403, {"error": "unattributed"} + # /attribute is the lightweight identity check; /resolve additionally + # returns the bottle's policy — the source-IP-keyed lookup the + # multi-tenant sidecar makes per request. + if route == "/resolve": + return 200, {"bottle_id": rec.bottle_id, "policy": rec.policy} return 200, {"bottle_id": rec.bottle_id} return 404, {"error": "not found"} @@ -134,6 +158,9 @@ class Handler(http.server.BaseHTTPRequestHandler): def do_POST(self) -> None: self._serve("POST") + def do_PUT(self) -> None: + self._serve("PUT") + def do_DELETE(self) -> None: self._serve("DELETE") diff --git a/bot_bottle/orchestrator/registry.py b/bot_bottle/orchestrator/registry.py index dd962a5..01eb9b4 100644 --- a/bot_bottle/orchestrator/registry.py +++ b/bot_bottle/orchestrator/registry.py @@ -64,9 +64,13 @@ class BottleRecord: identity_token: str state: str = "active" created_at: float = 0.0 - # Opaque JSON blob for forward-compat (policy refs, pool slot, ...) — - # the registry doesn't interpret it, so new fields need no migration. + # Opaque JSON blob for forward-compat (pool slot, ...) — the registry + # doesn't interpret it, so new fields need no migration. metadata: str = "" + # The bottle's sidecar policy (opaque JSON — egress allowlist / routes / + # git config). The registry stores and serves it verbatim, keyed by + # source IP; the multi-tenant sidecar interprets it. + policy: str = "" def redacted(self) -> dict[str, object]: """Public view for listing — never exposes the identity token.""" @@ -76,6 +80,7 @@ class BottleRecord: "state": self.state, "created_at": self.created_at, "metadata": self.metadata, + "policy": self.policy, } @@ -97,6 +102,10 @@ _MIGRATIONS = TableMigrations( # the "one active bottle per source IP" check cheap. "CREATE INDEX IF NOT EXISTS idx_orchestrator_bottles_source_ip " "ON orchestrator_bottles (source_ip)", + # v3 — per-bottle policy (opaque JSON the sidecar interprets): the + # egress allowlist / routes / git config selected by source IP. The + # multi-tenant sidecar resolves it per request via `attribute`. + "ALTER TABLE orchestrator_bottles ADD COLUMN policy TEXT NOT NULL DEFAULT ''", ], ) @@ -110,6 +119,7 @@ def _row_to_record(row: sqlite3.Row) -> BottleRecord: state=row["state"], created_at=row["created_at"], metadata=row["metadata"], + policy=row["policy"], ) @@ -136,6 +146,7 @@ class RegistryStore(DbStore): bottle_id: str | None = None, identity_token: str | None = None, metadata: str = "", + policy: str = "", ) -> BottleRecord: """Register (or replace) a bottle. Mints a bottle_id / identity token when not supplied. Live — this is the control-plane reload path.""" @@ -146,12 +157,13 @@ class RegistryStore(DbStore): state="active", created_at=time.time(), metadata=metadata, + policy=policy, ) with self._connect() as conn: conn.execute( "INSERT OR REPLACE INTO orchestrator_bottles " - "(bottle_id, source_ip, identity_token, state, created_at, metadata) " - "VALUES (?, ?, ?, ?, ?, ?)", + "(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) " + "VALUES (?, ?, ?, ?, ?, ?, ?)", ( rec.bottle_id, rec.source_ip, @@ -159,11 +171,23 @@ class RegistryStore(DbStore): rec.state, rec.created_at, rec.metadata, + rec.policy, ), ) self._chmod() return rec + def set_policy(self, bottle_id: str, policy: str) -> bool: + """Update a bottle's policy in place (live reload). Returns True if + the bottle exists.""" + with self._connect() as conn: + cur = conn.execute( + "UPDATE orchestrator_bottles SET policy = ? WHERE bottle_id = ?", + (policy, bottle_id), + ) + self._chmod() + return cur.rowcount > 0 + def deregister(self, bottle_id: str) -> bool: """Remove a bottle. Returns True if a row was deleted.""" with self._connect() as conn: diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index 3b2aca7..edc449f 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -46,10 +46,12 @@ class Orchestrator: image_ref: str = "", slot: int | None = None, metadata: str = "", + policy: 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) + """Register a bottle (with its sidecar policy) 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, policy=policy) req = LaunchRequest( op="launch", bottle_id=rec.bottle_id, @@ -77,9 +79,16 @@ class Orchestrator: return True def attribute(self, source_ip: str, identity_token: str) -> BottleRecord | None: - """Fail-closed attribution (delegates to the registry).""" + """Fail-closed attribution (delegates to the registry). The returned + record carries the bottle's `policy` — this is the source-IP-keyed + resolution the multi-tenant sidecar makes per request.""" return self.registry.attribute(source_ip, identity_token) + def set_policy(self, bottle_id: str, policy: str) -> bool: + """Update a bottle's sidecar policy in place (live reload). False if + the bottle is unknown.""" + return self.registry.set_policy(bottle_id, policy) + # --- consolidated sidecar ---------------------------------------------- def ensure_sidecar(self) -> None: diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index 9a8b615..6b0bed4 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -123,6 +123,54 @@ class TestDispatch(unittest.TestCase): self.assertEqual(200, status) self.assertEqual(False, payload["configured"]) + def test_launch_stores_policy_and_resolve_returns_it(self) -> None: + _, reg = dispatch( + self.orch, "POST", "/bottles", + _body({"source_ip": "10.243.0.5", "policy": '{"a":1}'}), + ) + status, payload = dispatch( + self.orch, "POST", "/resolve", + _body({"source_ip": "10.243.0.5", "identity_token": reg["identity_token"]}), + ) + self.assertEqual(200, status) + self.assertEqual(reg["bottle_id"], payload["bottle_id"]) + self.assertEqual('{"a":1}', payload["policy"]) + + def test_resolve_forbidden_on_bad_token(self) -> None: + dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.5"})) + status, _ = dispatch( + self.orch, "POST", "/resolve", + _body({"source_ip": "10.243.0.5", "identity_token": "nope"}), + ) + self.assertEqual(403, status) + + def test_put_policy_updates_live(self) -> None: + _, reg = dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})) + bid = reg["bottle_id"] + status, payload = dispatch( + self.orch, "PUT", f"/bottles/{bid}/policy", _body({"policy": '{"v":9}'}) + ) + self.assertEqual(200, status) + self.assertEqual(True, payload["updated"]) + _, resolved = dispatch( + self.orch, "POST", "/resolve", + _body({"source_ip": "10.243.0.1", "identity_token": reg["identity_token"]}), + ) + self.assertEqual('{"v":9}', resolved["policy"]) + + def test_put_policy_missing_bottle_404(self) -> None: + status, _ = dispatch( + self.orch, "PUT", "/bottles/ghost/policy", _body({"policy": "{}"}) + ) + self.assertEqual(404, status) + + def test_put_policy_requires_string(self) -> None: + _, reg = dispatch(self.orch, "POST", "/bottles", _body({"source_ip": "10.243.0.1"})) + status, _ = dispatch( + self.orch, "PUT", f"/bottles/{reg['bottle_id']}/policy", _body({}) + ) + self.assertEqual(400, status) + class TestServerRoundTrip(unittest.TestCase): def test_http_register_health_attribute(self) -> None: diff --git a/tests/unit/test_orchestrator_registry.py b/tests/unit/test_orchestrator_registry.py index 6962cb1..f7e8571 100644 --- a/tests/unit/test_orchestrator_registry.py +++ b/tests/unit/test_orchestrator_registry.py @@ -97,6 +97,30 @@ class TestRegistryStore(unittest.TestCase): self.assertNotIn("identity_token", rec.redacted()) self.assertEqual("10.243.0.1", rec.redacted()["source_ip"]) + def test_register_with_policy_resolves_via_attribution(self) -> None: + rec = self.store.register("10.243.0.1", policy='{"allow":["x"]}') + self.assertEqual('{"allow":["x"]}', rec.policy) + got = self.store.attribute("10.243.0.1", rec.identity_token) + assert got is not None + self.assertEqual('{"allow":["x"]}', got.policy) + + def test_set_policy_updates_live(self) -> None: + rec = self.store.register("10.243.0.1") + self.assertEqual("", rec.policy) + self.assertTrue(self.store.set_policy(rec.bottle_id, '{"v":2}')) + got = self.store.get(rec.bottle_id) + assert got is not None + self.assertEqual('{"v":2}', got.policy) + + def test_set_policy_missing_is_false(self) -> None: + self.assertFalse(self.store.set_policy("ghost", "{}")) + + def test_policy_defaults_empty_and_persists(self) -> None: + rec = self.store.register("10.243.0.1") + got = RegistryStore(self.db).get(rec.bottle_id) + assert got is not None + self.assertEqual("", got.policy) + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index d02927c..13f034b 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -86,6 +86,22 @@ class TestOrchestrator(unittest.TestCase): def test_teardown_unknown_is_false(self) -> None: self.assertFalse(self.orch.teardown_bottle("ghost")) + def test_launch_with_policy_is_resolvable(self) -> None: + rec = self.orch.launch_bottle("10.243.0.1", policy='{"routes":[]}') + got = self.orch.attribute("10.243.0.1", rec.identity_token) + assert got is not None + self.assertEqual('{"routes":[]}', got.policy) + + def test_set_policy_live_reload(self) -> None: + rec = self.orch.launch_bottle("10.243.0.3") + self.assertTrue(self.orch.set_policy(rec.bottle_id, '{"x":1}')) + got = self.orch.attribute("10.243.0.3", rec.identity_token) + assert got is not None + self.assertEqual('{"x":1}', got.policy) + + def test_set_policy_unknown_is_false(self) -> None: + self.assertFalse(self.orch.set_policy("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):