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
This commit is contained in:
2026-07-13 17:09:32 -04:00
parent 4692a92c19
commit af1d1ef304
6 changed files with 166 additions and 18 deletions
+37 -10
View File
@@ -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": [ <redacted record>, ... ]}
POST /bottles -> 201 {"bottle_id", "identity_token"} (launch)
body: {"source_ip", ["image_ref"], ["metadata"]}
DELETE /bottles/<bottle_id> -> 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": [ <redacted record>, ...]}
POST /bottles -> 201 {"bottle_id","identity_token"} (launch)
body: {"source_ip", ["image_ref"],
["metadata"], ["policy"]}
PUT /bottles/<bottle_id>/policy -> 200 {"updated": true} | 404 (live reload)
body: {"policy"}
DELETE /bottles/<bottle_id> -> 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")
+28 -4
View File
@@ -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:
+13 -4
View File
@@ -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: