feat(orchestrator): slice 13c(vii) — stable shared gateway CA

The consolidated gateway TLS-intercepts every bottle, so all agents must
trust ONE CA. Rather than host-generate + mount a CA, let the gateway's own
mitmproxy generate its CA into a persistent named volume and extract the
cert for agents — no host openssl, keeps gateway.py lean.

- DockerGateway mounts GATEWAY_CA_VOLUME at /home/mitmproxy/.mitmproxy so the
  self-generated CA survives container recreation (it must not rotate, or
  already-launched agents would stop trusting the gateway).
- ca_cert_pem(): read the CA cert (PEM) out of the running gateway
  (docker exec cat mitmproxy-ca-cert.pem) — the agent installs this into its
  trust store to accept the gateway's bumped certs.

pyright 0 errors; pylint 9.83/10; unit suite green (the 13 test_sidecar_init
/bin/sleep errors are pre-existing NixOS-local noise).

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

bot_bottle/backend/docker/consolidated_launch.py

@@ -0,0 +1,145 @@
+"""Consolidated bottle launch sequence for the docker backend (PRD 0070).
+
+Composes the orchestrator primitives into the register/teardown sequence that
+replaces the per-bottle sidecar bundle:
+
+  1. ensure the orchestrator control plane + shared gateway are up;
+  2. allocate the bottle a pinned source IP on the gateway network (the
+     attribution key), skipping the gateway's own address + live bottles;
+  3. register it (egress policy blob + slug metadata) → bottle id + identity
+     token;
+  4. provision its git-gate repos/creds into the running gateway.
+
+It returns a `LaunchContext` with everything the agent container needs to
+attach — network, pinned IP, the gateway's address (its proxy target), the
+orchestrator URL, and the identity token. The agent `docker run` itself is
+the backend's job (it owns provider provisioning); this owns the
+orchestrator-facing wiring so that sequence stays testable in isolation.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+from ...docker_cmd import run_docker
+from ...egress import EgressPlan
+from ...git_gate import GitGatePlan
+from ...orchestrator.client import OrchestratorClient
+from ...orchestrator.gateway import GATEWAY_NAME, GATEWAY_NETWORK
+from ...orchestrator.lifecycle import OrchestratorProcess
+from ...orchestrator.registration import registration_inputs
+from .gateway_net import next_free_ip
+from .gateway_provision import deprovision_git_gate, provision_git_gate
+
+
+class ConsolidatedLaunchError(RuntimeError):
+    """The consolidated register/provision sequence could not complete."""
+
+
+@dataclass(frozen=True)
+class LaunchContext:
+    """What the agent container needs to join the shared gateway."""
+
+    bottle_id: str
+    identity_token: str
+    source_ip: str          # the agent's pinned address (attribution key)
+    network: str            # the shared gateway network to attach to
+    gateway_ip: str         # the gateway's address — the agent's proxy target
+    orchestrator_url: str
+
+
+def _network_cidr(network: str) -> str:
+    """The gateway network's IPv4 subnet, or raise."""
+    proc = run_docker([
+        "docker", "network", "inspect",
+        "--format", "{{range .IPAM.Config}}{{.Subnet}}{{end}}", network,
+    ])
+    cidr = proc.stdout.strip()
+    if proc.returncode != 0 or not cidr:
+        raise ConsolidatedLaunchError(
+            f"gateway network {network} has no subnet: {proc.stderr.strip()}"
+        )
+    return cidr
+
+
+def _container_ip(name: str, network: str) -> str:
+    """A container's IPv4 address on `network`, or raise."""
+    proc = run_docker([
+        "docker", "inspect", "--format",
+        f'{{{{(index .NetworkSettings.Networks "{network}").IPAddress}}}}', name,
+    ])
+    ip = proc.stdout.strip()
+    if proc.returncode != 0 or not ip:
+        raise ConsolidatedLaunchError(
+            f"gateway {name} has no address on {network}: {proc.stderr.strip()}"
+        )
+    return ip
+
+
+def _taken_ips(client: OrchestratorClient, gateway_ip: str) -> list[str]:
+    """Every address already in use on the gateway network: the gateway
+    container plus every live bottle the registry knows."""
+    taken = [gateway_ip]
+    for rec in client.list_bottles():
+        src = rec.get("source_ip")
+        if isinstance(src, str) and src:
+            taken.append(src)
+    return taken
+
+
+def launch_consolidated(
+    egress_plan: EgressPlan,
+    git_gate_plan: GitGatePlan,
+    *,
+    image_ref: str = "",
+    process: OrchestratorProcess | None = None,
+    gateway_name: str = GATEWAY_NAME,
+    network: str = GATEWAY_NETWORK,
+) -> LaunchContext:
+    """Ensure the orchestrator + gateway are up, allocate + register the
+    bottle, and provision its git-gate state. Returns the agent's attach
+    context. Raises `ConsolidatedLaunchError` (or the primitives' own errors)
+    if any step fails — the caller tears down on failure."""
+    process = process or OrchestratorProcess()
+    url = process.ensure_running()
+    client = OrchestratorClient(url)
+
+    cidr = _network_cidr(network)
+    gateway_ip = _container_ip(gateway_name, network)
+    source_ip = next_free_ip(cidr, _taken_ips(client, gateway_ip))
+
+    inputs = registration_inputs(egress_plan)
+    reg = client.register_bottle(
+        source_ip, image_ref=image_ref, policy=inputs.policy, metadata=inputs.metadata,
+    )
+    try:
+        provision_git_gate(gateway_name, reg.bottle_id, git_gate_plan)
+    except Exception:
+        # Roll the registration back so a provisioning failure leaves no orphan.
+        client.teardown_bottle(reg.bottle_id)
+        raise
+    return LaunchContext(
+        bottle_id=reg.bottle_id,
+        identity_token=reg.identity_token,
+        source_ip=source_ip,
+        network=network,
+        gateway_ip=gateway_ip,
+        orchestrator_url=url,
+    )
+
+
+def teardown_consolidated(
+    bottle_id: str, *, orchestrator_url: str, gateway_name: str = GATEWAY_NAME,
+) -> None:
+    """Deregister the bottle and remove its git-gate state from the gateway.
+    Both steps are idempotent so this is safe from a cleanup trap."""
+    OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
+    deprovision_git_gate(gateway_name, bottle_id)
+
+
+__all__ = [
+    "LaunchContext",
+    "launch_consolidated",
+    "teardown_consolidated",
+    "ConsolidatedLaunchError",
+]

bot_bottle/orchestrator/gateway.py

@@ -31,6 +31,14 @@ GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
 # the source IP the gateway attributes by is the address on this network.
 GATEWAY_NETWORK = "bot-bottle-gateway"
 
+# mitmproxy's CA dir in the bundle. A persistent named volume here keeps the
+# gateway's self-generated CA STABLE across container recreation — every agent
+# installs this one CA to trust the shared gateway's TLS interception, so it
+# must not rotate when the gateway restarts.
+MITMPROXY_HOME = "/home/mitmproxy/.mitmproxy"
+GATEWAY_CA_VOLUME = "bot-bottle-gateway-mitmproxy"
+GATEWAY_CA_CERT = f"{MITMPROXY_HOME}/mitmproxy-ca-cert.pem"
+
 # The real sidecar-bundle image + its Dockerfile. Kept as a local constant
 # rather than imported from backend.docker.sidecar_bundle, which would drag
 # the whole backend layer into the lean orchestrator (see #359); unify when
@@ -143,11 +151,26 @@ class DockerGateway(Gateway):
             "--name", self.name,
             "--label", GATEWAY_LABEL,
             "--network", self.network,
+            # Persist the self-generated CA so it survives restarts (agents
+            # trust it) — see GATEWAY_CA_VOLUME.
+            "--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
             self.image_ref,
         ])
         if proc.returncode != 0:
             raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
 
+    def ca_cert_pem(self) -> str:
+        """The gateway's CA certificate (PEM) that agents install to trust its
+        TLS interception. Read from the running container; raises if the
+        gateway hasn't generated it yet (it's created on first mitmproxy
+        start)."""
+        proc = run_docker(["docker", "exec", self.name, "cat", GATEWAY_CA_CERT])
+        if proc.returncode != 0 or not proc.stdout.strip():
+            raise GatewayError(
+                f"gateway CA cert not available: {proc.stderr.strip() or 'empty'}"
+            )
+        return proc.stdout
+
     def stop(self) -> None:
         proc = run_docker(["docker", "rm", "--force", self.name])
         if proc.returncode != 0 and "No such container" not in proc.stderr:
@@ -157,4 +180,5 @@ class DockerGateway(Gateway):
 __all__ = [
     "Gateway", "DockerGateway", "GatewayError",
     "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
+    "GATEWAY_CA_VOLUME", "GATEWAY_CA_CERT",
 ]

tests/unit/test_consolidated_launch.py

@@ -0,0 +1,91 @@
+"""Unit: consolidated launch sequence — compose the orchestrator primitives (PRD 0070)."""
+
+from __future__ import annotations
+
+import unittest
+from pathlib import Path
+from unittest.mock import MagicMock, Mock, patch
+
+from bot_bottle.backend.docker.consolidated_launch import (
+    launch_consolidated,
+    teardown_consolidated,
+)
+from bot_bottle.egress import EgressPlan, EgressRoute
+from bot_bottle.git_gate import GitGatePlan
+from bot_bottle.orchestrator.client import RegisteredBottle
+
+_MOD = "bot_bottle.backend.docker.consolidated_launch"
+
+
+def _egress_plan() -> EgressPlan:
+    return EgressPlan(
+        slug="demo", routes_path=Path("/x"), routes=(EgressRoute(host="api.example.com"),),
+        token_env_map={},
+    )
+
+
+def _git_plan() -> GitGatePlan:
+    return GitGatePlan(
+        slug="demo", entrypoint_script=Path(), hook_script=Path(),
+        access_hook_script=Path(), upstreams=(),
+    )
+
+
+def _client(*, bottles: list[dict[str, object]] | None = None) -> Mock:
+    c = Mock()
+    c.list_bottles.return_value = bottles or []
+    c.register_bottle.return_value = RegisteredBottle("b1", "tok")
+    return c
+
+
+class TestLaunchConsolidated(unittest.TestCase):
+    def _run(self, client: Mock, provision: Mock | None = None):
+        process = MagicMock()
+        process.ensure_running.return_value = "http://orch:8080"
+        with patch(f"{_MOD}._network_cidr", return_value="172.18.0.0/16"), \
+                patch(f"{_MOD}._container_ip", return_value="172.18.0.2"), \
+                patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.provision_git_gate", provision or Mock()):
+            return launch_consolidated(_egress_plan(), _git_plan(), process=process)
+
+    def test_allocates_ip_registers_and_provisions(self) -> None:
+        client = _client()
+        provision = Mock()
+        ctx = self._run(client, provision)
+        # .1 is the router, .2 is the gateway → first bottle gets .3.
+        self.assertEqual("172.18.0.3", ctx.source_ip)
+        self.assertEqual("172.18.0.2", ctx.gateway_ip)
+        self.assertEqual("b1", ctx.bottle_id)
+        self.assertEqual("tok", ctx.identity_token)
+        self.assertEqual("http://orch:8080", ctx.orchestrator_url)
+        # Registered with the source IP + the egress policy blob.
+        kwargs = client.register_bottle.call_args
+        self.assertEqual("172.18.0.3", kwargs.args[0])
+        self.assertIn("api.example.com", kwargs.kwargs["policy"])
+        provision.assert_called_once()
+
+    def test_skips_gateway_and_live_bottle_addresses(self) -> None:
+        client = _client(bottles=[{"source_ip": "172.18.0.3"}])
+        ctx = self._run(client)
+        self.assertEqual("172.18.0.4", ctx.source_ip)  # .2 gw, .3 taken → .4
+
+    def test_provision_failure_rolls_back_registration(self) -> None:
+        client = _client()
+        provision = Mock(side_effect=RuntimeError("provision boom"))
+        with self.assertRaises(RuntimeError):
+            self._run(client, provision)
+        client.teardown_bottle.assert_called_once_with("b1")  # no orphan left
+
+
+class TestTeardownConsolidated(unittest.TestCase):
+    def test_deregisters_and_deprovisions(self) -> None:
+        client = Mock()
+        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
+                patch(f"{_MOD}.deprovision_git_gate") as deprov:
+            teardown_consolidated("b1", orchestrator_url="http://orch:8080")
+        client.teardown_bottle.assert_called_once_with("b1")
+        deprov.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_orchestrator_gateway.py

@@ -6,11 +6,15 @@ import unittest
 from unittest.mock import Mock, patch
 
 from bot_bottle.orchestrator.gateway import (
+    GATEWAY_CA_CERT,
     GATEWAY_NAME,
     DockerGateway,
     GatewayError,
 )
 
+
+_CA_PEM = "-----BEGIN CERTIFICATE-----\nMII...\n-----END CERTIFICATE-----\n"
+
 _RUN_DOCKER = "bot_bottle.orchestrator.gateway.run_docker"
 
 
@@ -53,6 +57,8 @@ class TestDockerGateway(unittest.TestCase):
         self.assertIn("bot-bottle-sidecars:latest", runs[0])
         # Runs on the shared gateway network so agents can reach it by IP.
         self.assertEqual(self.sc.network, runs[0][runs[0].index("--network") + 1])
+        # Persists its CA on a named volume so agents keep trusting it.
+        self.assertTrue(any("mitmproxy" in a for a in runs[0]))
 
     def test_ensure_running_creates_network_when_missing(self) -> None:
         calls: list[list[str]] = []
@@ -68,6 +74,17 @@ class TestDockerGateway(unittest.TestCase):
         creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
         self.assertEqual([["docker", "network", "create", self.sc.network]], creates)
 
+    def test_ca_cert_pem_reads_from_container(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(stdout=_CA_PEM)) as m:
+            self.assertEqual(_CA_PEM, self.sc.ca_cert_pem())
+        argv = m.call_args.args[0]
+        self.assertEqual(["docker", "exec", self.sc.name, "cat", GATEWAY_CA_CERT], argv)
+
+    def test_ca_cert_pem_raises_when_absent(self) -> None:
+        with patch(_RUN_DOCKER, return_value=_proc(returncode=1, stderr="No such file")):
+            with self.assertRaises(GatewayError):
+                self.sc.ca_cert_pem()
+
     def test_ensure_running_reuses_existing_network(self) -> None:
         calls: list[list[str]] = []