diff --git a/bot_bottle/orchestrator/control_plane.py b/bot_bottle/orchestrator/control_plane.py index c590915..064528f 100644 --- a/bot_bottle/orchestrator/control_plane.py +++ b/bot_bottle/orchestrator/control_plane.py @@ -16,6 +16,10 @@ vsock / unix-socket portability caveats): POST /attribute -> 200 {"bottle_id"} | 403 POST /resolve -> 200 {"bottle_id","policy"} | 403 body: {"source_ip","identity_token"} + GET /supervise/proposals -> 200 {"proposals": [ , ...]} + POST /supervise/respond -> 200 {"responded": true} | 409 (operator) + body: {"proposal_id","bottle_slug", + "decision", ["notes"],["final_file"]} `POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or tear down) the bottle in the registry AND broker the backend-native launch @@ -127,6 +131,39 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches return 403, {"error": "unattributed"} return 200, {"bottle_id": rec.bottle_id} + if method == "GET" and route == "/supervise/proposals": + # Operator TUI: pending supervise proposals across all bottles. + return 200, {"proposals": orch.supervise_pending()} + + if method == "POST" and route == "/supervise/respond": + # Operator decision: apply (approve/modify rewrites egress policy), + # write the queued response, audit — all server-side on the one DB. + try: + data = _parse_json_object(body) + except ValueError as e: + return 400, {"error": f"invalid JSON: {e}"} + proposal_id = data.get("proposal_id") + bottle_slug = data.get("bottle_slug") + decision = data.get("decision") + if not (isinstance(proposal_id, str) and proposal_id): + return 400, {"error": "proposal_id (string) is required"} + if not (isinstance(bottle_slug, str) and bottle_slug): + return 400, {"error": "bottle_slug (string) is required"} + if not (isinstance(decision, str) and decision): + return 400, {"error": "decision (string) is required"} + notes = data.get("notes") + final_file = data.get("final_file") + ok, err = orch.supervise_respond( + proposal_id, + bottle_slug=bottle_slug, + decision=decision, + notes=notes if isinstance(notes, str) else "", + final_file=final_file if isinstance(final_file, str) else None, + ) + if ok: + return 200, {"responded": True} + return 409, {"error": err} + if method == "POST" and route == "/resolve": # The per-request lookup the multi-tenant gateway makes: returns the # bottle's policy. Requires a matching (source_ip, identity_token) diff --git a/bot_bottle/orchestrator/service.py b/bot_bottle/orchestrator/service.py index c44e5be..5c34815 100644 --- a/bot_bottle/orchestrator/service.py +++ b/bot_bottle/orchestrator/service.py @@ -17,9 +17,37 @@ Launch lifecycle: from __future__ import annotations +import json +from datetime import datetime, timezone + from .broker import LaunchBroker, LaunchRequest, sign_request from .registry import BottleRecord, RegistryStore from .gateway import Gateway +from ..supervise import ( + AuditEntry, + COMPONENT_FOR_TOOL, + Response, + STATUS_APPROVED, + STATUS_MODIFIED, + STATUS_REJECTED, + TOOL_EGRESS_ALLOW, + TOOL_EGRESS_BLOCK, + list_all_pending_proposals, + read_proposal, + render_diff, + write_audit_entry, + write_response, +) + + +# Operator decision → Response.status. The apply half (egress tools) runs +# for approve/modify only. +_RESPOND_STATUS = { + "approve": STATUS_APPROVED, + "modify": STATUS_MODIFIED, + "reject": STATUS_REJECTED, +} +_APPLY_TOOLS = (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK) class Orchestrator: @@ -117,6 +145,83 @@ class Orchestrator: the bottle is unknown.""" return self.registry.set_policy(bottle_id, policy) + # --- supervise queue (operator approvals) ------------------------------ + # + # The orchestrator owns the single DB *and* the live policy, so operator + # decisions are applied here, server-side, and reached over HTTP by the + # host TUI (no direct-DB access, one path for every backend). + + def supervise_pending(self) -> list[dict[str, object]]: + """All pending proposals across bottles, FIFO, as JSON dicts + (`Proposal.to_dict`, round-trippable via `Proposal.from_dict`).""" + return [p.to_dict() for p in list_all_pending_proposals()] + + def _record_for_slug(self, slug: str) -> BottleRecord | None: + """The live registry record whose metadata carries `slug`, or None + (e.g. the bottle was torn down before the operator responded).""" + for rec in self.registry.all(): + try: + meta = json.loads(rec.metadata) if rec.metadata else {} + except ValueError: + meta = {} + if isinstance(meta, dict) and meta.get("slug") == slug: + return rec + return None + + def supervise_respond( + self, + proposal_id: str, + *, + bottle_slug: str, + decision: str, + notes: str = "", + final_file: str | None = None, + ) -> tuple[bool, str]: + """Record an operator decision on a queued proposal, applying it + server-side. `decision` is approve/modify/reject. + + Approve/modify on an egress tool rewrites the bottle's policy so the + gateway serves the new routes on its next `/resolve` (the live apply); + then the queued Response is written (unblocking the agent's MCP call) + and an audit entry recorded — all against the one DB. Returns + (ok, error): ok=False with a message when the proposal or decision is + unknown, or the bottle is gone so an approval can't be applied.""" + status = _RESPOND_STATUS.get(decision) + if status is None: + return False, f"unknown decision {decision!r}" + try: + proposal = read_proposal(bottle_slug, proposal_id) + except FileNotFoundError: + return False, "no such proposal" + + diff_before, diff_after = "", "" + if status in (STATUS_APPROVED, STATUS_MODIFIED) and proposal.tool in _APPLY_TOOLS: + new_policy = final_file if final_file is not None else proposal.proposed_file + rec = self._record_for_slug(bottle_slug) + if rec is None: + return False, ( + f"bottle {bottle_slug!r} is no longer registered; " + "cannot apply the route change" + ) + diff_before, diff_after = rec.policy, new_policy + self.set_policy(rec.bottle_id, new_policy) + + write_response(bottle_slug, Response( + proposal_id=proposal_id, status=status, notes=notes, final_file=final_file, + )) + component = COMPONENT_FOR_TOOL.get(proposal.tool) + if component is not None: + write_audit_entry(AuditEntry( + timestamp=datetime.now(timezone.utc).isoformat(), + bottle_slug=bottle_slug, + component=component, + operator_action=status, + operator_notes=notes, + justification=proposal.justification, + diff=render_diff(diff_before, diff_after, label=component), + )) + return True, "" + # --- consolidated gateway ---------------------------------------------- def ensure_gateway(self) -> None: diff --git a/tests/unit/test_orchestrator_control_plane.py b/tests/unit/test_orchestrator_control_plane.py index 6816990..9fd4fb0 100644 --- a/tests/unit/test_orchestrator_control_plane.py +++ b/tests/unit/test_orchestrator_control_plane.py @@ -13,11 +13,19 @@ import threading import unittest import urllib.request from pathlib import Path +from unittest.mock import patch 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 +from bot_bottle.store_manager import StoreManager +from bot_bottle.supervise import ( + Proposal, + TOOL_EGRESS_ALLOW, + sha256_hex, + write_proposal, +) def _body(obj: object) -> bytes: @@ -235,5 +243,70 @@ class TestServerRoundTrip(unittest.TestCase): self.assertEqual(reg["bottle_id"], attr["bottle_id"]) +class TestDispatchSupervise(unittest.TestCase): + """The /supervise/* routes over the pure dispatch().""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + db = root / "db" / "bot-bottle.db" + db.parent.mkdir(parents=True) + self._env = patch.dict("os.environ", { + "BOT_BOTTLE_ROOT": str(root), + "SUPERVISE_DB_PATH": str(db), + }) + self._env.start() + self.store = RegistryStore(db) + self.store.migrate() + StoreManager(db).migrate() + secret = secrets.token_bytes(16) + self.orch = Orchestrator(self.store, StubBroker(secret), secret) + + def tearDown(self) -> None: + self._env.stop() + self._tmp.cleanup() + + def _queue(self, slug: str, proposed: str) -> str: + self.store.register( + "10.243.0.1", metadata=json.dumps({"slug": slug}), policy="routes: []\n") + p = Proposal.new( + bottle_slug=slug, tool=TOOL_EGRESS_ALLOW, proposed_file=proposed, + justification="need it", current_file_hash=sha256_hex(proposed)) + write_proposal(p) + return p.id + + def test_list_pending(self) -> None: + pid = self._queue("demo", "routes:\n - host: google.com\n") + status, payload = dispatch(self.orch, "GET", "/supervise/proposals", b"") + self.assertEqual(200, status) + proposals = payload["proposals"] + assert isinstance(proposals, list) + self.assertEqual(pid, proposals[0]["id"]) + + def test_respond_approve_applies_and_clears(self) -> None: + pid = self._queue("demo", "routes:\n - host: google.com\n") + status, payload = dispatch( + self.orch, "POST", "/supervise/respond", + _body({"proposal_id": pid, "bottle_slug": "demo", "decision": "approve"}), + ) + self.assertEqual(200, status) + self.assertTrue(payload["responded"]) + _, listing = dispatch(self.orch, "GET", "/supervise/proposals", b"") + self.assertEqual([], listing["proposals"]) + + def test_respond_requires_fields(self) -> None: + status, _ = dispatch( + self.orch, "POST", "/supervise/respond", _body({"decision": "approve"})) + self.assertEqual(400, status) + + def test_respond_unknown_proposal_conflicts(self) -> None: + status, payload = dispatch( + self.orch, "POST", "/supervise/respond", + _body({"proposal_id": "ghost", "bottle_slug": "demo", "decision": "approve"}), + ) + self.assertEqual(409, status) + self.assertIn("no such proposal", str(payload["error"])) + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_orchestrator_service.py b/tests/unit/test_orchestrator_service.py index c8139c2..8957ba0 100644 --- a/tests/unit/test_orchestrator_service.py +++ b/tests/unit/test_orchestrator_service.py @@ -2,15 +2,26 @@ from __future__ import annotations +import json import secrets import tempfile import unittest from pathlib import Path +from unittest.mock import patch from bot_bottle.orchestrator.broker import LaunchBroker, LaunchRequest, StubBroker from bot_bottle.orchestrator.registry import RegistryStore from bot_bottle.orchestrator.service import Orchestrator from bot_bottle.orchestrator.gateway import Gateway +from bot_bottle.store_manager import StoreManager +from bot_bottle.supervise import ( + Proposal, + STATUS_APPROVED, + TOOL_EGRESS_ALLOW, + read_response, + sha256_hex, + write_proposal, +) class _FailingBroker(LaunchBroker): @@ -155,5 +166,111 @@ class TestOrchestrator(unittest.TestCase): ) +class TestOrchestratorSupervise(unittest.TestCase): + """Operator-approval flow: the orchestrator applies the decision + server-side against the single DB (queue + policy + audit).""" + + def setUp(self) -> None: + self._tmp = tempfile.TemporaryDirectory() + root = Path(self._tmp.name) + db = root / "db" / "bot-bottle.db" + db.parent.mkdir(parents=True) + # One DB for registry + supervise queue + audit (as in the VM). + self._env = patch.dict("os.environ", { + "BOT_BOTTLE_ROOT": str(root), + "SUPERVISE_DB_PATH": str(db), + }) + self._env.start() + self.store = RegistryStore(db) + self.store.migrate() + StoreManager(db).migrate() + secret = secrets.token_bytes(16) + self.orch = Orchestrator(self.store, StubBroker(secret), secret) + + def tearDown(self) -> None: + self._env.stop() + self._tmp.cleanup() + + def _register(self, slug: str, policy: str) -> str: + rec = self.store.register( + "10.243.0.1", metadata=json.dumps({"slug": slug}), policy=policy) + return rec.bottle_id + + def _queue(self, slug: str, proposed: str) -> str: + p = Proposal.new( + bottle_slug=slug, tool=TOOL_EGRESS_ALLOW, proposed_file=proposed, + justification="need it", current_file_hash=sha256_hex(proposed)) + write_proposal(p) + return p.id + + def test_pending_lists_queued_proposal(self) -> None: + self._register("demo", "routes: []\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + pending = self.orch.supervise_pending() + self.assertEqual(1, len(pending)) + self.assertEqual(pid, pending[0]["id"]) + self.assertEqual("demo", pending[0]["bottle_slug"]) + + def test_approve_applies_policy_writes_response_and_clears_pending(self) -> None: + bottle_id = self._register("demo", "routes:\n - host: existing.com\n") + new_routes = "routes:\n - host: google.com\n" + pid = self._queue("demo", new_routes) + ok, err = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="approve") + self.assertTrue(ok, err) + # policy live-applied so /resolve serves the new routes + rec = self.store.get(bottle_id) + assert rec is not None + self.assertEqual(new_routes, rec.policy) + # response written -> agent unblocks, proposal no longer pending + self.assertEqual(STATUS_APPROVED, read_response("demo", pid).status) + self.assertEqual([], self.orch.supervise_pending()) + + def test_modify_applies_final_file_not_proposed(self) -> None: + bottle_id = self._register("demo", "routes: []\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + edited = "routes:\n - host: example.com\n" + ok, _ = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="modify", final_file=edited) + self.assertTrue(ok) + rec = self.store.get(bottle_id) + assert rec is not None + self.assertEqual(edited, rec.policy) + + def test_reject_leaves_policy_unchanged(self) -> None: + bottle_id = self._register("demo", "routes:\n - host: existing.com\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + ok, _ = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="reject", notes="no") + self.assertTrue(ok) + rec = self.store.get(bottle_id) + assert rec is not None + self.assertEqual("routes:\n - host: existing.com\n", rec.policy) + self.assertEqual("rejected", read_response("demo", pid).status) + + def test_unknown_proposal_is_error(self) -> None: + ok, err = self.orch.supervise_respond( + "ghost", bottle_slug="demo", decision="approve") + self.assertFalse(ok) + self.assertIn("no such proposal", err) + + def test_unknown_decision_is_error(self) -> None: + self._register("demo", "routes: []\n") + pid = self._queue("demo", "routes:\n - host: google.com\n") + ok, err = self.orch.supervise_respond( + pid, bottle_slug="demo", decision="bogus") + self.assertFalse(ok) + self.assertIn("unknown decision", err) + + def test_approve_when_bottle_gone_cannot_apply(self) -> None: + # Proposal queued but the bottle was torn down before the operator + # acted: an egress apply has no target, so respond fails closed. + pid = self._queue("ghost-bottle", "routes:\n - host: google.com\n") + ok, err = self.orch.supervise_respond( + pid, bottle_slug="ghost-bottle", decision="approve") + self.assertFalse(ok) + self.assertIn("no longer registered", err) + + if __name__ == "__main__": unittest.main()