feat(reconcile): reprovision CA, git-gate, and egress tokens on gateway bring-up (PRD 0081)
prd-number-check / require-numbered-prds (pull_request) Successful in 10s
tracker-policy-pr / check-pr (pull_request) Successful in 16s
lint / lint (push) Failing after 1m1s
test / unit (pull_request) Successful in 56s
test / image-input-builds (pull_request) Successful in 1m0s
test / integration-docker (pull_request) Failing after 3m7s
test / coverage (pull_request) Has been skipped

On a gateway cold boot, reconcile every live agent VM against the fresh
gateway: push the new CA into each agent's trust store, re-provision
git-gate repos/creds from the persisted upstreams snapshot, and restore
egress tokens. Per-bottle failures are logged and skipped rather than
aborting the whole reconcile.

New: reconcile.py — attach_bottled_agents_to_gateway, _push_ca,
_reprovision_git_gate, _guest_ip_from_config.

New: git_gate/provision.py writes upstreams.json after key provisioning
so the bring-up reconcile can reconstruct the upstream table without the
manifest.

Wired into FirecrackerInfraService.ensure_running() cold-boot path;
base.py BottleBackend gets a no-op default. Old _reprovision_running_bottles
removed from consolidated_launch.py (superseded by reconcile.py). Tests
migrated and extended.
This commit is contained in:
2026-07-28 00:28:26 +00:00
parent 314b30c013
commit d066e4032b
8 changed files with 560 additions and 85 deletions
+14
View File
@@ -496,6 +496,20 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
(macos-container) die with a pointer — the default here.""" (macos-container) die with a pointer — the default here."""
die(f"backend {self.name!r} has no orchestrator control plane") die(f"backend {self.name!r} has no orchestrator control plane")
def attach_bottled_agents_to_gateway(self) -> None:
"""Reconcile all running bottles against the current gateway.
Called when the gateway is (re)brought up (cold-boot path) to restore
the three gateway-dependent services for every live bottle: CA trust,
git-gate repos/creds, and egress tokens. Per-bottle failures must be
logged and skipped — one unreachable agent must not block the rest.
Default: no-op. Backends that run a gateway (Firecracker, docker,
macOS) override this with a backend-native implementation that reaches
running agents via their transport (SSH for Firecracker, exec/cp for
docker and macOS). PRD 0081."""
return
@abstractmethod @abstractmethod
def prepare_cleanup(self) -> CleanupT: def prepare_cleanup(self) -> CleanupT:
"""Enumerate orphaned resources from previous bottles. No side """Enumerate orphaned resources from previous bottles. No side
@@ -26,22 +26,19 @@ The TAP slot allocation, rootfs build, and VM boot are the caller's job.
from __future__ import annotations from __future__ import annotations
import json
import subprocess import subprocess
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from ...egress import EgressPlan from ...egress import EgressPlan
from ...git_gate import GitGatePlan from ...git_gate import GitGatePlan
from ...log import info from ...orchestrator.client import OrchestratorClient
from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
from ...orchestrator.lifecycle import ( from ...orchestrator.lifecycle import (
OrchestratorStartError, # re-exported so callers can catch it OrchestratorStartError, # re-exported so callers can catch it
) )
from ...orchestrator.reprovision import reprovision_bottles
from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
from ..provision_bottle import deprovision_bottle, provision_bottle from ..provision_bottle import deprovision_bottle, provision_bottle
from . import cleanup, util from . import util
from .gateway import FirecrackerGateway from .gateway import FirecrackerGateway
from .infra import FirecrackerInfraService from .infra import FirecrackerInfraService
@@ -64,17 +61,6 @@ class LaunchContext:
env_var_secret: str = "" # encryption key injected into the agent's env env_var_secret: str = "" # encryption key injected into the agent's env
def _guest_ip_from_config(config_path: Path) -> str:
"""Read the kernel's configured guest IP from a Firecracker config."""
try:
config = json.loads(config_path.read_text())
args = config["boot-source"]["boot_args"]
ip_arg = next(part for part in args.split() if part.startswith("ip="))
return ip_arg.removeprefix("ip=").split(":", 1)[0]
except (OSError, ValueError, KeyError, TypeError, StopIteration):
return ""
def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> None: def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> None:
"""Mirror the exec-time key into guest tmpfs for restart recovery.""" """Mirror the exec-time key into guest tmpfs for restart recovery."""
proc = subprocess.run( proc = subprocess.run(
@@ -89,29 +75,6 @@ def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> Non
) )
def _reprovision_running_bottles(client: OrchestratorClient) -> None:
"""Read keys from live agent VMs and restore the restarted gateway."""
try:
secrets_by_ip: dict[str, str] = {}
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
proc = subprocess.run(
util.ssh_base_argv(private_key, guest_ip)
+ [f"cat {_ENV_VAR_SECRET_PATH}"],
capture_output=True, text=True, check=False,
)
if proc.returncode == 0 and proc.stdout.strip():
secrets_by_ip[guest_ip] = proc.stdout.strip()
count = reprovision_bottles(client, secrets_by_ip)
if count:
info(f"reprovisioned egress tokens for {count} Firecracker bottle(s)")
except (OSError, OrchestratorClientError) as exc:
info(f"egress token reprovision skipped: {exc}")
def launch_consolidated( def launch_consolidated(
egress_plan: EgressPlan, egress_plan: EgressPlan,
git_gate_plan: GitGatePlan, git_gate_plan: GitGatePlan,
@@ -126,7 +89,6 @@ def launch_consolidated(
service = FirecrackerInfraService() service = FirecrackerInfraService()
url = service.ensure_running() url = service.ensure_running()
client = OrchestratorClient(url) client = OrchestratorClient(url)
_reprovision_running_bottles(client)
# Read the gateway's provisioning transport + CA off the Gateway service. # Read the gateway's provisioning transport + CA off the Gateway service.
gateway = service.gateway() gateway = service.gateway()
+4 -1
View File
@@ -17,6 +17,7 @@ from ...orchestrator.lifecycle import DEFAULT_STARTUP_TIMEOUT_SECONDS
from . import infra_vm from . import infra_vm
from .gateway import FirecrackerGateway from .gateway import FirecrackerGateway
from .orchestrator import FirecrackerOrchestrator from .orchestrator import FirecrackerOrchestrator
from .reconcile import attach_bottled_agents_to_gateway
class FirecrackerInfraService(InfraService): class FirecrackerInfraService(InfraService):
@@ -67,9 +68,11 @@ class FirecrackerInfraService(InfraService):
# holds the signing key) mints the role-scoped `gateway` JWT for the # holds the signing key) mints the role-scoped `gateway` JWT for the
# gateway, which never sees the key (#469). # gateway, which never sees the key (#469).
orchestrator.ensure_running(startup_timeout=startup_timeout) orchestrator.ensure_running(startup_timeout=startup_timeout)
self.gateway().connect_to_orchestrator( gateway = self.gateway()
gateway.connect_to_orchestrator(
orchestrator.gateway_url(), orchestrator.mint_gateway_token()) orchestrator.gateway_url(), orchestrator.mint_gateway_token())
infra_vm.record_booted_version(want) infra_vm.record_booted_version(want)
attach_bottled_agents_to_gateway(url, gateway)
return url return url
def stop(self) -> None: def stop(self) -> None:
+198
View File
@@ -0,0 +1,198 @@
"""Bring-up reconcile: re-attach all running agent VMs to a freshly-booted gateway.
Called from the cold-boot branch of `FirecrackerInfraService.ensure_running()`
after `gateway.connect_to_orchestrator()` completes. Restores the three
gateway-dependent services for every live bottle:
- **CA** — push the current (freshly-minted) gateway CA to each agent's
trust store and run `update-ca-certificates`, so the agent trusts the
new CA on its next egress call.
- **git-gate** — re-provision each bottle's bare repos and per-repo creds
in the gateway, rebuilt from the persisted host-side state dir.
- **egress tokens** — read the agent's `ENV_VAR_SECRET` and feed
`reprovision_bottles` to restore the orchestrator's in-memory tokens.
Per-bottle failures are caught, logged, and skipped so one unreachable VM
does not block the others or the gateway coming up (PRD 0081).
"""
from __future__ import annotations
import json
import subprocess
from pathlib import Path
from ...bottle_state import git_gate_state_dir
from ...gateway import GatewayError
from ...gateway.git_gate.render import GitGateUpstream
from ...git_gate.plan import GitGatePlan
from ...log import info
from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
from ...orchestrator.reprovision import reprovision_bottles
from ..provision_gateway import GatewayProvisionError, provision_git_gate
from ..util import AGENT_CA_PATH
from . import cleanup, util
from .gateway import FirecrackerGateway
# Where persist_env_var_secret writes the key on the agent VM (matches
# consolidated_launch._ENV_VAR_SECRET_PATH — duplicated to avoid a
# circular import through infra.py).
_ENV_VAR_SECRET_PATH = "/run/bot-bottle/env-var-secret"
def _guest_ip_from_config(config_path: Path) -> str:
"""Read the agent VM's guest IP from its Firecracker config file."""
try:
config = json.loads(config_path.read_text())
args = config["boot-source"]["boot_args"]
ip_arg = next(p for p in args.split() if p.startswith("ip="))
return ip_arg.removeprefix("ip=").split(":", 1)[0]
except (OSError, ValueError, KeyError, TypeError, StopIteration):
return ""
def attach_bottled_agents_to_gateway(
orchestrator_url: str, gateway: FirecrackerGateway,
) -> None:
"""Reconcile all live agent VMs against the freshly-booted gateway.
Fetches the new CA, lists registered bottles, then for each live run dir
pushes the CA, re-provisions git-gate, and restores egress tokens. Each
per-bottle step is wrapped so a failure is logged and skipped."""
try:
ca_pem = gateway.ca_cert_pem()
except GatewayError as exc:
info(f"bring-up reconcile: could not fetch gateway CA, skipping: {exc}")
return
client = OrchestratorClient(orchestrator_url)
try:
bottles = client.list_bottles()
except OrchestratorClientError as exc:
info(f"bring-up reconcile: could not list bottles, skipping: {exc}")
return
source_ip_to_bottle_id: dict[str, str] = {
b["source_ip"]: b["bottle_id"]
for b in bottles
if isinstance(b.get("source_ip"), str) and isinstance(b.get("bottle_id"), str)
}
transport = gateway.provisioning_transport()
secrets_by_ip: dict[str, str] = {}
for run_dir in cleanup.live_run_dirs():
slug = run_dir.name
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
# CA: push the gateway's current certificate to the agent's trust store.
try:
_push_ca(private_key, guest_ip, ca_pem)
except Exception as exc:
info(f"bring-up reconcile: CA push to {slug!r} failed: {exc}")
# git-gate: re-provision repos + creds from the persisted state dir.
bottle_id = source_ip_to_bottle_id.get(guest_ip)
if bottle_id:
try:
_reprovision_git_gate(transport, bottle_id, slug)
except Exception as exc:
info(f"bring-up reconcile: git-gate for {slug!r} failed: {exc}")
# egress tokens: read ENV_VAR_SECRET from the agent's tmpfs.
proc = subprocess.run(
util.ssh_base_argv(private_key, guest_ip) + [f"cat {_ENV_VAR_SECRET_PATH}"],
capture_output=True, text=True, check=False,
)
if proc.returncode == 0 and proc.stdout.strip():
secrets_by_ip[guest_ip] = proc.stdout.strip()
# Restore in-memory egress tokens for all bottles that exposed a key.
if secrets_by_ip:
try:
count = reprovision_bottles(client, secrets_by_ip)
if count:
info(
f"bring-up reconcile: restored egress tokens for {count} bottle(s)"
)
except OrchestratorClientError as exc:
info(f"bring-up reconcile: egress token restore failed: {exc}")
def _push_ca(private_key: Path, guest_ip: str, ca_pem: str) -> None:
"""SSH the gateway CA PEM into the agent VM and update its trust store.
Runs as the SSH root user (the agent VM's dropbear accepts root). Each
step uses check=True so a failure raises and the caller's except clause
logs and continues."""
ssh = util.ssh_base_argv(private_key, guest_ip)
# mkdir is idempotent; rootfs builds may not preserve the target dir.
subprocess.run(
ssh + ["mkdir -p /usr/local/share/ca-certificates"],
capture_output=True, check=True,
)
subprocess.run(
ssh + [f"cat > {AGENT_CA_PATH}"],
input=ca_pem, text=True, capture_output=True, check=True,
)
subprocess.run(
ssh + [f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates"],
capture_output=True, check=True,
)
def _reprovision_git_gate(
transport: object, bottle_id: str, slug: str,
) -> None:
"""Re-provision one bottle's git-gate repos and creds from its state dir.
Reads `upstreams.json` (written by `provision_git_gate_dynamic_keys` at
launch) to reconstruct the GitGateUpstream table, then calls
`provision_git_gate` which places the credential files and inits the bare
repos in the gateway. No-op when there is no `upstreams.json` (the bottle
has no git upstreams)."""
state_dir = git_gate_state_dir(slug)
upstreams_file = state_dir / "upstreams.json"
if not upstreams_file.exists():
return
raw = json.loads(upstreams_file.read_text())
upstreams: list[GitGateUpstream] = []
for u in raw:
name = u["name"]
# The key in the state dir is preferred: gitea deploy keys are written
# there by provision_git_gate_dynamic_keys; static keys keep their
# original manifest path.
key_in_state = state_dir / f"{name}-key"
identity_file = (
str(key_in_state) if key_in_state.is_file()
else u.get("identity_file", "")
)
known_hosts = state_dir / f"{name}-known_hosts"
upstreams.append(GitGateUpstream(
name=name,
upstream_url=u["upstream_url"],
upstream_host=u.get("upstream_host", ""),
upstream_port=u.get("upstream_port", ""),
identity_file=identity_file,
known_host_key=u.get("known_host_key", ""),
known_hosts_file=known_hosts if known_hosts.is_file() else Path(),
))
if not upstreams:
return
plan = GitGatePlan(
slug=slug,
entrypoint_script=state_dir / "git_gate_entrypoint.sh",
hook_script=state_dir / "git_gate_pre_receive.sh",
access_hook_script=state_dir / "git_gate_access_hook.sh",
upstreams=tuple(upstreams),
)
provision_git_gate(transport, bottle_id, plan) # type: ignore[arg-type]
__all__ = ["attach_bottled_agents_to_gateway"]
+27 -1
View File
@@ -8,6 +8,7 @@ imported (`deploy_key_provisioner`) to keep its cost off the host path.
from __future__ import annotations from __future__ import annotations
import json
import os import os
import dataclasses import dataclasses
from pathlib import Path from pathlib import Path
@@ -138,7 +139,32 @@ def provision_git_gate_dynamic_keys(
if upstream.name not in updated_names: if upstream.name not in updated_names:
updated.append(upstream) updated.append(upstream)
return dataclasses.replace(plan, upstreams=tuple(updated)) final_plan = dataclasses.replace(plan, upstreams=tuple(updated))
_write_upstreams_snapshot(stage_dir, final_plan.upstreams)
return final_plan
def _write_upstreams_snapshot(
stage_dir: Path, upstreams: tuple[GitGateUpstream, ...]
) -> None:
"""Persist the fully-resolved upstream table to `upstreams.json` in
`stage_dir` so the bring-up reconcile can re-provision git-gate without
the manifest. Written after dynamic keys are provisioned so every
identity_file is set."""
data = [
{
"name": u.name,
"upstream_url": u.upstream_url,
"upstream_host": u.upstream_host,
"upstream_port": u.upstream_port,
"identity_file": u.identity_file,
"known_host_key": u.known_host_key,
}
for u in upstreams
]
snapshot = stage_dir / "upstreams.json"
snapshot.write_text(json.dumps(data, indent=2))
snapshot.chmod(0o600)
__all__ = [ __all__ = [
@@ -2,7 +2,6 @@
from __future__ import annotations from __future__ import annotations
import json
import subprocess import subprocess
import tempfile import tempfile
import unittest import unittest
@@ -107,21 +106,6 @@ class TestDockerReprovision(unittest.TestCase):
class TestFirecrackerReprovision(unittest.TestCase): class TestFirecrackerReprovision(unittest.TestCase):
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_extracts_guest_ip_from_config(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
run_dir = self._run_dir(Path(tmp))
self.assertEqual("10.243.0.3", fc._guest_ip_from_config(run_dir / "config.json"))
self.assertEqual("", fc._guest_ip_from_config(run_dir / "missing.json"))
def test_persists_key_over_stdin_not_argv(self) -> None: def test_persists_key_over_stdin_not_argv(self) -> None:
with patch.object(fc.util, "ssh_base_argv", return_value=["ssh", "guest"]), \ with patch.object(fc.util, "ssh_base_argv", return_value=["ssh", "guest"]), \
patch.object(fc.subprocess, "run", return_value=_proc()) as run: patch.object(fc.subprocess, "run", return_value=_proc()) as run:
@@ -135,29 +119,6 @@ class TestFirecrackerReprovision(unittest.TestCase):
with self.assertRaisesRegex(fc.ConsolidatedLaunchError, "denied"): with self.assertRaisesRegex(fc.ConsolidatedLaunchError, "denied"):
fc.persist_env_var_secret(Path("/key"), "10.0.0.1", "secret") fc.persist_env_var_secret(Path("/key"), "10.0.0.1", "secret")
def test_reads_live_vm_key_and_reprovisions(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
run_dir = self._run_dir(Path(tmp))
client = Mock()
with 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(stdout="secret\n")), \
patch.object(fc, "reprovision_bottles", return_value=1) as restore, \
patch.object(fc, "info"):
fc._reprovision_running_bottles(client)
restore.assert_called_once_with(client, {"10.243.0.3": "secret"})
def test_unreadable_vm_is_skipped(self) -> None:
with tempfile.TemporaryDirectory() as tmp:
run_dir = self._run_dir(Path(tmp))
client = Mock()
with 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)), \
patch.object(fc, "reprovision_bottles", return_value=0) as restore:
fc._reprovision_running_bottles(client)
restore.assert_called_once_with(client, {})
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+35 -4
View File
@@ -21,6 +21,7 @@ from bot_bottle.backend.firecracker.orchestrator import FirecrackerOrchestrator
# there (not at their home modules). # there (not at their home modules).
_ORCH_CLS = "bot_bottle.backend.firecracker.infra.FirecrackerOrchestrator" _ORCH_CLS = "bot_bottle.backend.firecracker.infra.FirecrackerOrchestrator"
_GW_CLS = "bot_bottle.backend.firecracker.infra.FirecrackerGateway" _GW_CLS = "bot_bottle.backend.firecracker.infra.FirecrackerGateway"
_RECONCILE = "bot_bottle.backend.firecracker.infra.attach_bottled_agents_to_gateway"
class TestAccessors(unittest.TestCase): class TestAccessors(unittest.TestCase):
@@ -53,10 +54,13 @@ class TestEnsureRunning(unittest.TestCase):
patch.object(infra_vm, "_health_ok", return_value=True), \ patch.object(infra_vm, "_health_ok", return_value=True), \
patch.object(infra_vm, "_pidfile_alive", return_value=True), \ patch.object(infra_vm, "_pidfile_alive", return_value=True), \
patch.object(infra_vm, "stop") as stop, \ patch.object(infra_vm, "stop") as stop, \
patch.object(infra_vm, "ensure_built") as built: patch.object(infra_vm, "ensure_built") as built, \
patch(_RECONCILE) as reconcile:
url = FirecrackerInfraService().ensure_running() url = FirecrackerInfraService().ensure_running()
stop.assert_not_called() stop.assert_not_called()
built.assert_not_called() built.assert_not_called()
# Adopt path: no reconcile — gateway state is intact.
reconcile.assert_not_called()
ip = infra_vm.netpool.orch_slot().guest_ip ip = infra_vm.netpool.orch_slot().guest_ip
self.assertEqual(f"http://{ip}:{infra_vm.ORCHESTRATOR_PORT}", url) self.assertEqual(f"http://{ip}:{infra_vm.ORCHESTRATOR_PORT}", url)
@@ -73,7 +77,8 @@ class TestEnsureRunning(unittest.TestCase):
patch.object(infra_vm, "_pidfile_alive", return_value=True), \ patch.object(infra_vm, "_pidfile_alive", return_value=True), \
patch.object(infra_vm, "stop") as stop, \ patch.object(infra_vm, "stop") as stop, \
patch.object(infra_vm, "ensure_built"), \ patch.object(infra_vm, "ensure_built"), \
patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls: patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls, \
patch(_RECONCILE) as reconcile:
orch_cls.return_value.gateway_url.return_value = "http://10.243.255.1:8099" orch_cls.return_value.gateway_url.return_value = "http://10.243.255.1:8099"
orch_cls.return_value.mint_gateway_token.return_value = "gw.jwt" orch_cls.return_value.mint_gateway_token.return_value = "gw.jwt"
FirecrackerInfraService().ensure_running() FirecrackerInfraService().ensure_running()
@@ -85,6 +90,8 @@ class TestEnsureRunning(unittest.TestCase):
"http://10.243.255.1:8099", "gw.jwt") "http://10.243.255.1:8099", "gw.jwt")
# The fresh boot records the current version for the next launcher. # The fresh boot records the current version for the next launcher.
self.assertEqual("v-current\n", (d / "booted-version").read_text()) self.assertEqual("v-current\n", (d / "booted-version").read_text())
# Cold boot: reconcile fires after gateway is up.
reconcile.assert_called_once()
def test_reboots_when_gateway_dead(self): def test_reboots_when_gateway_dead(self):
# Orchestrator healthy + marker current, but the gateway VM is gone -> # Orchestrator healthy + marker current, but the gateway VM is gone ->
@@ -99,11 +106,13 @@ class TestEnsureRunning(unittest.TestCase):
patch.object(infra_vm, "_pidfile_alive", return_value=False), \ patch.object(infra_vm, "_pidfile_alive", return_value=False), \
patch.object(infra_vm, "stop") as stop, \ patch.object(infra_vm, "stop") as stop, \
patch.object(infra_vm, "ensure_built"), \ patch.object(infra_vm, "ensure_built"), \
patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls: patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls, \
patch(_RECONCILE) as reconcile:
FirecrackerInfraService().ensure_running() FirecrackerInfraService().ensure_running()
stop.assert_called_once() stop.assert_called_once()
orch_cls.return_value.ensure_running.assert_called_once() orch_cls.return_value.ensure_running.assert_called_once()
gw_cls.return_value.connect_to_orchestrator.assert_called_once() gw_cls.return_value.connect_to_orchestrator.assert_called_once()
reconcile.assert_called_once()
def test_boots_both_when_no_running_pair(self): def test_boots_both_when_no_running_pair(self):
with tempfile.TemporaryDirectory() as td: with tempfile.TemporaryDirectory() as td:
@@ -112,7 +121,8 @@ class TestEnsureRunning(unittest.TestCase):
patch.object(infra_vm, "_health_ok", return_value=False), \ patch.object(infra_vm, "_health_ok", return_value=False), \
patch.object(infra_vm, "stop") as stop, \ patch.object(infra_vm, "stop") as stop, \
patch.object(infra_vm, "ensure_built") as built, \ patch.object(infra_vm, "ensure_built") as built, \
patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls: patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls, \
patch(_RECONCILE) as reconcile:
FirecrackerInfraService().ensure_running() FirecrackerInfraService().ensure_running()
stop.assert_called_once() # clear stale VMs first stop.assert_called_once() # clear stale VMs first
built.assert_called_once() built.assert_called_once()
@@ -120,6 +130,27 @@ class TestEnsureRunning(unittest.TestCase):
# reach the control plane at startup). # reach the control plane at startup).
orch_cls.return_value.ensure_running.assert_called_once() orch_cls.return_value.ensure_running.assert_called_once()
gw_cls.return_value.connect_to_orchestrator.assert_called_once() gw_cls.return_value.connect_to_orchestrator.assert_called_once()
# Cold boot: reconcile fires to restore CA, git-gate, and egress tokens.
reconcile.assert_called_once()
def test_reconcile_receives_url_and_fresh_gateway(self):
# reconcile gets the orchestrator URL and the gateway service (not the
# class). Verify the args to ensure it can reach the right endpoints.
with tempfile.TemporaryDirectory() as td:
with patch.object(infra_vm, "_infra_dir", return_value=Path(td)), \
patch.object(infra_vm, "expected_version", return_value="v-current"), \
patch.object(infra_vm, "_health_ok", return_value=False), \
patch.object(infra_vm, "stop"), \
patch.object(infra_vm, "ensure_built"), \
patch(_ORCH_CLS) as orch_cls, patch(_GW_CLS) as gw_cls, \
patch(_RECONCILE) as reconcile:
orch_cls.return_value.url.return_value = "http://10.243.255.1:8099"
FirecrackerInfraService().ensure_running()
call_args = reconcile.call_args
# First arg: the orchestrator's host URL.
self.assertEqual("http://10.243.255.1:8099", call_args.args[0])
# Second arg: the gateway service instance (not the class).
self.assertIs(gw_cls.return_value, call_args.args[1])
if __name__ == "__main__": if __name__ == "__main__":
+280
View File
@@ -0,0 +1,280 @@
"""Unit: bring-up reconcile — CA push, git-gate reprovision, egress tokens.
Covers attach_bottled_agents_to_gateway (reconcile.py): each per-bottle step
fires with correct args, failures on one bottle don't block others, and the
egress-token restore path works end-to-end. Does NOT spin up VMs or real SSH.
"""
from __future__ import annotations
import json
import tempfile
import unittest
from pathlib import Path
from subprocess import CalledProcessError, CompletedProcess
from unittest.mock import MagicMock, call, patch
_RECONCILE = "bot_bottle.backend.firecracker.reconcile"
class TestGuestIpFromConfig(unittest.TestCase):
def test_reads_ip_from_boot_args(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _guest_ip_from_config
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
json.dump({
"boot-source": {"boot_args": "console=ttyS0 ip=10.243.0.5:10.243.0.6:255.255.255.254"},
}, f)
name = f.name
self.assertEqual("10.243.0.5", _guest_ip_from_config(Path(name)))
def test_returns_empty_on_missing_file(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _guest_ip_from_config
self.assertEqual("", _guest_ip_from_config(Path("/nonexistent/config.json")))
def test_returns_empty_on_bad_json(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _guest_ip_from_config
with tempfile.NamedTemporaryFile(mode="w", suffix=".json", delete=False) as f:
f.write("not-json")
name = f.name
self.assertEqual("", _guest_ip_from_config(Path(name)))
class TestPushCa(unittest.TestCase):
def test_runs_three_ssh_commands_in_order(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _push_ca
with tempfile.TemporaryDirectory() as td:
key = Path(td) / "key"
key.write_text("k")
with patch(f"{_RECONCILE}.subprocess.run") as run:
run.return_value = CompletedProcess([], 0)
_push_ca(key, "10.0.0.1", "-----BEGIN CERTIFICATE-----\n")
calls = run.call_args_list
self.assertEqual(3, len(calls))
# Each call's argv is a list; the remote command is the last element.
self.assertIn("mkdir", calls[0].args[0][-1])
self.assertIn("cat >", calls[1].args[0][-1])
self.assertIn("update-ca-certificates", calls[2].args[0][-1])
def test_passes_ca_pem_via_stdin(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _push_ca
with tempfile.TemporaryDirectory() as td:
key = Path(td) / "key"
key.write_text("k")
with patch(f"{_RECONCILE}.subprocess.run") as run:
run.return_value = CompletedProcess([], 0)
_push_ca(key, "10.0.0.1", "MY-CA-PEM")
cat_call = run.call_args_list[1]
self.assertEqual("MY-CA-PEM", cat_call.kwargs.get("input"))
def test_raises_on_ssh_failure(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _push_ca
with tempfile.TemporaryDirectory() as td:
key = Path(td) / "key"
key.write_text("k")
with patch(f"{_RECONCILE}.subprocess.run") as run:
run.side_effect = CalledProcessError(1, "ssh")
with self.assertRaises(CalledProcessError):
_push_ca(key, "10.0.0.1", "pem")
class TestReprovisionGitGate(unittest.TestCase):
def _make_state_dir(self, upstreams: list[dict]) -> Path:
d = Path(tempfile.mkdtemp())
(d / "upstreams.json").write_text(json.dumps(upstreams))
(d / "git_gate_pre_receive.sh").write_text("#!/bin/sh")
(d / "git_gate_access_hook.sh").write_text("#!/bin/sh")
(d / "git_gate_entrypoint.sh").write_text("#!/bin/sh")
return d
def test_no_op_when_no_upstreams_json(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _reprovision_git_gate
transport = MagicMock()
with tempfile.TemporaryDirectory() as td:
# The state dir exists but has no upstreams.json
with patch(f"{_RECONCILE}.git_gate_state_dir", return_value=Path(td)):
_reprovision_git_gate(transport, "bottle-123", "my-slug")
transport.exec.assert_not_called()
def test_calls_provision_git_gate_with_upstreams(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _reprovision_git_gate
transport = MagicMock()
upstreams = [{
"name": "bot-bottle",
"upstream_url": "ssh://git@gitea.example/org/bot-bottle.git",
"upstream_host": "gitea.example",
"upstream_port": "22",
"identity_file": "/home/node/.ssh/id_ed25519",
"known_host_key": "ssh-ed25519 AAAA...",
}]
state_dir = self._make_state_dir(upstreams)
# Write the key file so identity_file falls back to state dir path.
(state_dir / "bot-bottle-key").write_bytes(b"PRIVATE")
try:
with patch(f"{_RECONCILE}.git_gate_state_dir", return_value=state_dir), \
patch(f"{_RECONCILE}.provision_git_gate") as prov:
_reprovision_git_gate(transport, "bottle-abc", "my-slug")
prov.assert_called_once()
_, bottle_id, plan = prov.call_args.args
self.assertEqual("bottle-abc", bottle_id)
self.assertEqual(1, len(plan.upstreams))
self.assertEqual("bot-bottle", plan.upstreams[0].name)
# Key in state dir takes priority over identity_file in JSON.
self.assertEqual(str(state_dir / "bot-bottle-key"), plan.upstreams[0].identity_file)
finally:
import shutil; shutil.rmtree(state_dir, ignore_errors=True)
def test_falls_back_to_manifest_identity_file_when_no_key_in_state(self) -> None:
from bot_bottle.backend.firecracker.reconcile import _reprovision_git_gate
transport = MagicMock()
upstreams = [{
"name": "repo",
"upstream_url": "ssh://git@github.com/org/repo.git",
"upstream_host": "github.com",
"upstream_port": "22",
"identity_file": "/home/node/.ssh/id_ed25519",
"known_host_key": "",
}]
state_dir = self._make_state_dir(upstreams)
# No `repo-key` in state dir — static manifest path should be used.
try:
with patch(f"{_RECONCILE}.git_gate_state_dir", return_value=state_dir), \
patch(f"{_RECONCILE}.provision_git_gate") as prov:
_reprovision_git_gate(transport, "bottle-xyz", "my-slug")
prov.assert_called_once()
_, _, plan = prov.call_args.args
self.assertEqual("/home/node/.ssh/id_ed25519", plan.upstreams[0].identity_file)
finally:
import shutil; shutil.rmtree(state_dir, ignore_errors=True)
class TestAttachBottledAgentsToGateway(unittest.TestCase):
"""Integration-level: the full reconcile loop against mocked SSH + gateway."""
def _make_run_dir(self, tmp: Path, slug: str, guest_ip: str) -> Path:
run_dir = tmp / slug
run_dir.mkdir()
key = run_dir / "bottle_id_ed25519"
key.write_text("PRIVATE")
cfg = {
"boot-source": {
"boot_args": f"console=ttyS0 ip={guest_ip}:{guest_ip}:255.255.255.254",
}
}
(run_dir / "config.json").write_text(json.dumps(cfg))
return run_dir
def _make_gateway(self, ca_pem: str = "-----BEGIN CERTIFICATE-----\n") -> MagicMock:
gw = MagicMock()
gw.ca_cert_pem.return_value = ca_pem
gw.provisioning_transport.return_value = MagicMock()
return gw
def test_skips_all_when_ca_fetch_fails(self) -> None:
from bot_bottle.backend.firecracker.gateway import FirecrackerGateway
from bot_bottle.backend.firecracker.reconcile import attach_bottled_agents_to_gateway
from bot_bottle.gateway import GatewayError
gw = MagicMock(spec=FirecrackerGateway)
gw.ca_cert_pem.side_effect = GatewayError("timeout")
with patch(f"{_RECONCILE}.cleanup.live_run_dirs", return_value=()):
attach_bottled_agents_to_gateway("http://orch:8099", gw)
# Nothing else is called when CA fetch fails.
gw.provisioning_transport.assert_not_called()
def test_ca_push_called_per_live_bottle(self) -> None:
from bot_bottle.backend.firecracker.reconcile import attach_bottled_agents_to_gateway
gw = self._make_gateway()
with tempfile.TemporaryDirectory() as td:
tmp = Path(td)
rd1 = self._make_run_dir(tmp, "agent-abc12", "10.243.0.5")
rd2 = self._make_run_dir(tmp, "agent-xyz99", "10.243.0.7")
with patch(f"{_RECONCILE}.cleanup.live_run_dirs", return_value=(rd1, rd2)), \
patch(f"{_RECONCILE}.OrchestratorClient") as client_cls, \
patch(f"{_RECONCILE}._push_ca") as push_ca, \
patch(f"{_RECONCILE}._reprovision_git_gate"), \
patch(f"{_RECONCILE}.subprocess.run",
return_value=CompletedProcess([], 1)):
client_cls.return_value.list_bottles.return_value = []
attach_bottled_agents_to_gateway("http://orch:8099", gw)
# CA pushed to both bottles.
self.assertEqual(2, push_ca.call_count)
def test_per_bottle_ca_failure_does_not_block_others(self) -> None:
from bot_bottle.backend.firecracker.reconcile import attach_bottled_agents_to_gateway
gw = self._make_gateway()
with tempfile.TemporaryDirectory() as td:
tmp = Path(td)
rd1 = self._make_run_dir(tmp, "agent-fail1", "10.243.0.5")
rd2 = self._make_run_dir(tmp, "agent-ok99", "10.243.0.7")
with patch(f"{_RECONCILE}.cleanup.live_run_dirs", return_value=(rd1, rd2)), \
patch(f"{_RECONCILE}.OrchestratorClient") as client_cls, \
patch(f"{_RECONCILE}._push_ca",
side_effect=[CalledProcessError(1, "ssh"), None]) as push_ca, \
patch(f"{_RECONCILE}._reprovision_git_gate"), \
patch(f"{_RECONCILE}.subprocess.run",
return_value=CompletedProcess([], 1)):
client_cls.return_value.list_bottles.return_value = []
# Must not raise even though the first bottle's CA push fails.
attach_bottled_agents_to_gateway("http://orch:8099", gw)
# Both bottles were attempted.
self.assertEqual(2, push_ca.call_count)
def test_egress_token_restored_for_bottles_with_secret(self) -> None:
from bot_bottle.backend.firecracker.reconcile import attach_bottled_agents_to_gateway
gw = self._make_gateway()
with tempfile.TemporaryDirectory() as td:
tmp = Path(td)
rd = self._make_run_dir(tmp, "agent-tok11", "10.243.0.5")
bottles = [{"bottle_id": "bid-001", "source_ip": "10.243.0.5"}]
with patch(f"{_RECONCILE}.cleanup.live_run_dirs", return_value=(rd,)), \
patch(f"{_RECONCILE}.OrchestratorClient") as client_cls, \
patch(f"{_RECONCILE}._push_ca"), \
patch(f"{_RECONCILE}._reprovision_git_gate"), \
patch(f"{_RECONCILE}.reprovision_bottles") as reprov, \
patch(f"{_RECONCILE}.subprocess.run",
return_value=CompletedProcess([], 0, stdout="secret-key\n")):
client_cls.return_value.list_bottles.return_value = bottles
reprov.return_value = 1
attach_bottled_agents_to_gateway("http://orch:8099", gw)
reprov.assert_called_once()
_, secrets = reprov.call_args.args
self.assertIn("10.243.0.5", secrets)
self.assertEqual("secret-key", secrets["10.243.0.5"])
def test_git_gate_reprovision_called_with_bottle_id(self) -> None:
from bot_bottle.backend.firecracker.reconcile import attach_bottled_agents_to_gateway
gw = self._make_gateway()
with tempfile.TemporaryDirectory() as td:
tmp = Path(td)
rd = self._make_run_dir(tmp, "agent-git11", "10.243.0.5")
bottles = [{"bottle_id": "bid-git", "source_ip": "10.243.0.5"}]
with patch(f"{_RECONCILE}.cleanup.live_run_dirs", return_value=(rd,)), \
patch(f"{_RECONCILE}.OrchestratorClient") as client_cls, \
patch(f"{_RECONCILE}._push_ca"), \
patch(f"{_RECONCILE}._reprovision_git_gate") as reprov_gw, \
patch(f"{_RECONCILE}.subprocess.run",
return_value=CompletedProcess([], 1)):
client_cls.return_value.list_bottles.return_value = bottles
attach_bottled_agents_to_gateway("http://orch:8099", gw)
reprov_gw.assert_called_once()
transport_arg, bottle_id_arg, slug_arg = reprov_gw.call_args.args
self.assertEqual("bid-git", bottle_id_arg)
self.assertEqual("agent-git11", slug_arg)
def test_run_dir_without_key_file_is_skipped(self) -> None:
from bot_bottle.backend.firecracker.reconcile import attach_bottled_agents_to_gateway
gw = self._make_gateway()
with tempfile.TemporaryDirectory() as td:
tmp = Path(td)
rd = self._make_run_dir(tmp, "agent-nokey", "10.243.0.5")
(rd / "bottle_id_ed25519").unlink() # remove the key
with patch(f"{_RECONCILE}.cleanup.live_run_dirs", return_value=(rd,)), \
patch(f"{_RECONCILE}.OrchestratorClient") as client_cls, \
patch(f"{_RECONCILE}._push_ca") as push_ca, \
patch(f"{_RECONCILE}._reprovision_git_gate"):
client_cls.return_value.list_bottles.return_value = []
attach_bottled_agents_to_gateway("http://orch:8099", gw)
push_ca.assert_not_called()
if __name__ == "__main__":
unittest.main()