feat(orchestrator): slice 6 — source-IP-keyed multi-tenant policy (#352)

The orchestrator side of the multi-tenant consolidated sidecar: hold each
bottle's sidecar policy and serve it by verified source IP, with live
reload. One shared sidecar can now get per-bottle config keyed on who's
calling.

  * registry.py — a `policy` column (migration v3, opaque JSON the sidecar
    interprets) on BottleRecord; `register(..., policy=)` stores it,
    `set_policy(bottle_id, policy)` updates it live, and `attribute` returns
    it (the source-IP-keyed resolution).
  * service.py — `launch_bottle(..., policy=)` and `set_policy`.
  * control_plane.py — `POST /bottles` accepts `policy`; `PUT
    /bottles/<id>/policy` live-reloads it; `POST /resolve` returns
    {bottle_id, policy} for a verified (source_ip, token) — the per-request
    call the multi-tenant sidecar makes; `/attribute` stays identity-only.

Scope note: this is the control-plane / state half. The data-plane half —
the egress mitmproxy addon (and git-gate) selecting allowlist / DLP /
token-injection per client IP by calling `/resolve` — is the next slice
(route agent bottles through the shared sidecar). The orchestrator stays
policy-agnostic: it stores and serves the blob verbatim.

Tests: registry policy store/update/persist; Orchestrator launch-with-policy
+ live set_policy; control-plane resolve returns policy (403 on bad token),
PUT policy updates / 404 / 400. Verified live over HTTP (launch -> resolve
-> PUT reload -> resolve reflects). Full suite green.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01WBMWTEtQdJ4W5UrWuLHCck

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:

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()

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):