PRD: reprovision on gateway startup - bottle ca (chunk 1/5) #519
@@ -512,6 +512,27 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
|
||||
compose ls`; firecracker cross-references its running gateway
|
||||
containers against per-bottle metadata."""
|
||||
|
||||
@abstractmethod
|
||||
def attach_bottled_agents_to_gateway(self) -> None:
|
||||
"""Reconcile every already-running bottle against the freshly-(re)booted
|
||||
gateway (PRD 0081). The host calls this on the gateway bring-up path —
|
||||
the cold-boot branch only, never on an adopt of a healthy current
|
||||
gateway — so a gateway rebuild/restart doesn't silently break running
|
||||
bottles' gateway-dependent state.
|
||||
|
||||
Reconciling running bottles is a backend responsibility: only the
|
||||
backend can enumerate its agents and reach them (firecracker over SSH,
|
||||
docker/macOS over `exec`/`cp`), so every backend must implement it —
|
||||
cross-backend by construction. Each per-bottle step is best-effort: one
|
||||
unreachable or malformed bottle is logged and skipped so it never blocks
|
||||
|
|
||||
the others or the gateway coming up.
|
||||
|
||||
Currently reconciles the **CA**: it replaces each agent's trusted CA
|
||||
with the current gateway CA (unconditional — there is one gateway, so no
|
||||
fingerprint match is needed), which rotates the CA for free on every
|
||||
bring-up. Git-gate re-provision and egress-token restore fold into this
|
||||
same reconcile on the same trigger (see #516)."""
|
||||
|
||||
@classmethod
|
||||
@abstractmethod
|
||||
def is_available(cls) -> bool:
|
||||
|
||||
@@ -140,3 +140,7 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
|
||||
|
||||
def enumerate_active(self) -> Sequence[ActiveAgent]:
|
||||
return _enumerate.enumerate_active()
|
||||
|
||||
def attach_bottled_agents_to_gateway(self) -> None:
|
||||
from .consolidated_launch import attach_bottled_agents_to_gateway
|
||||
attach_bottled_agents_to_gateway()
|
||||
|
||||
@@ -13,6 +13,7 @@ orchestrator-facing wiring so that sequence stays testable in isolation.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ... import log
|
||||
@@ -25,6 +26,7 @@ from .infra import INFRA_NAME, DockerInfraService
|
||||
from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
|
||||
from ...orchestrator.reprovision import reprovision_bottles
|
||||
from ..provision_bottle import deprovision_bottle, provision_bottle
|
||||
from ..util import AGENT_CA_PATH
|
||||
from .gateway_transport import DockerGatewayTransport
|
||||
from .gateway_net import next_free_ip
|
||||
|
||||
@@ -79,6 +81,77 @@ def _network_container_ips(network: str) -> list[str]:
|
||||
return ips
|
||||
|
||||
|
||||
def _push_ca_to_container(container: str, ca_pem: str) -> None:
|
||||
"""Replace one running agent container's trusted gateway CA with `ca_pem`
|
||||
and rebuild its trust store via `docker cp` + `docker exec`. Unconditional
|
||||
install — there is one gateway, so no fingerprint match is needed. Raises
|
||||
`ConsolidatedLaunchError` on failure."""
|
||||
mkdir = run_docker(
|
||||
["docker", "exec", container, "mkdir", "-p",
|
||||
"/usr/local/share/ca-certificates"]
|
||||
)
|
||||
if mkdir.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
f"CA push to {container} failed (mkdir): {mkdir.stderr.strip()}"
|
||||
)
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".crt") as tmp:
|
||||
tmp.write(ca_pem)
|
||||
tmp.flush()
|
||||
cp = run_docker(["docker", "cp", tmp.name, f"{container}:{AGENT_CA_PATH}"])
|
||||
if cp.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
f"CA push to {container} failed (cp): {cp.stderr.strip()}"
|
||||
)
|
||||
ex = run_docker(
|
||||
["docker", "exec", container, "sh", "-c",
|
||||
f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates"]
|
||||
)
|
||||
if ex.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
f"CA push to {container} failed (update-ca-certificates): "
|
||||
f"{ex.stderr.strip()}"
|
||||
)
|
||||
|
||||
|
||||
def attach_bottled_agents_to_gateway(
|
||||
ca_pem: str | None = None,
|
||||
*,
|
||||
network: str = GATEWAY_NETWORK,
|
||||
gateway_name: str = INFRA_NAME,
|
||||
) -> None:
|
||||
"""Reconcile every running agent container against the freshly-(re)created
|
||||
gateway (PRD 0081): replace each agent's trusted CA with the current gateway
|
||||
CA, so a gateway rebuild doesn't silently break their TLS egress (#510).
|
||||
|
||||
Per-container steps are best-effort: one failure is logged and skipped so it
|
||||
never blocks the others or the gateway coming up."""
|
||||
if ca_pem is None:
|
||||
ca_pem = DockerInfraService().gateway().ca_cert_pem()
|
||||
try:
|
||||
proc = run_docker([
|
||||
"docker", "network", "inspect", "--format",
|
||||
"{{range .Containers}}{{.Name}}\n{{end}}", network,
|
||||
])
|
||||
except OSError as exc:
|
||||
log.info(f"CA reconcile skipped: {exc}")
|
||||
return
|
||||
reconciled = 0
|
||||
for line in proc.stdout.splitlines():
|
||||
name = line.strip()
|
||||
if not name or name == gateway_name:
|
||||
continue
|
||||
|
didericis
commented
Same comment as below, never skip Same comment as below, never skip
|
||||
try:
|
||||
_push_ca_to_container(name, ca_pem)
|
||||
reconciled += 1
|
||||
except (OSError, ConsolidatedLaunchError) as exc:
|
||||
log.info(f"CA reconcile skipped for {name}: {exc}")
|
||||
if reconciled:
|
||||
log.info(
|
||||
"reconciled gateway CA into running docker bottle(s)",
|
||||
context={"count": reconciled},
|
||||
)
|
||||
|
||||
|
||||
def _reprovision_running_bottles(
|
||||
orchestrator_url: str,
|
||||
network: str = GATEWAY_NETWORK,
|
||||
@@ -185,6 +258,7 @@ def deprovision_consolidated(
|
||||
__all__ = [
|
||||
"LaunchContext",
|
||||
"launch_consolidated",
|
||||
"attach_bottled_agents_to_gateway",
|
||||
"deprovision_consolidated",
|
||||
"ConsolidatedLaunchError",
|
||||
]
|
||||
|
||||
@@ -200,7 +200,7 @@ class DockerGateway(Gateway):
|
||||
f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
|
||||
)
|
||||
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> None:
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> bool:
|
||||
# Bind to this orchestrator (PRD 0070): stash the URL its daemons resolve
|
||||
# policy against + the pre-minted `gateway` token they present, then bring
|
||||
# the container up. The gateway never mints, so it holds no signing key —
|
||||
@@ -227,7 +227,7 @@ class DockerGateway(Gateway):
|
||||
# just when the container is absent.
|
||||
self._ensure_network()
|
||||
if self.is_running() and self._running_image_is_current():
|
||||
return
|
||||
return False
|
||||
# Clear any stale (stopped OR outdated-image) container holding the
|
||||
# fixed name, then start fresh. `rm --force` on an absent name is a
|
||||
# tolerated no-op.
|
||||
@@ -271,6 +271,9 @@ class DockerGateway(Gateway):
|
||||
# pre-connect window (they retry /resolve per request).
|
||||
if self._control_network:
|
||||
self._connect_control_network()
|
||||
# (Re)created a fresh container — signal a cold boot so the caller
|
||||
# reconciles running bottles against it (PRD 0081).
|
||||
return True
|
||||
|
||||
def _connect_control_network(self) -> None:
|
||||
"""Ensure the `--internal` control network exists and attach the gateway
|
||||
|
||||
@@ -142,9 +142,16 @@ class DockerInfraService(InfraService):
|
||||
# mints the role-scoped `gateway` JWT here and hands it to the gateway,
|
||||
# which never sees the key (#469). `connect_to_orchestrator` is
|
||||
# idempotent.
|
||||
gateway.connect_to_orchestrator(
|
||||
cold_booted = gateway.connect_to_orchestrator(
|
||||
orchestrator.gateway_url(), orchestrator.mint_gateway_token(),
|
||||
)
|
||||
# A (re)created gateway container minted/mounted its CA afresh and lost
|
||||
# every bottle's per-bottle state; reconcile the already-running bottles
|
||||
# against it (PRD 0081). Skipped when a healthy current gateway was left
|
||||
# untouched — no cold boot, nothing to reconcile.
|
||||
if cold_booted:
|
||||
from .backend import DockerBottleBackend
|
||||
DockerBottleBackend().attach_bottled_agents_to_gateway()
|
||||
return orchestrator.url()
|
||||
|
||||
def stop(self) -> None:
|
||||
|
||||
@@ -119,6 +119,10 @@ class FirecrackerBottleBackend(
|
||||
def enumerate_active(self) -> Sequence[ActiveAgent]:
|
||||
return _enumerate.enumerate_active()
|
||||
|
||||
def attach_bottled_agents_to_gateway(self) -> None:
|
||||
from .consolidated_launch import attach_bottled_agents_to_gateway
|
||||
attach_bottled_agents_to_gateway()
|
||||
|
||||
def supervise_mcp_url(self, plan: FirecrackerBottlePlan) -> str:
|
||||
return plan.agent_supervise_url
|
||||
|
||||
|
||||
@@ -32,6 +32,7 @@ from dataclasses import dataclass
|
||||
from pathlib import Path
|
||||
|
||||
from ...egress import EgressPlan
|
||||
from ...gateway import GatewayError
|
||||
from ...git_gate import GitGatePlan
|
||||
from ...log import info
|
||||
from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
|
||||
@@ -41,6 +42,7 @@ from ...orchestrator.lifecycle import (
|
||||
from ...orchestrator.reprovision import reprovision_bottles
|
||||
from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
|
||||
from ..provision_bottle import deprovision_bottle, provision_bottle
|
||||
from ..util import AGENT_CA_PATH
|
||||
from . import cleanup, util
|
||||
from .gateway import FirecrackerGateway
|
||||
from .infra import FirecrackerInfraService
|
||||
@@ -89,6 +91,56 @@ def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> Non
|
||||
)
|
||||
|
||||
|
||||
def _push_gateway_ca_to_agent(private_key: Path, guest_ip: str, ca_pem: str) -> None:
|
||||
"""Replace one running agent VM's trusted gateway CA with `ca_pem` and
|
||||
rebuild its trust store over SSH. Unconditional install — there is one
|
||||
gateway, so no fingerprint match is needed. Bare-pipe input keeps the cert
|
||||
off argv. Raises `ConsolidatedLaunchError` on failure."""
|
||||
install = (
|
||||
"umask 022; mkdir -p /usr/local/share/ca-certificates && "
|
||||
f"cat > {AGENT_CA_PATH} && chmod 644 {AGENT_CA_PATH} && "
|
||||
"update-ca-certificates"
|
||||
)
|
||||
proc = subprocess.run(
|
||||
util.ssh_base_argv(private_key, guest_ip) + [install],
|
||||
input=ca_pem, capture_output=True, text=True, check=False,
|
||||
)
|
||||
if proc.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
|
didericis
commented
Nit, but should rename this "InfraLaunchError" and have a consistent Nit, but should rename this "InfraLaunchError" and have a consistent `infra_launch` file across backends. "consolidated launch" is unclear. In glossary explain "infra" refers to the gateway and orchestrator services.
|
||||
f"CA push to {guest_ip} failed: "
|
||||
f"{proc.stderr.strip() or '<no stderr>'}"
|
||||
)
|
||||
|
||||
|
||||
def attach_bottled_agents_to_gateway() -> None:
|
||||
"""Reconcile every running agent VM against the freshly-booted gateway
|
||||
(PRD 0081). Replaces each agent's trusted CA with the current gateway CA —
|
||||
the gateway rootfs is ephemeral, so a cold boot mints a fresh CA and this is
|
||||
what distributes it (a gateway rebuild doubles as a free CA rotation, #510).
|
||||
|
||||
Per-bottle steps are best-effort: one unreachable or malformed VM is logged
|
||||
and skipped so it never blocks the others or the gateway coming up."""
|
||||
gateway = FirecrackerGateway()
|
||||
try:
|
||||
ca_pem = gateway.ca_cert_pem()
|
||||
except GatewayError as exc:
|
||||
info(f"bring-up reconcile skipped: gateway CA unavailable: {exc}")
|
||||
|
didericis
commented
Don't skip, do a hard fail here Don't skip, do a hard fail here
|
||||
return
|
||||
reconciled = 0
|
||||
for run_dir in cleanup.live_run_dirs():
|
||||
guest_ip = _guest_ip_from_config(run_dir / "config.json")
|
||||
private_key = run_dir / "bottle_id_ed25519"
|
||||
if not guest_ip or not private_key.is_file():
|
||||
continue
|
||||
try:
|
||||
_push_gateway_ca_to_agent(private_key, guest_ip, ca_pem)
|
||||
reconciled += 1
|
||||
except (OSError, ConsolidatedLaunchError) as exc:
|
||||
info(f"CA reconcile skipped for {guest_ip}: {exc}")
|
||||
|
didericis
commented
Same, do a hard fail/never skip. If the attachment doesn't work we want to vocalize it Same, do a hard fail/never skip. If the attachment doesn't work we want to vocalize it
|
||||
if reconciled:
|
||||
info(f"reconciled gateway CA into {reconciled} running Firecracker bottle(s)")
|
||||
|
||||
|
||||
def _reprovision_running_bottles(client: OrchestratorClient) -> None:
|
||||
"""Read keys from live agent VMs and restore the restarted gateway."""
|
||||
try:
|
||||
@@ -161,6 +213,7 @@ def deprovision_consolidated(
|
||||
__all__ = [
|
||||
"LaunchContext",
|
||||
"launch_consolidated",
|
||||
"attach_bottled_agents_to_gateway",
|
||||
"deprovision_consolidated",
|
||||
"ConsolidatedLaunchError",
|
||||
"OrchestratorStartError",
|
||||
|
||||
@@ -69,13 +69,17 @@ class FirecrackerGateway(Gateway):
|
||||
self._orchestrator_url = ""
|
||||
self._gateway_token = ""
|
||||
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> None:
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> bool:
|
||||
"""Boot the gateway VM on its link resolving policy against
|
||||
`orchestrator_url`, then seed `gateway_token` over SSH. The orchestrator's
|
||||
guest IP is parsed from the URL and passed on the cmdline (`bb_orch=`), so
|
||||
the baked init stays IP-independent. Boot the orchestrator first — the
|
||||
gateway daemons reach it at startup. The gateway never mints, so it holds
|
||||
no signing key, only this token (#469)."""
|
||||
no signing key, only this token (#469).
|
||||
|
||||
Always returns True: the firecracker gateway VM is booted fresh here
|
||||
(this runs only on the infra cold-boot path), so it is always a cold
|
||||
boot that minted a new CA — the caller reconciles running bottles."""
|
||||
self._orchestrator_url = orchestrator_url
|
||||
self._gateway_token = gateway_token
|
||||
if not self._orchestrator_url:
|
||||
@@ -105,6 +109,7 @@ class FirecrackerGateway(Gateway):
|
||||
"the gateway JWT to the gateway VM (its data plane will not start)",
|
||||
)
|
||||
self._vm = vm
|
||||
return True
|
||||
|
||||
def is_running(self) -> bool:
|
||||
return infra_vm._pidfile_alive(infra_vm._gw_dir())
|
||||
|
||||
@@ -67,9 +67,17 @@ class FirecrackerInfraService(InfraService):
|
||||
# holds the signing key) mints the role-scoped `gateway` JWT for the
|
||||
# gateway, which never sees the key (#469).
|
||||
orchestrator.ensure_running(startup_timeout=startup_timeout)
|
||||
self.gateway().connect_to_orchestrator(
|
||||
cold_booted = self.gateway().connect_to_orchestrator(
|
||||
orchestrator.gateway_url(), orchestrator.mint_gateway_token())
|
||||
infra_vm.record_booted_version(want)
|
||||
# The fresh gateway rootfs minted a new CA and lost every bottle's
|
||||
# git-gate/token state; reconcile the already-running bottles against
|
||||
# it so a gateway rebuild doesn't silently break their egress
|
||||
# (PRD 0081). Cold-boot only — the adopt fast-paths above never reach
|
||||
# here, so a healthy current gateway is left untouched.
|
||||
if cold_booted:
|
||||
from .backend import FirecrackerBottleBackend
|
||||
FirecrackerBottleBackend().attach_bottled_agents_to_gateway()
|
||||
return url
|
||||
|
||||
def stop(self) -> None:
|
||||
|
||||
@@ -117,5 +117,9 @@ class MacosContainerBottleBackend(
|
||||
def enumerate_active(self) -> Sequence[ActiveAgent]:
|
||||
return _enumerate.enumerate_active()
|
||||
|
||||
def attach_bottled_agents_to_gateway(self) -> None:
|
||||
from .consolidated_launch import attach_bottled_agents_to_gateway
|
||||
attach_bottled_agents_to_gateway()
|
||||
|
||||
def supervise_mcp_url(self, plan: MacosContainerBottlePlan) -> str:
|
||||
return plan.agent_supervise_url
|
||||
|
||||
@@ -34,6 +34,7 @@ every real command arrives through `container exec`.
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tempfile
|
||||
from dataclasses import dataclass
|
||||
|
||||
from ...egress import EgressPlan
|
||||
@@ -43,6 +44,7 @@ from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
|
||||
from ...orchestrator.reprovision import reprovision_bottles
|
||||
from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
|
||||
from ..provision_bottle import deprovision_bottle, provision_bottle
|
||||
from ..util import AGENT_CA_PATH
|
||||
from . import util as container_mod
|
||||
from .enumerate import CONTAINER_NAME_PREFIX, EnumerationError, enumerate_active
|
||||
from .gateway import GATEWAY_NETWORK
|
||||
@@ -98,6 +100,66 @@ def ensure_gateway(
|
||||
return endpoint
|
||||
|
||||
|
||||
def _push_ca_to_container(name: str, ca_pem: str) -> None:
|
||||
"""Replace one running agent container's trusted gateway CA with `ca_pem`
|
||||
and rebuild its trust store via `container cp` + `container exec`.
|
||||
Unconditional install — there is one gateway, so no fingerprint match is
|
||||
needed. Raises `ConsolidatedLaunchError` on failure."""
|
||||
mkdir = container_mod.run_container_argv(
|
||||
["container", "exec", name, "mkdir", "-p",
|
||||
"/usr/local/share/ca-certificates"]
|
||||
)
|
||||
if mkdir.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
f"CA push to {name} failed (mkdir): {(mkdir.stderr or '').strip()}"
|
||||
)
|
||||
with tempfile.NamedTemporaryFile("w", suffix=".crt") as tmp:
|
||||
tmp.write(ca_pem)
|
||||
tmp.flush()
|
||||
cp = container_mod.run_container_argv(
|
||||
["container", "cp", tmp.name, f"{name}:{AGENT_CA_PATH}"]
|
||||
)
|
||||
if cp.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
f"CA push to {name} failed (cp): {(cp.stderr or '').strip()}"
|
||||
)
|
||||
ex = container_mod.run_container_argv(
|
||||
["container", "exec", name, "sh", "-c",
|
||||
f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates"]
|
||||
)
|
||||
if ex.returncode != 0:
|
||||
raise ConsolidatedLaunchError(
|
||||
f"CA push to {name} failed (update-ca-certificates): "
|
||||
f"{(ex.stderr or '').strip()}"
|
||||
)
|
||||
|
||||
|
||||
def attach_bottled_agents_to_gateway(ca_pem: str | None = None) -> None:
|
||||
"""Reconcile every running agent container against the freshly-(re)created
|
||||
gateway (PRD 0081): replace each agent's trusted CA with the current gateway
|
||||
CA, so a gateway rebuild doesn't silently break their TLS egress (#510).
|
||||
|
||||
Per-container steps are best-effort: one failure is logged and skipped so it
|
||||
never blocks the others or the gateway coming up."""
|
||||
if ca_pem is None:
|
||||
ca_pem = MacosInfraService().ca_cert_pem()
|
||||
try:
|
||||
agents = list(enumerate_active())
|
||||
except (EnumerationError, OSError) as exc:
|
||||
info(f"CA reconcile skipped: {exc}")
|
||||
|
didericis
commented
same issue as before/should fail hard same issue as before/should fail hard
|
||||
return
|
||||
reconciled = 0
|
||||
for agent in agents:
|
||||
name = f"{CONTAINER_NAME_PREFIX}{agent.slug}"
|
||||
try:
|
||||
_push_ca_to_container(name, ca_pem)
|
||||
reconciled += 1
|
||||
except (OSError, ConsolidatedLaunchError) as exc:
|
||||
info(f"CA reconcile skipped for {name}: {exc}")
|
||||
if reconciled:
|
||||
info(f"reconciled gateway CA into {reconciled} running macOS bottle(s)")
|
||||
|
||||
|
||||
def _reprovision_running_bottles(endpoint: GatewayEndpoint) -> None:
|
||||
"""Recover keys from live Apple containers and restore gateway tokens."""
|
||||
try:
|
||||
@@ -200,6 +262,7 @@ __all__ = [
|
||||
"GatewayEndpoint",
|
||||
"LaunchContext",
|
||||
"ensure_gateway",
|
||||
"attach_bottled_agents_to_gateway",
|
||||
"live_source_ips",
|
||||
"register_agent",
|
||||
"deprovision_consolidated",
|
||||
|
||||
@@ -104,11 +104,15 @@ class MacosGateway(Gateway):
|
||||
container_mod.build_image(
|
||||
self.image_ref, str(self._repo_root), dockerfile="Dockerfile.gateway")
|
||||
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> None:
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> bool:
|
||||
"""Bind the gateway to this orchestrator and (re)start it, dual-homed on
|
||||
the agent + control networks, resolving policy against `orchestrator_url`
|
||||
(the orchestrator's control-network address — Apple has no container
|
||||
DNS) and presenting `gateway_token`."""
|
||||
DNS) and presenting `gateway_token`.
|
||||
|
||||
Always returns True: this recreates the gateway container unconditionally
|
||||
(a cold boot), so the caller reconciles running bottles against it
|
||||
(PRD 0081)."""
|
||||
self._orchestrator_url = orchestrator_url
|
||||
self._gateway_token = gateway_token
|
||||
# Fail closed on a missing policy source or token: the resolver-only data
|
||||
@@ -156,6 +160,7 @@ class MacosGateway(Gateway):
|
||||
f"gateway container failed to start: "
|
||||
f"{(result.stderr or '').strip() or '<no stderr>'}"
|
||||
)
|
||||
return True
|
||||
|
||||
def is_running(self) -> bool:
|
||||
return container_mod.container_is_running(self.name)
|
||||
|
||||
@@ -138,7 +138,15 @@ class MacosInfraService(InfraService):
|
||||
# `gateway` JWT and hands it to the gateway, which never sees the key
|
||||
# (#469).
|
||||
url = orchestrator.url()
|
||||
gateway.connect_to_orchestrator(url, orchestrator.mint_gateway_token())
|
||||
cold_booted = gateway.connect_to_orchestrator(
|
||||
url, orchestrator.mint_gateway_token())
|
||||
# A (re)created gateway container minted/mounted its CA afresh and lost
|
||||
# every bottle's per-bottle state; reconcile the already-running bottles
|
||||
# against it (PRD 0081). Skipped when a healthy current gateway was left
|
||||
# untouched — no cold boot, nothing to reconcile.
|
||||
if cold_booted:
|
||||
from .backend import MacosContainerBottleBackend
|
||||
MacosContainerBottleBackend().attach_bottled_agents_to_gateway()
|
||||
return url
|
||||
|
||||
def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
|
||||
|
||||
@@ -122,12 +122,18 @@ class Gateway(abc.ABC):
|
||||
return
|
||||
|
||||
@abc.abstractmethod
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> None:
|
||||
def connect_to_orchestrator(self, orchestrator_url: str, gateway_token: str) -> bool:
|
||||
"""Bind the gateway to this orchestrator and bring it up: store the URL +
|
||||
the pre-minted `gateway` token as instance state, then (re)start the
|
||||
gateway unit carrying the mitmproxy CA + that token, resolving policy
|
||||
against `orchestrator_url`. Idempotent — a healthy, current gateway on
|
||||
the same binding is left alone; a changed binding reconciles it."""
|
||||
the same binding is left alone; a changed binding reconciles it.
|
||||
|
||||
Returns True when the gateway was actually (re)brought up (a cold boot,
|
||||
so its mitmproxy minted a fresh CA and lost its per-bottle state), False
|
||||
when a healthy current gateway was left untouched. The bring-up flow
|
||||
uses this as the cold-boot signal that gates the running-bottle
|
||||
reconcile (`attach_bottled_agents_to_gateway`, PRD 0081)."""
|
||||
|
||||
@abc.abstractmethod
|
||||
def is_running(self) -> bool:
|
||||
|
||||
@@ -159,5 +159,126 @@ class TestFirecrackerReprovision(unittest.TestCase):
|
||||
restore.assert_called_once_with(client, {})
|
||||
|
||||
|
||||
class TestFirecrackerCaReconcile(unittest.TestCase):
|
||||
"""PRD 0081: attach_bottled_agents_to_gateway pushes the current gateway CA
|
||||
into every running agent VM over SSH."""
|
||||
|
||||
def _run_dir(self, root: Path, ip: str = "10.243.0.3") -> Path:
|
||||
run_dir = root / "demo"
|
||||
run_dir.mkdir()
|
||||
(run_dir / "bottle_id_ed25519").write_text("key")
|
||||
(run_dir / "config.json").write_text(json.dumps({
|
||||
"boot-source": {"boot_args": f"root=/dev/vda ip={ip}::gw:mask::eth0:off"}
|
||||
}))
|
||||
return run_dir
|
||||
|
||||
def test_pushes_current_ca_over_stdin_and_rebuilds_trust_store(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
run_dir = self._run_dir(Path(tmp))
|
||||
gw = Mock()
|
||||
gw.ca_cert_pem.return_value = "PEM-DATA"
|
||||
with patch.object(fc, "FirecrackerGateway", return_value=gw), \
|
||||
patch.object(fc.cleanup, "live_run_dirs", return_value=(run_dir,)), \
|
||||
patch.object(fc.util, "ssh_base_argv", return_value=["ssh", "guest"]), \
|
||||
patch.object(fc.subprocess, "run", return_value=_proc()) as run, \
|
||||
patch.object(fc, "info"):
|
||||
fc.attach_bottled_agents_to_gateway()
|
||||
# The cert goes over stdin (off argv), and the trust store is rebuilt.
|
||||
self.assertEqual("PEM-DATA", run.call_args.kwargs["input"])
|
||||
script = run.call_args.args[0][-1]
|
||||
self.assertIn(fc.AGENT_CA_PATH, script)
|
||||
self.assertIn("update-ca-certificates", script)
|
||||
|
||||
def test_unreachable_vm_is_skipped_not_fatal(self) -> None:
|
||||
with tempfile.TemporaryDirectory() as tmp:
|
||||
run_dir = self._run_dir(Path(tmp))
|
||||
gw = Mock()
|
||||
gw.ca_cert_pem.return_value = "PEM"
|
||||
with patch.object(fc, "FirecrackerGateway", return_value=gw), \
|
||||
patch.object(fc.cleanup, "live_run_dirs", return_value=(run_dir,)), \
|
||||
patch.object(fc.util, "ssh_base_argv", return_value=["ssh", "guest"]), \
|
||||
patch.object(fc.subprocess, "run", return_value=_proc(1, stderr="down")), \
|
||||
patch.object(fc, "info") as info:
|
||||
fc.attach_bottled_agents_to_gateway() # does not raise
|
||||
self.assertTrue(any("skipped" in c.args[0] for c in info.call_args_list))
|
||||
|
||||
def test_missing_gateway_ca_skips_the_whole_reconcile(self) -> None:
|
||||
gw = Mock()
|
||||
gw.ca_cert_pem.side_effect = fc.GatewayError("no CA yet")
|
||||
with patch.object(fc, "FirecrackerGateway", return_value=gw), \
|
||||
patch.object(fc.cleanup, "live_run_dirs") as live, \
|
||||
patch.object(fc, "info"):
|
||||
fc.attach_bottled_agents_to_gateway()
|
||||
live.assert_not_called() # bailed before enumerating agents
|
||||
|
||||
|
||||
class TestDockerCaReconcile(unittest.TestCase):
|
||||
def test_pushes_ca_to_each_agent_container_excluding_gateway(self) -> None:
|
||||
# network inspect lists the gateway (excluded) + one agent + a blank line.
|
||||
inspect = _proc(stdout="bot-bottle-orch-gateway\nbot-bottle-a\n\n")
|
||||
with patch.object(
|
||||
docker, "run_docker",
|
||||
side_effect=[inspect, _proc(), _proc(), _proc()],
|
||||
) as run, patch.object(docker.log, "info"):
|
||||
docker.attach_bottled_agents_to_gateway(
|
||||
"PEM", gateway_name="bot-bottle-orch-gateway")
|
||||
argvs = [c.args[0] for c in run.call_args_list]
|
||||
# First call enumerates; the agent then gets mkdir + cp + exec — never
|
||||
# the gateway container.
|
||||
self.assertNotIn(
|
||||
"bot-bottle-orch-gateway",
|
||||
" ".join(tok for a in argvs[1:] for tok in a),
|
||||
)
|
||||
cp = next(a for a in argvs if a[:2] == ["docker", "cp"])
|
||||
self.assertEqual(f"bot-bottle-a:{docker.AGENT_CA_PATH}", cp[-1])
|
||||
exec_argv = next(a for a in argvs if "update-ca-certificates" in a[-1])
|
||||
self.assertIn(docker.AGENT_CA_PATH, exec_argv[-1])
|
||||
|
||||
def test_one_container_failure_does_not_block_others(self) -> None:
|
||||
inspect = _proc(stdout="a\nb\n")
|
||||
# a: mkdir fails (skip); b: mkdir/cp/exec all succeed.
|
||||
with patch.object(
|
||||
docker, "run_docker",
|
||||
side_effect=[inspect, _proc(1, stderr="gone"), _proc(), _proc(), _proc()],
|
||||
), patch.object(docker.log, "info") as info:
|
||||
docker.attach_bottled_agents_to_gateway("PEM", gateway_name="gw")
|
||||
self.assertTrue(
|
||||
any("skipped for a" in str(c.args[0]) for c in info.call_args_list)
|
||||
)
|
||||
|
||||
def test_fetches_ca_from_gateway_when_not_supplied(self) -> None:
|
||||
svc = Mock()
|
||||
svc.gateway.return_value.ca_cert_pem.return_value = "FETCHED"
|
||||
with patch.object(docker, "DockerInfraService", return_value=svc), \
|
||||
patch.object(docker, "run_docker", return_value=_proc(stdout="")), \
|
||||
patch.object(docker.log, "info"):
|
||||
docker.attach_bottled_agents_to_gateway(network="net", gateway_name="gw")
|
||||
svc.gateway.return_value.ca_cert_pem.assert_called_once()
|
||||
|
||||
|
||||
class TestMacosCaReconcile(unittest.TestCase):
|
||||
def test_pushes_ca_to_each_running_agent_container(self) -> None:
|
||||
agent = SimpleNamespace(slug="demo")
|
||||
with patch.object(mac, "enumerate_active", return_value=[agent]), \
|
||||
patch.object(mac.container_mod, "run_container_argv",
|
||||
return_value=_proc()) as run, \
|
||||
patch.object(mac, "info"):
|
||||
mac.attach_bottled_agents_to_gateway("PEM")
|
||||
argvs = [c.args[0] for c in run.call_args_list]
|
||||
name = f"{mac.CONTAINER_NAME_PREFIX}demo"
|
||||
self.assertTrue(any(name in a for a in argvs))
|
||||
self.assertTrue(
|
||||
any("update-ca-certificates" in a[-1] for a in argvs if a[-2:-1] == ["-c"])
|
||||
)
|
||||
self.assertTrue(any(mac.AGENT_CA_PATH in " ".join(a) for a in argvs))
|
||||
|
||||
def test_enumeration_failure_is_best_effort(self) -> None:
|
||||
with patch.object(mac, "enumerate_active",
|
||||
side_effect=mac.EnumerationError("failed")), \
|
||||
patch.object(mac, "info") as info:
|
||||
mac.attach_bottled_agents_to_gateway("PEM") # does not raise
|
||||
self.assertIn("skipped", info.call_args.args[0])
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
|
||||
@@ -29,10 +29,20 @@ class TestDockerInfraService(unittest.TestCase):
|
||||
self.orch.gateway_url.return_value = f"http://{ORCHESTRATOR_NAME}:8099"
|
||||
self.orch.mint_gateway_token.return_value = "gw.jwt"
|
||||
self.gw = MagicMock()
|
||||
# Default to a cold boot (gateway (re)created) so the reconcile branch
|
||||
# runs; the actual reconcile is stubbed so these composition tests stay
|
||||
# isolated from docker.
|
||||
self.gw.connect_to_orchestrator.return_value = True
|
||||
for name, mock in (("orchestrator", self.orch), ("gateway", self.gw)):
|
||||
p = patch.object(self.svc, name, return_value=mock)
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
ap = patch(
|
||||
"bot_bottle.backend.docker.backend.DockerBottleBackend"
|
||||
".attach_bottled_agents_to_gateway"
|
||||
)
|
||||
self.attach = ap.start()
|
||||
self.addCleanup(ap.stop)
|
||||
|
||||
def test_url_delegates_to_the_orchestrator(self) -> None:
|
||||
self.assertEqual("http://127.0.0.1:8099", self.svc.url())
|
||||
@@ -58,6 +68,20 @@ class TestDockerInfraService(unittest.TestCase):
|
||||
f"http://{ORCHESTRATOR_NAME}:8099", "gw.jwt")
|
||||
self.orch.mint_gateway_token.assert_called_once()
|
||||
|
||||
def test_cold_boot_reconciles_running_bottles(self) -> None:
|
||||
# A (re)created gateway (connect returns True) triggers the bring-up
|
||||
# reconcile so running bottles re-trust the fresh gateway (PRD 0081).
|
||||
self.gw.connect_to_orchestrator.return_value = True
|
||||
self.svc.ensure_running()
|
||||
self.attach.assert_called_once_with()
|
||||
|
||||
def test_no_reconcile_when_gateway_left_untouched(self) -> None:
|
||||
# A healthy current gateway (connect returns False) is not a cold boot,
|
||||
# so there is nothing to reconcile.
|
||||
self.gw.connect_to_orchestrator.return_value = False
|
||||
self.svc.ensure_running()
|
||||
self.attach.assert_not_called()
|
||||
|
||||
def test_stop_removes_both_containers(self) -> None:
|
||||
with patch(_RUN) as run:
|
||||
self.svc.stop()
|
||||
|
||||
@@ -61,7 +61,10 @@ class TestFirecrackerGatewayConnect(unittest.TestCase):
|
||||
booted = infra_vm.InfraVm(guest_ip="10.243.255.3", private_key=Path("/k"))
|
||||
with patch.object(infra_vm, "boot_vm", return_value=booted) as boot, \
|
||||
patch.object(infra_vm, "push_secret") as push:
|
||||
gw.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
cold_booted = gw.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
# The VM is booted fresh here (cold-boot path only), so it always
|
||||
# signals a cold boot → the caller reconciles running bottles (PRD 0081).
|
||||
self.assertTrue(cold_booted)
|
||||
# Booted on the gateway link with the gateway role; the orchestrator's
|
||||
# guest IP (parsed off the URL) rides the cmdline as bb_orch.
|
||||
kw = boot.call_args.kwargs
|
||||
|
||||
@@ -40,7 +40,21 @@ class TestAccessors(unittest.TestCase):
|
||||
self.assertEqual(f"http://{ip}:{infra_vm.ORCHESTRATOR_PORT}", svc.url())
|
||||
|
||||
|
||||
_ATTACH = (
|
||||
"bot_bottle.backend.firecracker.backend.FirecrackerBottleBackend"
|
||||
".attach_bottled_agents_to_gateway"
|
||||
)
|
||||
|
||||
|
||||
class TestEnsureRunning(unittest.TestCase):
|
||||
def setUp(self) -> None:
|
||||
# The cold-boot branch reconciles running bottles against the fresh
|
||||
# gateway (PRD 0081). Stub it so these composition tests stay isolated
|
||||
# from SSH; the wiring itself is asserted in test_cold_boot_reconciles*.
|
||||
ap = patch(_ATTACH)
|
||||
self.attach = ap.start()
|
||||
self.addCleanup(ap.stop)
|
||||
|
||||
def test_adopts_when_healthy_alive_and_version_matches(self):
|
||||
# Healthy control plane + live gateway + existing key + matching version
|
||||
# marker -> adopt (no stop/build/boot); returns the orchestrator URL.
|
||||
@@ -59,6 +73,8 @@ class TestEnsureRunning(unittest.TestCase):
|
||||
built.assert_not_called()
|
||||
ip = infra_vm.netpool.orch_slot().guest_ip
|
||||
self.assertEqual(f"http://{ip}:{infra_vm.ORCHESTRATOR_PORT}", url)
|
||||
# Adopt path — no cold boot, so no running-bottle reconcile.
|
||||
self.attach.assert_not_called()
|
||||
|
||||
def test_reboots_both_when_version_stale(self):
|
||||
# Healthy control plane but the running pair booted an OLDER image
|
||||
@@ -120,6 +136,8 @@ class TestEnsureRunning(unittest.TestCase):
|
||||
# reach the control plane at startup).
|
||||
orch_cls.return_value.ensure_running.assert_called_once()
|
||||
gw_cls.return_value.connect_to_orchestrator.assert_called_once()
|
||||
# A cold boot minted a fresh gateway CA — reconcile running bottles.
|
||||
self.attach.assert_called_once_with()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
|
||||
@@ -44,6 +44,17 @@ class TestMacosGatewayConnect(unittest.TestCase):
|
||||
def test_default_name(self) -> None:
|
||||
self.assertEqual(GATEWAY_NAME, self.gw.name)
|
||||
|
||||
def test_connect_signals_cold_boot(self) -> None:
|
||||
# The container is recreated unconditionally, so connect always signals
|
||||
# a cold boot → the caller reconciles running bottles (PRD 0081).
|
||||
run = Mock(return_value=_proc())
|
||||
with patch(f"{_GW}.container_mod") as mod, \
|
||||
patch(f"{_GW}.host_gateway_ca_dir", return_value=Path("/ca")):
|
||||
mod.dns_server.return_value = "1.1.1.1"
|
||||
mod.bind_mount_spec.side_effect = _spec
|
||||
mod.run_container_argv = run
|
||||
self.assertTrue(self.gw.connect_to_orchestrator(_ORCH_URL, _TOKEN))
|
||||
|
||||
def test_refuses_without_orchestrator_url(self) -> None:
|
||||
with patch(f"{_GW}.container_mod") as mod:
|
||||
with self.assertRaises(GatewayError):
|
||||
|
||||
@@ -24,10 +24,20 @@ class TestInfraEnsureRunning(unittest.TestCase):
|
||||
self.orch.url.return_value = "http://192.168.128.2:8099"
|
||||
self.orch.mint_gateway_token.return_value = "gw.jwt"
|
||||
self.gw = MagicMock()
|
||||
# Default to a cold boot (gateway (re)created) so the reconcile branch
|
||||
# runs; the actual reconcile is stubbed so these composition tests stay
|
||||
# isolated from the `container` CLI.
|
||||
self.gw.connect_to_orchestrator.return_value = True
|
||||
for name, mock in (("orchestrator", self.orch), ("gateway", self.gw)):
|
||||
p = patch.object(self.svc, name, return_value=mock)
|
||||
p.start()
|
||||
self.addCleanup(p.stop)
|
||||
ap = patch(
|
||||
"bot_bottle.backend.macos_container.backend"
|
||||
".MacosContainerBottleBackend.attach_bottled_agents_to_gateway"
|
||||
)
|
||||
self.attach = ap.start()
|
||||
self.addCleanup(ap.stop)
|
||||
|
||||
def test_composes_networks_builds_and_brings_up_orchestrator_first(self) -> None:
|
||||
with patch(f"{_INFRA}.ensure_networks") as nets:
|
||||
@@ -46,6 +56,20 @@ class TestInfraEnsureRunning(unittest.TestCase):
|
||||
"http://192.168.128.2:8099", "gw.jwt")
|
||||
self.orch.mint_gateway_token.assert_called_once()
|
||||
|
||||
def test_cold_boot_reconciles_running_bottles(self) -> None:
|
||||
# A (re)created gateway (connect returns True) triggers the bring-up
|
||||
# reconcile so running bottles re-trust the fresh gateway (PRD 0081).
|
||||
self.gw.connect_to_orchestrator.return_value = True
|
||||
with patch(f"{_INFRA}.ensure_networks"):
|
||||
self.svc.ensure_running()
|
||||
self.attach.assert_called_once_with()
|
||||
|
||||
def test_no_reconcile_when_gateway_left_untouched(self) -> None:
|
||||
self.gw.connect_to_orchestrator.return_value = False
|
||||
with patch(f"{_INFRA}.ensure_networks"):
|
||||
self.svc.ensure_running()
|
||||
self.attach.assert_not_called()
|
||||
|
||||
def test_is_healthy_delegates_to_the_orchestrator(self) -> None:
|
||||
self.orch.is_healthy.return_value = True
|
||||
self.assertTrue(self.svc.is_healthy())
|
||||
|
||||
@@ -86,9 +86,12 @@ class TestDockerGateway(unittest.TestCase):
|
||||
return _proc()
|
||||
|
||||
with patch(_RUN_DOCKER, side_effect=fake):
|
||||
self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
cold_booted = self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
self.assertEqual([], [c for c in calls if c[:2] == ["docker", "run"]])
|
||||
self.assertEqual([], [c for c in calls if c[:2] == ["docker", "rm"]])
|
||||
# A healthy current gateway left untouched is not a cold boot — the
|
||||
# bring-up flow skips the running-bottle reconcile (PRD 0081).
|
||||
self.assertFalse(cold_booted)
|
||||
|
||||
def test_ensure_running_recreates_when_image_is_stale(self) -> None:
|
||||
# Running, but the container was built from an OLD image → recreate so
|
||||
@@ -106,9 +109,12 @@ class TestDockerGateway(unittest.TestCase):
|
||||
return _proc()
|
||||
|
||||
with patch(_RUN_DOCKER, side_effect=fake):
|
||||
self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
cold_booted = self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
self.assertEqual(1, len([c for c in calls if c[:2] == ["docker", "run"]]))
|
||||
self.assertTrue(any(c[:2] == ["docker", "rm"] for c in calls))
|
||||
# A recreated container minted/mounted its CA afresh — a cold boot that
|
||||
# triggers the running-bottle reconcile (PRD 0081).
|
||||
self.assertTrue(cold_booted)
|
||||
|
||||
def test_ensure_running_starts_the_singleton_when_absent(self) -> None:
|
||||
calls: list[list[str]] = []
|
||||
@@ -118,7 +124,8 @@ class TestDockerGateway(unittest.TestCase):
|
||||
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
|
||||
|
||||
with patch(_RUN_DOCKER, side_effect=fake):
|
||||
self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
cold_booted = self.sc.connect_to_orchestrator(_ORCH_URL, _TOKEN)
|
||||
self.assertTrue(cold_booted) # started fresh → reconcile running bottles
|
||||
|
||||
runs = [c for c in calls if c[:2] == ["docker", "run"]]
|
||||
self.assertEqual(1, len(runs))
|
||||
|
||||
Reference in New Issue
Block a user
We should enforce the flow more in here/only overwrite specific methods so things are more consistent. This is a larger theme we should have as an ADR somewhere... as much as possible, try to encode shared behavior in the base backend logic. We want to lean towards LESS freedom in the specific backend implementations, and MORE consistency.
Basic structure is I think uniform. Rough shape