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
This commit is contained in:
2026-07-13 22:12:18 -04:00
parent d9c8a4645d
commit 6492cc258a
2 changed files with 41 additions and 0 deletions
+17
View File
@@ -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]] = []