Compare commits

...

18 Commits

Author SHA1 Message Date
didericis-claude b25cd72fc3 test(backend): cover poll_ca_cert timeout paths for diff-coverage gate
test / integration-docker (pull_request) Successful in 12s
tracker-policy-pr / check-pr (pull_request) Successful in 10s
test / unit (pull_request) Successful in 35s
test / stage-firecracker-inputs (pull_request) Successful in 2s
test / build-infra (pull_request) Successful in 3m37s
test / integration-firecracker (pull_request) Successful in 1m30s
test / coverage (pull_request) Successful in 1m32s
test / publish-infra (pull_request) Has been skipped
test / integration-docker (push) Successful in 13s
prd-number / assign-numbers (push) Failing after 19s
lint / lint (push) Successful in 43s
Update Quality Badges / update-badges (push) Failing after 41s
test / unit (push) Successful in 1m42s
test / stage-firecracker-inputs (push) Successful in 2s
test / build-infra (push) Failing after 7s
test / integration-firecracker (push) Has been skipped
test / coverage (push) Has been skipped
test / publish-infra (push) Has been skipped
Add tests for the three uncovered paths introduced by the poll_ca_cert
extraction: the timeout + sleep branches in backend/util, and the
TimeoutError → GatewayError and TimeoutError → die() conversions in the
macOS and Firecracker callers.
2026-07-21 04:54:40 +00:00
didericis-claude 8e43c26ab4 fix(backend): extract poll_ca_cert helper and fix PRD port docs
Extract the shared CA cert polling loop into `backend/util.poll_ca_cert`
(firecracker and macos backends were duplicating deadline/sleep/raise logic).
Each caller now wraps a fetch lambda and converts TimeoutError to its own
error type. Also corrects the PRD port publication line from {port}:{port}
to {host_port}:8099.
2026-07-21 04:54:40 +00:00
didericis-claude 14ff4fe186 fix(lifecycle): build gateway before infra and fix orchestrator port mapping
- `_build_images()` now builds Dockerfile.gateway → Dockerfile.orchestrator
  → Dockerfile.infra in order; Dockerfile.infra starts FROM bot-bottle-gateway
  so the base must exist on clean hosts.
- Publish mapping corrected from `self.port:self.port` to `self.port:DEFAULT_PORT`
  (8099) — gateway_init hardcodes the orchestrator on port 8099 inside the
  container, so the host-side published port must map to that fixed internal port.
- BOT_BOTTLE_ORCHESTRATOR_URL inside the container now always points to
  127.0.0.1:8099, not self.port, since gateway daemons reach the orchestrator
  over loopback at the fixed internal port.
- Update test_ensure_running_builds_both_images → _all_images for the new
  three-step build sequence; add test_publish_maps_host_port_to_fixed_internal_port
  to lock in the port-mapping fix.

Addresses the P1 findings from the didericis-codex review on PR #432.
2026-07-21 04:54:40 +00:00
didericis-claude cae1215f63 test(lifecycle): cover edge paths to satisfy diff-coverage gate
Add three new tests:
- noop when healthy but docker inspect fails (returns True → don't churn)
- build failure raises GatewayError
- _ensure_network creates the network when it doesn't exist

Also update the integration test to use new OrchestratorService API
(infra_name/image instead of orchestrator_name/gateway_name/gateway_image).

Brings diff-coverage from 86% to 90.3% against origin/main.
2026-07-21 04:54:40 +00:00
didericis-claude 28766d7733 fix(pyright): resolve type errors introduced by lifecycle refactor
- Remove unused INFRA_IMAGE import from test_orchestrator_lifecycle
- Update integration test to use new single-container OrchestratorService
  API (infra_name/image replaces orchestrator_name/gateway_name/gateway_image)
- Move type: ignore to the lambda line in gateway_init SIGHUP handler
- Break two long lines in test_orchestrator_lifecycle
2026-07-21 04:54:40 +00:00
didericis-claude 819f967844 fix(lint): resolve pylint findings in gateway_init
- Extract _sigkill_all() to cut nesting depth below the 5-block limit
- Add pylint: disable=consider-using-with on Popen (process must outlive caller)
- Break long SIGHUP signal line to stay within 100 chars
2026-07-21 04:54:40 +00:00
didericis-claude 2f45f5afec feat(docker): consolidate to single infra container under gateway_init supervise tree
Collapses the two-container Docker model (gateway + orchestrator) into one
bot-bottle-infra container, matching the macOS and Firecracker backends.

- Dockerfile.infra: now a shared gateway+orchestrator base (COPY bot_bottle
  from orchestrator build, no CMD override)
- Dockerfile.infra.fc: new Firecracker-specific layer (buildah/crun/netavark)
- gateway_init: adds orchestrator daemon with _OPT_IN_DAEMONS gating so it
  only starts when BOT_BOTTLE_GATEWAY_DAEMONS explicitly includes it
- orchestrator/lifecycle: OrchestratorService manages one infra container;
  builds orchestrator (intermediate) then infra; live source bind-mounted at
  /bot-bottle-src with PYTHONPATH so the subprocess uses the checkout
- backend/consolidated_util: extracts provision_bottle + teardown_consolidated
  shared across all three backends; removes duplication in docker/fc/macos
  consolidated_launch modules
- firecracker/infra_vm: builds four images (orchestrator→gateway→infra→infra.fc)
- All unit tests updated and passing (1878 tests)
- PRD status: Draft → Active
2026-07-21 04:54:40 +00:00
didericis-claude 31a5ec2fc8 docs(prd): consolidate infra backend for docker
Adds PRD for collapsing the Docker backend from two containers
(gateway + orchestrator) to a single bot-bottle-infra container,
restructuring Dockerfile.infra as the shared base, and extracting
duplicated CA polling / teardown / provision helpers into a shared
backend utility module.

Closes #431
2026-07-21 04:49:18 +00:00
didericis-claude 1f192d785a refactor(config_store): replace generic key/value table with explicit schema
test / stage-firecracker-inputs (push) Successful in 2s
test / integration-docker (push) Successful in 17s
Update Quality Badges / update-badges (push) Failing after 42s
lint / lint (push) Successful in 47s
test / unit (push) Successful in 1m41s
test / build-infra (push) Successful in 3m43s
test / integration-firecracker (push) Successful in 1m28s
test / coverage (push) Successful in 1m29s
test / publish-infra (push) Successful in 1m49s
Switch orchestrator_config from a generic key/text store to a typed
single-row table with a dedicated teardown_timeout_seconds REAL column.
The CHECK (id = 1) constraint enforces at most one config row.

Remove TEARDOWN_TIMEOUT_KEY constant; update OrchestratorConfigStore to
expose get/set/delete_teardown_timeout_seconds() instead of the generic
get/set/delete(key). Resolving the DB value in resolve_teardown_timeout()
no longer needs a float() conversion since the column is already REAL.
2026-07-21 00:40:59 -04:00
didericis-codex 853b6d1678 test(orchestrator): satisfy config store type checking 2026-07-21 00:40:59 -04:00
didericis-claude cf9a53d582 fix: resolve teardown timeout before registration to prevent orphaned state
If resolve_teardown_timeout() raised after launch_consolidated()/register_agent()
returned, the teardown callback was never registered, leaving the bottle
registered with no cleanup path. Resolve the timeout into a local variable
before the registration call so any failure aborts before the bottle exists.

Add tests covering OrchestratorConfigStore, resolve_teardown_timeout priority
ordering (env > db > default), and the source-order invariant that the resolver
runs before registration in all three backends.
2026-07-21 00:40:59 -04:00
didericis-claude 0b36c3eb48 fix: make orchestrator teardown timeout configurable (closes #435)
Resolves via ENV VAR -> orchestrator DB config -> default (30 s, up from 5 s):
  BOT_BOTTLE_ORCHESTRATOR_TEARDOWN_TIMEOUT_SECONDS
  teardown_timeout_seconds key in new orchestrator_config table (bot-bottle.db)

New OrchestratorConfigStore (same DbStore/TableMigrations pattern as the
registry) stores the DB-level setting. resolve_teardown_timeout() implements
the priority chain and is called at stack.callback registration time in all
three backends (macos_container, docker, firecracker).
2026-07-21 00:40:59 -04:00
didericis-claude 28fcc3f2d2 test: cover inspect_container_network_ip parser directly
test / stage-firecracker-inputs (pull_request) Successful in 5s
test / unit (pull_request) Successful in 35s
test / integration-docker (pull_request) Successful in 9s
tracker-policy-pr / check-pr (pull_request) Successful in 8s
test / build-infra (pull_request) Successful in 3m48s
test / integration-firecracker (pull_request) Successful in 1m43s
test / coverage (pull_request) Successful in 1m53s
test / publish-infra (pull_request) Has been skipped
test / integration-docker (push) Successful in 11s
test / unit (push) Successful in 35s
test / stage-firecracker-inputs (push) Successful in 2s
Update Quality Badges / update-badges (push) Failing after 40s
lint / lint (push) Successful in 2m43s
test / build-infra (push) Successful in 3m25s
test / integration-firecracker (push) Successful in 1m45s
test / coverage (push) Successful in 1m50s
test / publish-infra (push) Successful in 2m7s
The None/empty-string distinction is the invariant that prevents
live_source_ips from treating a failed inspect as a legitimately
no-address container. Add TestInspectContainerNetworkIp to util tests
covering: IP found, CIDR stripping, no address yet, absent network
list, non-zero exit, malformed JSON, and unexpected JSON shape.
2026-07-21 03:49:19 +00:00
didericis-claude 571030b8e8 fix: require authoritative container snapshot before reconciling
test / stage-firecracker-inputs (pull_request) Successful in 5s
tracker-policy-pr / check-pr (pull_request) Successful in 12s
test / integration-docker (pull_request) Successful in 33s
test / unit (pull_request) Successful in 39s
lint / lint (push) Successful in 49s
test / build-infra (pull_request) Successful in 3m23s
test / integration-firecracker (pull_request) Successful in 1m42s
test / coverage (pull_request) Failing after 1m26s
test / publish-infra (pull_request) Has been skipped
A failed container list or a partial per-container inspect were
previously indistinguishable from a legitimately empty/partial live
set, so reconciliation could silently unregister healthy bottles.

- enumerate_active() now raises EnumerationError instead of returning []
  when `container list` fails
- Add inspect_container_network_ip() to util, which returns None on
  inspect failure (vs "" for "no DHCP address yet"), so live_source_ips
  can tell the two apart
- live_source_ips() raises EnumerationError on either failure mode;
  register_agent() catches it and skips reconciliation, same as
  OrchestratorClientError
- Update tests: rename test_empty_when_the_cli_fails to
  test_raises_when_the_cli_fails, update patching to the new function,
  add coverage for list failure, per-container inspect failure, and the
  launch-not-blocked path
2026-07-21 03:39:18 +00:00
didericis-codex 288b205a44 test(macos): isolate registration tests from container discovery
test / stage-firecracker-inputs (pull_request) Successful in 2s
test / integration-docker (pull_request) Successful in 6s
tracker-policy-pr / check-pr (pull_request) Successful in 5s
test / unit (pull_request) Successful in 30s
lint / lint (push) Successful in 42s
test / build-infra (pull_request) Successful in 3m18s
test / integration-firecracker (pull_request) Successful in 1m27s
test / coverage (pull_request) Successful in 1m51s
test / publish-infra (pull_request) Has been skipped
2026-07-21 03:27:16 +00:00
didericis 0c1d27b605 fix(egress): the unattributed message must name the token-mismatch cause too
`/resolve` fail-closes on a missing/ambiguous registry row *and* on a
request whose identity token doesn't match. The message named only the
first, so a bottle that was registered correctly but sent no token read as
"not registered" and sent the reader looking for a deregistered bottle.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:27:16 +00:00
didericis 69361114d1 fix(egress): name the real fault when a deny-all is not an allowlist miss
An unattributed bottle, an unreachable orchestrator, and an unparseable
policy all become a deny-all Config, and a deny-all is indistinguishable
from "policy loaded, host not allowed" at the decision point — both are just
"no matching route". So every one of them reported `host X is not in the
bottle's egress.routes allowlist`, which reads as a config problem and sends
the operator hunting for a route that was never missing. Diagnosing a
bricked registration cost hours for exactly this reason.

Carry the structural reason on Config and prefer it in decide(). A genuine
allowlist miss — a policy that loaded and simply lacks the host — keeps the
original wording, so the message now tells the two cases apart.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:27:16 +00:00
didericis e4d53fd360 fix(orchestrator): reap registry rows whose bottle is no longer running
A registry row only ever left the registry two ways: an explicit
teardown_bottle (the launcher's cleanup callback) or the same-IP supersede
sweep in register(). Neither runs when the launching CLI dies hard, so the
row outlives its container.

That orphan is not inert. Source IPs are recycled by the backend's DHCP and
by_source_ip fail-closes on ambiguity, so a leftover row at a reused address
resolves no policy at all for the next bottle that lands there — and a
bottle with no policy denies every host, which surfaces to the agent as
"host X is not in the allowlist" for hosts that were never the problem.

Add reap_absent/reconcile and call it from the macOS launch path before
registering, so each launch self-heals the registry. Restores the invariant
the data plane needs: at most one active row per live address, and none for
a dead one. The second half matters as much as the first — when several rows
claim a *live* address the newest wins and the rest are swept, otherwise a
recycled address stays ambiguous, which is exactly the bricked state.

The host supplies the live set because the orchestrator runs inside the
infra container and cannot see the backend. A grace window exempts rows
younger than it, so reconciliation cannot race a bottle still coming up, and
a reconcile failure is logged rather than blocking an otherwise-fine launch.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-21 03:27:16 +00:00
41 changed files with 1685 additions and 432 deletions
+13 -36
View File
@@ -1,45 +1,22 @@
# Firecracker single infra-VM image (PRD 0070 Stage B).
# Shared infra image: gateway data plane + orchestrator control plane.
#
# The per-host infra VM runs the orchestrator control plane, the gateway
# data plane, AND builds agent images (buildah) — all in one microVM (see
# backend/firecracker/infra_vm.py). It composes:
# * FROM the gateway image (mitmproxy / git / gitleaks / supervise + the
# flat daemon modules) — now trixie-based, so buildah 1.39 is available;
# * `COPY --from` the orchestrator image's content (the single definition
# of the control-plane payload — see Dockerfile.orchestrator), so this
# VM and the docker backend share one orchestrator definition; and
# * buildah, installed HERE only (the docker orchestrator/gateway images
# never carry it).
# Used directly by the Docker backend (run as one `bot-bottle-infra`
# container, replacing the prior two-container split). The Firecracker
# backend extends this via Dockerfile.infra.fc, adding buildah/crun/
# netavark for in-VM agent-image building.
#
# Dockerfile.orchestrator is the single definition of the orchestrator
# content (the lean `bot_bottle` package on python:3.12-slim). Both this
# image and Dockerfile.infra.fc pull it in via `COPY --from`.
#
# multi-`FROM` can't union two bases (that's multi-stage, not multiple
# inheritance), so the orchestrator content is pulled in via `COPY --from`
# rather than a second base. Both images share the trixie `python:3.12-slim`
# base, so the copy is clean (same python; future installed deps copy too).
#
# The docker backend keeps orchestrator + gateway as separate images; this
# combined image exists only for the Firecracker single-VM cut. Splitting a
# service back into its own VM later is a routing change, not a repackaging
# (PRD 0070's "secret concentration"; a disposable builder can boot from
# this same image on its own TAP).
FROM bot-bottle-gateway:latest
# --- in-VM agent-image builder (PRD 0069 Stage 3) -------------------
# The Firecracker backend builds users' agent Dockerfiles *inside this VM*
# with buildah (rootless, daemonless) instead of on the host — no host
# Docker daemon, no root-equivalent `docker` group. `crun` is the OCI
# runtime; `netavark` + `aardvark-dns` are the network backend for `FROM`
# pulls + `RUN` egress. Requires the trixie base (buildah 1.39: bookworm's
# 1.28 can't parse Dockerfile heredocs that agent images use).
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
buildah crun netavark aardvark-dns \
&& rm -rf /var/lib/apt/lists/*
# vfs + chroot: buildah works as root in the bare microVM (no
# fuse-overlayfs / overlay module / subuid maps). Matches image_builder.
ENV STORAGE_DRIVER=vfs \
BUILDAH_ISOLATION=chroot
# The orchestrator content, pulled from its single definition. The gateway
# image already has the flat daemon modules under /app; this adds the full
# `bot_bottle` package so `python3 -m bot_bottle.orchestrator` resolves.
# The orchestrator content, from its single definition. The gateway image
# already has the flat daemon modules under /app; this adds the full
# `bot_bottle` package so `python3 -m bot_bottle.orchestrator` resolves —
# used by gateway_init when BOT_BOTTLE_GATEWAY_DAEMONS includes `orchestrator`.
COPY --from=bot-bottle-orchestrator:latest /app/bot_bottle /app/bot_bottle
+23
View File
@@ -0,0 +1,23 @@
# Firecracker infra VM image (PRD 0070 Stage B).
#
# Extends the shared infra base (Dockerfile.infra: gateway + orchestrator
# control plane) with the in-VM agent-image builder. The Firecracker backend
# builds users' agent Dockerfiles *inside this VM* with buildah (rootless,
# daemonless) instead of on the host — no host Docker daemon, no
# root-equivalent `docker` group.
#
# Requires the trixie base from bot-bottle-gateway (buildah 1.39: bookworm's
# 1.28 can't parse Dockerfile heredocs that agent images use).
#
# `crun` is the OCI runtime; `netavark` + `aardvark-dns` are the network
# backend for `FROM` pulls + `RUN` egress. `vfs` + `chroot`: buildah works
# as root in the bare microVM (no fuse-overlayfs / overlay module / subuid
# maps). Matches image_builder.
FROM bot-bottle-infra:latest
RUN apt-get update \
&& apt-get install -y --no-install-recommends \
buildah crun netavark aardvark-dns \
&& rm -rf /var/lib/apt/lists/*
ENV STORAGE_DRIVER=vfs \
BUILDAH_ISOLATION=chroot
+60
View File
@@ -0,0 +1,60 @@
"""Shared helpers for the consolidated launch sequence (PRD 0070).
Logic that was duplicated across the docker, macos_container, and
firecracker consolidated_launch modules — extracted so each backend
imports it rather than re-implementing it.
"""
from __future__ import annotations
from ..egress import EgressPlan
from ..git_gate import GitGatePlan
from ..orchestrator.client import OrchestratorClient
from ..orchestrator.registration import registration_inputs
from .docker.gateway_provision import GatewayTransport, deprovision_git_gate, provision_git_gate
def provision_bottle(
client: OrchestratorClient,
source_ip: str,
egress_plan: EgressPlan,
git_gate_plan: GitGatePlan,
transport: GatewayTransport,
*,
image_ref: str = "",
tokens: dict[str, str] | None = None,
):
"""Register the bottle and provision its git-gate state. Rolls back the
registration if provisioning fails so no orphan is left. Returns the
`RegisteredBottle` from the orchestrator."""
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
source_ip, image_ref=image_ref, policy=inputs.policy,
metadata=inputs.metadata, tokens=tokens,
)
try:
provision_git_gate(transport, reg.bottle_id, git_gate_plan)
except Exception:
client.teardown_bottle(reg.bottle_id)
raise
return reg
def teardown_consolidated(
bottle_id: str,
transport: GatewayTransport,
*,
orchestrator_url: str,
timeout: float | None = None,
) -> None:
"""Deregister the bottle and remove its git-gate state. Both steps are
idempotent so this is safe from a cleanup trap."""
from ..orchestrator.config_store import DEFAULT_TEARDOWN_TIMEOUT_SECONDS
OrchestratorClient(
orchestrator_url,
timeout=timeout if timeout is not None else DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
).teardown_bottle(bottle_id)
deprovision_git_gate(transport, bottle_id)
__all__ = ["provision_bottle", "teardown_consolidated"]
@@ -1,19 +1,13 @@
"""Consolidated bottle launch sequence for the docker backend (PRD 0070).
Composes the orchestrator primitives into the register/teardown sequence that
replaces the per-bottle gateway:
Composes the orchestrator primitives into the register/teardown sequence:
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.
1. ensure the single infra container (control plane + gateway) is up;
2. allocate the bottle a pinned source IP on the gateway network;
3. register it and provision its git-gate repos/creds into the 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
Returns a `LaunchContext` with everything the agent container needs to
attach. The agent `docker run` itself is the backend's job; this owns the
orchestrator-facing wiring so that sequence stays testable in isolation.
"""
@@ -25,15 +19,12 @@ 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 OrchestratorService
from ...orchestrator.registration import registration_inputs
from ...orchestrator.gateway import GATEWAY_NETWORK
from ...orchestrator.lifecycle import INFRA_NAME, OrchestratorService
from ..consolidated_util import provision_bottle
from ..consolidated_util import teardown_consolidated as _teardown_util
from .gateway_provision import DockerGatewayTransport
from .gateway_net import next_free_ip
from .gateway_provision import (
DockerGatewayTransport,
deprovision_git_gate,
provision_git_gate,
)
class ConsolidatedLaunchError(RuntimeError):
@@ -75,24 +66,21 @@ def _container_ip(name: str, network: str) -> str:
ip = proc.stdout.strip()
if proc.returncode != 0 or not ip:
raise ConsolidatedLaunchError(
f"gateway {name} has no address on {network}: {proc.stderr.strip()}"
f"container {name} has no address on {network}: {proc.stderr.strip()}"
)
return ip
def _network_container_ips(network: str) -> list[str]:
"""Every address currently assigned on the gateway network — the ground
truth for "in use": the gateway + orchestrator infrastructure containers
and every live agent. Read from the network so a new bottle can't collide
with anything actually attached (a registry-only view would miss the
orchestrator/gateway containers)."""
truth for "in use": the infra container and every live agent. Read from
the network so a new bottle can't collide with anything actually attached."""
proc = run_docker([
"docker", "network", "inspect", "--format",
"{{range .Containers}}{{.IPv4Address}} {{end}}", network,
])
ips: list[str] = []
for entry in proc.stdout.split():
# entries look like "172.20.0.2/16" — keep the address.
ips.append(entry.split("/", 1)[0])
return ips
@@ -104,33 +92,24 @@ def launch_consolidated(
image_ref: str = "",
tokens: dict[str, str] | None = None,
service: OrchestratorService | None = None,
gateway_name: str = GATEWAY_NAME,
infra_name: str = INFRA_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."""
"""Ensure the infra container is up, allocate + register the bottle, and
provision its git-gate state. Returns the agent's attach context."""
service = service or OrchestratorService()
url = service.ensure_running()
client = OrchestratorClient(url)
cidr = _network_cidr(network)
gateway_ip = _container_ip(gateway_name, network)
gateway_ip = _container_ip(infra_name, network)
source_ip = next_free_ip(cidr, _network_container_ips(network))
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
source_ip, image_ref=image_ref, policy=inputs.policy,
metadata=inputs.metadata, tokens=tokens,
transport = DockerGatewayTransport(infra_name)
reg = provision_bottle(
client, source_ip, egress_plan, git_gate_plan, transport,
image_ref=image_ref, tokens=tokens,
)
try:
provision_git_gate(
DockerGatewayTransport(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,
@@ -142,12 +121,12 @@ def launch_consolidated(
def teardown_consolidated(
bottle_id: str, *, orchestrator_url: str, gateway_name: str = GATEWAY_NAME,
bottle_id: str, *, orchestrator_url: str, infra_name: str = INFRA_NAME,
timeout: float | None = None,
) -> 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(DockerGatewayTransport(gateway_name), bottle_id)
"""Deregister the bottle and remove its git-gate state. Idempotent."""
_teardown_util(bottle_id, DockerGatewayTransport(infra_name),
orchestrator_url=orchestrator_url, timeout=timeout)
__all__ = [
+5 -1
View File
@@ -62,6 +62,7 @@ from .compose import (
write_compose_file,
)
from .consolidated_compose import consolidated_agent_compose
from ...orchestrator.config_store import resolve_teardown_timeout
from .consolidated_launch import launch_consolidated, teardown_consolidated
from ...orchestrator.gateway import DockerGateway
@@ -133,11 +134,14 @@ def launch(
token_values = egress_resolve_token_values(
plan.egress_plan.token_env_map, effective_env,
)
teardown_timeout = resolve_teardown_timeout()
ctx = launch_consolidated(
plan.egress_plan, git_gate_plan, image_ref=plan.image, tokens=token_values,
)
stack.callback(
teardown_consolidated, ctx.bottle_id, orchestrator_url=ctx.orchestrator_url,
teardown_consolidated, ctx.bottle_id,
orchestrator_url=ctx.orchestrator_url,
timeout=teardown_timeout,
)
# Step 4: install the SHARED gateway CA into the agent (replaces the
@@ -33,8 +33,7 @@ from ...orchestrator.client import OrchestratorClient
from ...orchestrator.lifecycle import (
OrchestratorStartError, # re-exported so callers can catch it
)
from ...orchestrator.registration import registration_inputs
from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
from ..consolidated_util import provision_bottle, teardown_consolidated as _teardown_util
from . import infra_vm
@@ -68,18 +67,11 @@ def launch_consolidated(
url = infra.control_plane_url
client = OrchestratorClient(url)
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
guest_ip, image_ref=image_ref, policy=inputs.policy,
metadata=inputs.metadata, tokens=tokens,
transport = infra_vm.gateway_transport()
reg = provision_bottle(
client, guest_ip, egress_plan, git_gate_plan, transport,
image_ref=image_ref, tokens=tokens,
)
try:
provision_git_gate(
infra_vm.gateway_transport(), reg.bottle_id, git_gate_plan)
except Exception:
client.teardown_bottle(reg.bottle_id)
raise
# The shared gateway CA every agent on this host trusts for TLS
# interception — fetched from the infra VM over SSH.
return LaunchContext(
@@ -91,13 +83,15 @@ def launch_consolidated(
)
def teardown_consolidated(bottle_id: str, *, orchestrator_url: str) -> None:
def teardown_consolidated(
bottle_id: str, *, orchestrator_url: str, timeout: float | None = None,
) -> None:
"""Deregister the bottle and remove its git-gate state from the gateway
VM. Both steps are idempotent so this is safe from a cleanup trap. Does
NOT stop the infra VM — it's a persistent per-host singleton shared by
every bottle."""
OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
deprovision_git_gate(infra_vm.gateway_transport(), bottle_id)
_teardown_util(bottle_id, infra_vm.gateway_transport(),
orchestrator_url=orchestrator_url, timeout=timeout)
__all__ = [
@@ -41,7 +41,7 @@ from . import util
_ARTIFACT_FORMAT = "1"
_REPO_ROOT = Path(__file__).resolve().parents[3]
_DOCKERFILES = ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra")
_DOCKERFILES = ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra", "Dockerfile.infra.fc")
_DEFAULT_BASE = "https://gitea.dideric.is"
_DEFAULT_OWNER = "didericis"
+16 -13
View File
@@ -33,6 +33,7 @@ from pathlib import Path
from typing import Generator
from ...log import die, info
from .. import util as backend_util
from ..docker import util as docker_mod
from ..docker.gateway_provision import GatewayProvisionError
from . import firecracker_vm, infra_artifact, netpool, util
@@ -93,19 +94,18 @@ class InfraVm:
"""The gateway's mitmproxy CA (PEM) that agents install to trust its
TLS interception. Generated a moment after boot, so this polls over
SSH until it appears (mirrors DockerGateway.ca_cert_pem)."""
deadline = time.monotonic() + timeout
while True:
def _fetch() -> str | None:
proc = subprocess.run(
util.ssh_base_argv(self.private_key, self.guest_ip)
+ [f"cat {_GATEWAY_CA_PATH}"],
capture_output=True, text=True, timeout=15, check=False,
)
if proc.returncode == 0 and "BEGIN CERTIFICATE" in proc.stdout:
return proc.stdout
if time.monotonic() >= deadline:
die(f"gateway CA not available after {timeout:g}s: "
f"{proc.stderr.strip() or 'empty'}")
time.sleep(_HEALTH_POLL_SECONDS)
ok = proc.returncode == 0 and "BEGIN CERTIFICATE" in proc.stdout
return proc.stdout if ok else None
try:
return backend_util.poll_ca_cert(_fetch, timeout=timeout)
except TimeoutError as exc:
die(str(exc))
def ensure_built() -> None:
@@ -125,16 +125,19 @@ def ensure_built() -> None:
def build_infra_images_with_docker() -> None:
"""Build the three fixed images from source with host Docker: orchestrator,
gateway, then the combined infra image (`COPY --from` orchestrator, `FROM`
gateway). The launch host uses this only in `BOT_BOTTLE_INFRA_BUILD=local`
mode; `publish_infra` uses it off-host to produce the published artifact."""
"""Build the four fixed images from source with host Docker: orchestrator,
gateway, the shared infra base (Dockerfile.infra), then the Firecracker
infra image (Dockerfile.infra.fc: FROM infra + buildah). The launch host
uses this only in `BOT_BOTTLE_INFRA_BUILD=local` mode; `publish_infra`
uses it off-host to produce the published artifact."""
docker_mod.build_image(
_ORCHESTRATOR_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.orchestrator")
docker_mod.build_image(
_GATEWAY_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.gateway")
docker_mod.build_image(
_INFRA_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.infra")
"bot-bottle-infra:latest", str(_REPO_ROOT), dockerfile="Dockerfile.infra")
docker_mod.build_image(
_INFRA_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.infra.fc")
def build_infra_rootfs_dir() -> Path:
+3
View File
@@ -52,6 +52,7 @@ from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
from . import firecracker_vm, image_builder, isolation_probe, netpool, util
from .bottle import FirecrackerBottle
from .bottle_plan import FirecrackerBottlePlan
from ...orchestrator.config_store import resolve_teardown_timeout
from .consolidated_launch import (
launch_consolidated,
teardown_consolidated,
@@ -112,6 +113,7 @@ def launch(
token_values = egress_resolve_token_values(
plan.egress_plan.token_env_map, effective_env,
)
teardown_timeout = resolve_teardown_timeout()
ctx = launch_consolidated(
plan.egress_plan, git_gate_plan,
guest_ip=slot.guest_ip,
@@ -121,6 +123,7 @@ def launch(
stack.callback(
teardown_consolidated, ctx.bottle_id,
orchestrator_url=ctx.orchestrator_url,
timeout=teardown_timeout,
)
# Step 5: install the SHARED gateway CA (replaces the per-bottle CA).
@@ -36,9 +36,11 @@ from dataclasses import dataclass
from ...egress import EgressPlan
from ...git_gate import GitGatePlan
from ...orchestrator.client import OrchestratorClient
from ...orchestrator.registration import registration_inputs
from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
from ...log import info
from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
from ..consolidated_util import provision_bottle, teardown_consolidated as _teardown_util
from . import util as container_mod
from .enumerate import CONTAINER_NAME_PREFIX, EnumerationError, enumerate_active
from .gateway import GATEWAY_NETWORK
from .gateway_provision import AppleGatewayTransport
from .infra import MacosInfraService, OrchestratorStartError
@@ -89,6 +91,32 @@ def ensure_gateway(
)
def live_source_ips(network: str) -> list[str]:
"""Every running agent container's address on `network`.
The reconciliation input: the orchestrator lives inside the infra
container and cannot enumerate the host's containers, so the host has to
tell it which bottles are actually up. Containers that have not been
assigned an address yet contribute nothing the reap's grace window, not
this list, is what protects an in-flight launch.
Raises `EnumerationError` when the live set cannot be determined
authoritatively: either the container listing fails or any individual
inspect fails. Callers must skip reconciliation in that case to avoid
unregistering healthy bottles."""
ips: list[str] = []
for agent in enumerate_active():
name = f"{CONTAINER_NAME_PREFIX}{agent.slug}"
ip = container_mod.inspect_container_network_ip(name, network)
if ip is None:
raise EnumerationError(
f"container inspect {name!r} failed; live set is not authoritative"
)
if ip:
ips.append(ip)
return ips
def register_agent(
egress_plan: EgressPlan,
git_gate_plan: GitGatePlan,
@@ -103,17 +131,20 @@ def register_agent(
container it is the attribution key the gateway resolves policy by.
Raises on failure; the caller tears down."""
client = OrchestratorClient(endpoint.orchestrator_url)
inputs = registration_inputs(egress_plan)
reg = client.register_bottle(
source_ip, image_ref=image_ref, policy=inputs.policy,
metadata=inputs.metadata, tokens=tokens,
)
# Self-heal before registering: a launcher that died hard (SIGKILL, closed
# terminal, host sleep) never ran its teardown callback, leaving an active
# row with no container. vmnet recycles addresses, so such a row can
# collide with this bottle's — and `by_source_ip` fail-closes on ambiguity,
# which would resolve no policy at all and deny every host. Best-effort: a
# reconciliation failure must not block an otherwise-fine launch.
try:
provision_git_gate(AppleGatewayTransport(), 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
client.reconcile(live_source_ips(endpoint.network))
except (OrchestratorClientError, EnumerationError) as e:
info(f"registry reconciliation skipped: {e}")
reg = provision_bottle(
client, source_ip, egress_plan, git_gate_plan, AppleGatewayTransport(),
image_ref=image_ref, tokens=tokens,
)
return LaunchContext(
bottle_id=reg.bottle_id,
identity_token=reg.identity_token,
@@ -124,18 +155,21 @@ def register_agent(
)
def teardown_consolidated(bottle_id: str, *, orchestrator_url: str) -> None:
def teardown_consolidated(
bottle_id: str, *, orchestrator_url: str, timeout: float | None = None,
) -> 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. Does NOT
stop the gateway it's a persistent per-host singleton."""
OrchestratorClient(orchestrator_url).teardown_bottle(bottle_id)
deprovision_git_gate(AppleGatewayTransport(), bottle_id)
_teardown_util(bottle_id, AppleGatewayTransport(),
orchestrator_url=orchestrator_url, timeout=timeout)
__all__ = [
"GatewayEndpoint",
"LaunchContext",
"ensure_gateway",
"live_source_ips",
"register_agent",
"teardown_consolidated",
"ConsolidatedLaunchError",
@@ -19,6 +19,10 @@ CONTAINER_NAME_PREFIX = "bot-bottle-"
_INFRA_NAMES = frozenset({INFRA_NAME})
class EnumerationError(RuntimeError):
"""container list failed; the resulting live set is not authoritative."""
def enumerate_active() -> list[ActiveAgent]:
result = subprocess.run(
["container", "list", "--quiet"],
@@ -27,7 +31,10 @@ def enumerate_active() -> list[ActiveAgent]:
check=False,
)
if result.returncode != 0:
return []
raise EnumerationError(
f"container list failed: "
f"{(result.stderr or '').strip() or '<no stderr>'}"
)
out: list[ActiveAgent] = []
for name in sorted(line.strip() for line in result.stdout.splitlines()):
if not name.startswith(CONTAINER_NAME_PREFIX) or name in _INFRA_NAMES:
+9 -10
View File
@@ -53,6 +53,7 @@ from ...paths import (
HOST_DB_FILENAME,
host_control_plane_token,
)
from .. import util as backend_util
from . import util as container_mod
from .gateway import (
DEFAULT_CA_TIMEOUT_SECONDS,
@@ -263,18 +264,16 @@ class MacosInfraService:
interception. Read out of the container (the CA lives on a
container-internal path, not a host mount); polls because mitmproxy
writes it a beat after start."""
deadline = time.monotonic() + timeout
while True:
def _fetch() -> str | None:
result = container_mod.run_container_argv(
["container", "exec", self._name, "cat", GATEWAY_CA_CERT])
if result.returncode == 0 and result.stdout.strip():
return result.stdout
if time.monotonic() >= deadline:
raise GatewayError(
f"gateway CA not available in {self._name} after {timeout:g}s: "
f"{(result.stderr or '').strip() or 'empty'}"
)
time.sleep(_CA_POLL_SECONDS)
return result.stdout if result.returncode == 0 and result.stdout.strip() else None
try:
return backend_util.poll_ca_cert(_fetch, timeout=timeout)
except TimeoutError as exc:
raise GatewayError(
f"gateway CA not available in {self._name} after {timeout:g}s"
) from exc
def stop(self) -> None:
"""Remove the infra container (idempotent). The DB volume persists."""
@@ -65,6 +65,7 @@ from .gateway_hosts import (
set_gateway_host,
)
from .bottle_plan import MacosContainerBottlePlan
from ...orchestrator.config_store import resolve_teardown_timeout
from .consolidated_launch import (
GatewayEndpoint,
ensure_gateway,
@@ -142,6 +143,7 @@ def launch(
token_values = egress_resolve_token_values(
plan.egress_plan.token_env_map, effective_env,
)
teardown_timeout = resolve_teardown_timeout()
ctx = register_agent(
plan.egress_plan,
plan.git_gate_plan,
@@ -153,6 +155,7 @@ def launch(
stack.callback(
teardown_consolidated, ctx.bottle_id,
orchestrator_url=ctx.orchestrator_url,
timeout=teardown_timeout,
)
info(
f"agent {plan.container_name} registered "
@@ -572,6 +572,41 @@ def try_container_ipv4_on_network(name: str, network: str) -> str:
return ""
def inspect_container_network_ip(name: str, network: str) -> str | None:
"""IP of `name` on `network`, distinguishing inspect failure from "not yet".
Returns:
- the IP string when the container has one on `network`
- "" when inspect succeeds but no address is assigned yet (in-flight DHCP)
- None when the inspect command itself fails (authoritative list impossible)
"""
result = subprocess.run(
[_CONTAINER, "inspect", name],
capture_output=True, text=True, check=False,
)
if result.returncode != 0:
return None
try:
data = json.loads(result.stdout or "[]")
except json.JSONDecodeError:
return None
if isinstance(data, list):
data = data[0] if data else {}
if not isinstance(data, dict):
return None
status = data.get("status")
networks = status.get("networks") if isinstance(status, dict) else None
if not isinstance(networks, list):
return ""
for entry in networks:
if not isinstance(entry, dict) or entry.get("network") != network:
continue
raw = entry.get("ipv4Address")
if isinstance(raw, str) and raw:
return raw.split("/", 1)[0]
return ""
def wait_container_ipv4_on_network(
name: str, network: str, *, timeout: float = 15.0, poll: float = 0.25,
) -> str:
+20
View File
@@ -7,6 +7,8 @@ from __future__ import annotations
import hashlib
import os
import ssl
import time
from collections.abc import Callable
from pathlib import Path
from typing import TYPE_CHECKING
@@ -15,6 +17,24 @@ from ..log import die, info
if TYPE_CHECKING:
from ..egress import EgressPlan
_CA_POLL_INTERVAL = 0.5
def poll_ca_cert(fetch: Callable[[], str | None], *, timeout: float) -> str:
"""Poll `fetch` until it returns a non-empty PEM string or `timeout` expires.
`fetch` should return the PEM on success and `None` (or empty string) when
the cert is not yet available. Raises `TimeoutError` if the cert never
appears within `timeout` seconds."""
deadline = time.monotonic() + timeout
while True:
result = fetch()
if result:
return result
if time.monotonic() >= deadline:
raise TimeoutError(f"CA cert not available after {timeout:g}s")
time.sleep(_CA_POLL_INTERVAL)
# Debian-family CA layout, shared by every backend (all guest images
# are Debian-family). AGENT_CA_PATH is the source path that
+1
View File
@@ -379,6 +379,7 @@ class EgressAddon:
env,
request_method=flow.request.method,
request_headers=req_headers,
deny_reason=config.deny_reason,
)
if decision.action == "block":
+45 -6
View File
@@ -89,6 +89,14 @@ LOG_FULL = 2 # log block/warn events + full request and response bodies
class Config:
routes: tuple[Route, ...]
log: int = LOG_OFF
# Why this Config is a deny-all, when it is one for a reason *other* than
# the bottle's own policy genuinely not listing the host. A deny-all is
# indistinguishable from "policy loaded, host not allowed" at the decision
# point — both are simply "no matching route" — so without this the
# operator sees `host X is not in the allowlist` and goes hunting for a
# missing route that was never the problem. Empty for a normally-parsed
# policy; `decide` prefers it over the allowlist wording when set.
deny_reason: str = ""
@dataclass(frozen=True)
@@ -405,16 +413,40 @@ class PolicyResolverLike(typing.Protocol):
...
# Deny-all explanations. Each names the *actual* failure so an operator isn't
# sent looking for a missing egress route when the bottle never had a policy
# to begin with — the failure mode that made a bricked registration read like
# a misconfigured allowlist.
DENY_UNATTRIBUTED = (
"egress: this request was not attributed to any bottle, so no egress "
"policy applies and every host is denied. Either the bottle's registry "
"row is missing/ambiguous (torn down, or another bottle claimed its "
"source IP), or the request carried no matching identity token — check "
"that the caller's proxy URL includes it. This is not an allowlist problem."
)
DENY_UNPARSEABLE = (
"egress: this bottle's egress policy could not be parsed, so it is being "
"treated as deny-all. Fix the bottle's egress.routes; every host is denied "
"until it loads."
)
DENY_RESOLVER_ERROR = (
"egress: the orchestrator could not be reached to resolve this bottle's "
"egress policy, so every host is denied (fail-closed). Check that the "
"control plane is up; this is not an allowlist problem."
)
def _config_from_policy(policy: "str | None") -> "Config":
"""Parse a resolved policy blob into a Config, fail-closed: None / empty /
unparseable all become a deny-all Config (no routes every request
blocked)."""
blocked). Each deny-all carries the reason it is one, so the block message
names the real fault instead of blaming the allowlist."""
if not policy:
return Config(routes=()) # unattributed or empty → deny-all
return Config(routes=(), deny_reason=DENY_UNATTRIBUTED)
try:
return load_config(policy)
except ValueError:
return Config(routes=()) # unparseable policy → deny
return Config(routes=(), deny_reason=DENY_UNPARSEABLE)
def resolve_client_config(
@@ -428,7 +460,7 @@ def resolve_client_config(
try:
policy = resolver.resolve(client_ip, identity_token)
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return Config(routes=()) # orchestrator unreachable/errored → deny
return Config(routes=(), deny_reason=DENY_RESOLVER_ERROR)
return _config_from_policy(policy)
@@ -457,7 +489,7 @@ def resolve_client_context(
client_ip, identity_token,
)
except Exception: # noqa: BLE001 # pylint: disable=broad-exception-caught
return Config(routes=()), "", {} # orchestrator unreachable/errored → deny
return Config(routes=(), deny_reason=DENY_RESOLVER_ERROR), "", {}
return _config_from_policy(policy), (bottle_id or ""), tokens
@@ -572,12 +604,16 @@ def decide(
*,
request_method: str = "GET",
request_headers: typing.Mapping[str, str] | None = None,
deny_reason: str = "",
) -> Decision:
"""`deny_reason` is `Config.deny_reason`: when the deny-all came from a
missing/unparseable policy rather than the bottle's own allowlist, report
that instead of implying a route is merely absent."""
route = match_route(routes, request_host)
if route is None:
return Decision(
action="block",
reason=(
reason=deny_reason or (
f"egress: host {request_host!r} is not in the "
f"bottle's egress.routes allowlist. Declare a "
f"route for it or remove the request."
@@ -852,6 +888,9 @@ __all__ = [
"is_git_push_request",
"is_git_fetch_request",
"load_config",
"DENY_UNATTRIBUTED",
"DENY_UNPARSEABLE",
"DENY_RESOLVER_ERROR",
"resolve_client_config",
"resolve_client_context",
"PolicyResolverLike",
+35 -15
View File
@@ -61,6 +61,11 @@ class _DaemonSpec:
_EGRESS_ONLY_ENV_PREFIXES: tuple[str, ...] = ("EGRESS_TOKEN_",)
_READY_GATED_DAEMONS: tuple[str, ...] = ("git-gate", "git-http")
# Daemons that must be requested explicitly via BOT_BOTTLE_GATEWAY_DAEMONS
# and are NOT started in the default (env-var-unset) case. The orchestrator
# only runs in the combined infra container, never in a standalone gateway.
_OPT_IN_DAEMONS: frozenset[str] = frozenset({"orchestrator"})
def _env_for_daemon(name: str, base_env: dict[str, str]) -> dict[str, str]:
"""Egress sees the full bundle env. Everyone else gets a copy
@@ -75,7 +80,14 @@ def _env_for_daemon(name: str, base_env: dict[str, str]) -> dict[str, str]:
}
# The orchestrator is listed first so it starts before the gateway daemons,
# giving the control plane a head start to accept /resolve calls. The gateway
# daemons tolerate early /resolve failures and retry per-request.
_DAEMONS: tuple[_DaemonSpec, ...] = (
_DaemonSpec("orchestrator", (
"python3", "-m", "bot_bottle.orchestrator",
"--host", "0.0.0.0", "--port", "8099", "--broker", "stub",
)),
_DaemonSpec("egress", ("/bin/sh", "/app/egress-entrypoint.sh")),
_DaemonSpec("git-gate", ("/bin/sh", "/git-gate-entrypoint.sh")),
_DaemonSpec("git-http", ("python3", "-m", "bot_bottle.git_http_backend")),
@@ -103,18 +115,20 @@ def _selected_daemons(
env: dict[str, str],
all_daemons: Sequence[_DaemonSpec] | None = None,
) -> tuple[_DaemonSpec, ...]:
"""Filter the daemon set by the BOT_BOTTLE_GATEWAY_DAEMONS env
var. Unknown names in the list are ignored the caller is the
source of truth for which daemons are wired.
"""Filter the daemon set by the BOT_BOTTLE_GATEWAY_DAEMONS env var.
`all_daemons` defaults to `_DAEMONS` resolved at call time (not
at definition time), so tests can monkey-patch the module-level
`_DAEMONS` and have the new value take effect."""
When the var is unset/empty, return all non-opt-in daemons (the
standard gateway subset). Opt-in daemons (e.g. `orchestrator`) only
run when explicitly named they never start in a plain gateway
container that doesn't set the env var. Unknown names are ignored.
`all_daemons` defaults to `_DAEMONS` resolved at call time (not at
definition time), so tests can pass a custom list."""
if all_daemons is None:
all_daemons = _DAEMONS
raw = env.get("BOT_BOTTLE_GATEWAY_DAEMONS", "").strip()
if not raw:
return tuple(all_daemons)
return tuple(d for d in all_daemons if d.name not in _OPT_IN_DAEMONS)
wanted = {n.strip() for n in raw.split(",") if n.strip()}
return tuple(d for d in all_daemons if d.name in wanted)
@@ -136,7 +150,7 @@ def _pump(name: str, stream: IO[bytes]) -> None:
def _spawn(spec: _DaemonSpec) -> subprocess.Popen[bytes]:
env = _env_for_daemon(spec.name, dict(os.environ))
proc = subprocess.Popen(
proc = subprocess.Popen( # pylint: disable=consider-using-with
_argv_for_daemon(spec.name, spec.argv, env),
stdout=subprocess.PIPE,
stderr=subprocess.STDOUT,
@@ -183,6 +197,14 @@ class _Supervisor:
except ProcessLookupError:
pass
def _sigkill_all(self) -> None:
for _, p in self.procs:
if p.poll() is None:
try:
p.kill()
except ProcessLookupError:
pass
def request_restart(self, daemon_name: str) -> bool:
"""Queue a daemon restart for the main loop to process.
@@ -235,12 +257,7 @@ class _Supervisor:
f"grace ({_GRACE_SECONDS:.0f}s) elapsed; SIGKILL on "
f"{', '.join(still_running)}"
)
for _, p in self.procs:
if p.poll() is None:
try:
p.kill()
except ProcessLookupError:
pass
self._sigkill_all()
done = all(p.poll() is not None for _, p in self.procs)
if done:
@@ -361,7 +378,10 @@ def main(argv: Sequence[str] | None = None) -> int:
# --signal HUP <bundle>` after writing routes.yaml. The kernel
# delivers SIGHUP to PID 1 (this supervisor); forward it to
# mitmdump so it reloads its addon.
signal.signal(signal.SIGHUP, lambda *_: sup.forward_signal(signal.SIGHUP, "egress")) # type: ignore
signal.signal(
signal.SIGHUP,
lambda *_: sup.forward_signal(signal.SIGHUP, "egress"), # type: ignore[misc]
)
while not sup.tick():
time.sleep(_POLL_INTERVAL)
+15
View File
@@ -15,6 +15,7 @@ from __future__ import annotations
import json
import urllib.error
import urllib.request
from collections.abc import Iterable
from dataclasses import dataclass
from ..paths import host_control_plane_token
@@ -147,6 +148,20 @@ class OrchestratorClient:
raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}")
return True
def reconcile(
self, live_source_ips: Iterable[str], *, grace_seconds: float | None = None,
) -> list[str]:
"""Drop registry rows for bottles that are no longer running
(`POST /reconcile`), returning the reaped bottle ids. `live_source_ips`
is the caller's enumeration of its live bottles — the orchestrator
can't see the backend from inside the infra container."""
body: dict[str, object] = {"live_source_ips": list(live_source_ips)}
if grace_seconds is not None:
body["grace_seconds"] = grace_seconds
payload = self._ok("POST", "/reconcile", body)
reaped = payload.get("reaped")
return [r for r in reaped if isinstance(r, str)] if isinstance(reaped, list) else []
def set_policy(self, bottle_id: str, policy: str) -> bool:
"""Live-reload a bottle's policy (`PUT /bottles/<id>/policy`). False on
404 (unknown bottle)."""
+107
View File
@@ -0,0 +1,107 @@
"""Per-host orchestrator configuration store (settings in bot-bottle.db).
Co-tenants the shared `bot-bottle.db` via the `DbStore` framework. Settings
are readable by the host launch path directly (no HTTP round-trip to the
orchestrator), so they take effect even before the orchestrator is reachable.
"""
from __future__ import annotations
import os
import sqlite3
from pathlib import Path
from ..db_store import DbStore
from ..migrations import TableMigrations
from ..paths import host_db_path
TEARDOWN_TIMEOUT_ENV = "BOT_BOTTLE_ORCHESTRATOR_TEARDOWN_TIMEOUT_SECONDS"
DEFAULT_TEARDOWN_TIMEOUT_SECONDS = 30.0
_MIGRATIONS = TableMigrations(
"orchestrator_config",
[
"""
CREATE TABLE IF NOT EXISTS orchestrator_config (
id INTEGER PRIMARY KEY CHECK (id = 1),
teardown_timeout_seconds REAL
)
""",
],
)
class OrchestratorConfigStore(DbStore):
"""Orchestrator settings in the shared host DB."""
def __init__(self, db_path: Path | None = None) -> None:
super().__init__(db_path or host_db_path(), _MIGRATIONS)
def _connect(self) -> sqlite3.Connection:
conn = super()._connect()
conn.execute("PRAGMA busy_timeout=5000")
return conn
def get_teardown_timeout_seconds(self) -> float | None:
"""Return the configured teardown timeout, or None if not set."""
try:
with self._connection() as conn:
row = conn.execute(
"SELECT teardown_timeout_seconds FROM orchestrator_config WHERE id = 1"
).fetchone()
except sqlite3.OperationalError:
return None
return row["teardown_timeout_seconds"] if row else None
def set_teardown_timeout_seconds(self, value: float) -> None:
"""Persist the teardown timeout."""
with self._connection() as conn:
conn.execute(
"INSERT OR REPLACE INTO orchestrator_config"
" (id, teardown_timeout_seconds) VALUES (1, ?)",
(value,),
)
self._chmod()
def delete_teardown_timeout_seconds(self) -> bool:
"""Clear the stored teardown timeout. Returns True if a value existed."""
with self._connection() as conn:
cur = conn.execute(
"UPDATE orchestrator_config SET teardown_timeout_seconds = NULL"
" WHERE id = 1 AND teardown_timeout_seconds IS NOT NULL"
)
return cur.rowcount > 0
def resolve_teardown_timeout(db_path: Path | None = None) -> float:
"""Return the teardown timeout to use, in priority order:
1. ``BOT_BOTTLE_ORCHESTRATOR_TEARDOWN_TIMEOUT_SECONDS`` env var
2. ``teardown_timeout_seconds`` in the orchestrator config DB
3. ``DEFAULT_TEARDOWN_TIMEOUT_SECONDS`` (30 s)
"""
raw = os.environ.get(TEARDOWN_TIMEOUT_ENV, "").strip()
if raw:
try:
value = float(raw)
if value > 0:
return value
except ValueError:
pass
store = OrchestratorConfigStore(db_path)
if not store.is_migrated():
store.migrate()
db_value = store.get_teardown_timeout_seconds()
if db_value is not None and db_value > 0:
return db_value
return DEFAULT_TEARDOWN_TIMEOUT_SECONDS
__all__ = [
"OrchestratorConfigStore",
"resolve_teardown_timeout",
"TEARDOWN_TIMEOUT_ENV",
"DEFAULT_TEARDOWN_TIMEOUT_SECONDS",
]
+24
View File
@@ -13,6 +13,9 @@ vsock / unix-socket portability caveats):
PUT /bottles/<bottle_id>/policy -> 200 {"updated": true} | 404 (live reload)
body: {"policy"}
DELETE /bottles/<bottle_id> -> 200 {"torn_down": true} | 404 (teardown)
POST /reconcile -> 200 {"reaped": [bottle_id, ...]}
body: {"live_source_ips": [...],
["grace_seconds"]}
POST /attribute -> 200 {"bottle_id"} | 403
POST /resolve -> 200 {"bottle_id","policy"} | 403
body: {"source_ip","identity_token"}
@@ -141,6 +144,27 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
return 200, {"torn_down": True}
return 404, {"error": "no such bottle"}
if method == "POST" and route == "/reconcile":
# Host-driven self-heal: the caller enumerates its live bottles (only
# the host can see the backend) and the orchestrator drops rows for
# every other active bottle. Trusted-caller only — an agent that could
# reach this would be able to unregister its neighbours.
try:
data = _parse_json_object(body)
except ValueError as e:
return 400, {"error": f"invalid JSON: {e}"}
raw_ips = data.get("live_source_ips")
if not isinstance(raw_ips, list):
return 400, {"error": "live_source_ips (list of strings) is required"}
live = [ip for ip in raw_ips if isinstance(ip, str) and ip]
grace = data.get("grace_seconds")
kwargs = (
{"grace_seconds": float(grace)}
if isinstance(grace, (int, float)) and not isinstance(grace, bool)
else {}
)
return 200, {"reaped": orch.reconcile(live, **kwargs)}
if method == "POST" and route == "/attribute":
try:
data = _parse_json_object(body)
+157 -168
View File
@@ -1,17 +1,15 @@
"""Orchestrator + gateway lifecycle (PRD 0070, docker slice).
Runs the orchestrator control plane **as a container** on the shared gateway
network, alongside the gateway container. This is the PRD's "virtualize the
orchestrator": container↔container between the gateway and the orchestrator
avoids the host firewall (which drops containerhost traffic), and the gateway
reaches the control plane by container name over docker DNS. The host CLI
reaches it via a published loopback port.
Runs both the orchestrator control plane and the gateway data plane inside
a single `bot-bottle-infra` container on the shared gateway network
matching the structure already used by the macOS and Firecracker backends.
`gateway_init` is PID 1 and supervises both; the infra container is an
idempotent per-host singleton.
The orchestrator runs with the **register-only broker** the *backend*
launches agent containers (compose), so the orchestrator needs no docker
socket. That keeps this control-plane container unprivileged; the host manages
both containers. `ensure_running` is an idempotent singleton (fixed container
names + the published port).
The combined container replaces the prior two-container split
(bot-bottle-orchestrator + bot-bottle-orch-gateway). The host CLI reaches
the control plane via a published loopback port; gateway daemons reach it
over 127.0.0.1 (same container).
"""
from __future__ import annotations
@@ -26,49 +24,70 @@ from pathlib import Path
from .. import log
from ..docker_cmd import run_docker
from ..paths import CONTROL_PLANE_TOKEN_ENV, bot_bottle_root, host_control_plane_token
from .gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK, DockerGateway, GatewayError
from ..supervise import DB_PATH_IN_CONTAINER
from .gateway import (
GATEWAY_CA_VOLUME,
GATEWAY_DOCKERFILE,
GATEWAY_IMAGE,
GATEWAY_NETWORK,
GatewayError,
MITMPROXY_HOME,
_host_db_dir,
)
DEFAULT_PORT = 8099
ORCHESTRATOR_NAME = "bot-bottle-orchestrator"
ORCHESTRATOR_LABEL = "bot-bottle-orchestrator=1"
# The control-plane's own runtime image — lean (python + the stdlib-only
# `bot_bottle` package, bind-mounted at run time), distinct from the heavy
# gateway data-plane image it used to borrow (#384). Env override for
# operators pinning a published build.
DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
INFRA_NAME = "bot-bottle-infra"
INFRA_LABEL = "bot-bottle-infra=1"
# The combined infra image: gateway data plane + orchestrator content.
# Built from Dockerfile.infra (FROM gateway + COPY --from orchestrator).
INFRA_IMAGE = os.environ.get("BOT_BOTTLE_INFRA_IMAGE", "bot-bottle-infra:latest")
INFRA_DOCKERFILE = "Dockerfile.infra"
# Baked as a container label so `ensure_running` can detect whether the
# running container is executing the current bind-mounted source.
INFRA_SOURCE_HASH_LABEL = "bot-bottle-infra-source-hash"
# Orchestrator image: the single canonical definition of the control-plane
# content (lean: python:3.12-slim + bot_bottle package, no mitmproxy/git).
# Used as a build intermediate: `Dockerfile.infra` COPY --from this image.
ORCHESTRATOR_IMAGE = os.environ.get(
"BOT_BOTTLE_ORCHESTRATOR_IMAGE", "bot-bottle-orchestrator:latest"
)
ORCHESTRATOR_DOCKERFILE = "Dockerfile.orchestrator"
# Baked onto the container as a label so `ensure_running` can tell whether the
# running process is executing the *current* bind-mounted source — see
# `source_hash`.
ORCHESTRATOR_SOURCE_HASH_LABEL = "bot-bottle-orchestrator-source-hash"
# The repo root is bind-mounted into the control-plane container so
# `python -m bot_bottle.orchestrator` resolves the package (the orchestrator
# is stdlib-only, so the lean orchestrator image's python is enough).
_REPO_ROOT = Path(__file__).resolve().parents[2]
_APP_DIR = "/app"
# The gateway daemons + orchestrator the infra container runs.
# BOT_BOTTLE_GATEWAY_DAEMONS listing `orchestrator` opts it in to
# gateway_init's supervise tree (see gateway_init._OPT_IN_DAEMONS).
_INFRA_DAEMONS = "egress,git-http,supervise,orchestrator"
# The bind-mount path for the live control-plane source inside the
# container. Separate from /app so the gateway's baked scripts
# (egress_addon.py, egress-entrypoint.sh) are not overlaid.
_SRC_IN_CONTAINER = "/bot-bottle-src"
# Bot-bottle host-root bind-mount inside the container (DB + state).
_ROOT_IN_CONTAINER = "/bot-bottle-root"
# The supervise daemon writes proposals into the host DB directory.
_SUPERVISE_DB_DIR_IN_CONTAINER = os.path.dirname(DB_PATH_IN_CONTAINER)
_HEALTH_POLL_SECONDS = 0.25
DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
_REPO_ROOT = Path(__file__).resolve().parents[2]
class OrchestratorStartError(RuntimeError):
"""The orchestrator container did not become healthy within the timeout."""
"""The infra container did not become healthy within the timeout."""
def source_hash(repo_root: Path) -> str:
"""Content hash of the orchestrator's bind-mounted Python source (the
`bot_bottle` package the control-plane process imports). This only
changes when the code that would actually run inside the container
changes `ensure_running` recreates the container on a mismatch and
otherwise leaves a healthy one alone, so a bottle launch that isn't
accompanied by a code change doesn't restart the process and drop every
*other* active bottle's in-memory egress tokens (`Orchestrator._tokens`
in `service.py`, never persisted to disk by design)."""
`bot_bottle` package the control-plane process imports). Changes only
when the code that would actually run changes `ensure_running`
recreates the container on a mismatch so a code change takes effect,
but leaves a healthy up-to-date container alone to preserve in-memory
egress tokens."""
h = hashlib.sha256()
for path in sorted((repo_root / "bot_bottle").rglob("*.py")):
h.update(str(path.relative_to(repo_root)).encode())
@@ -77,57 +96,37 @@ def source_hash(repo_root: Path) -> str:
class OrchestratorService:
"""Manages the orchestrator control-plane container + the shared gateway.
"""Manages the single per-host infra container (control plane + gateway).
Callers only need `ensure_running()` + `url`.
`orchestrator_name` / `orchestrator_label` let backends run independent
orchestrators on the same host without name collisions (e.g. the
Firecracker backend uses `bot-bottle-fc-orchestrator` alongside the Docker
backend's `bot-bottle-orchestrator`); `gateway_name` gives the paired
gateway container the same treatment (e.g. isolated integration tests
that can't share the production `GATEWAY_NAME` singleton). Subclass and
override `_gateway()` for anything `_gateway_image`/`gateway_name` can't
express (a genuinely backend-specific gateway variant)."""
`infra_name` / `infra_label` let backends run independent infra containers
on the same host without name collisions (e.g. isolated integration tests
that can't share the production INFRA_NAME singleton)."""
def __init__(
self,
*,
port: int = DEFAULT_PORT,
network: str = GATEWAY_NETWORK,
image: str = ORCHESTRATOR_IMAGE,
gateway_image: str = GATEWAY_IMAGE,
gateway_name: str = GATEWAY_NAME,
image: str = INFRA_IMAGE,
repo_root: Path = _REPO_ROOT,
host_root: Path | None = None,
orchestrator_name: str = ORCHESTRATOR_NAME,
orchestrator_label: str = ORCHESTRATOR_LABEL,
infra_name: str = INFRA_NAME,
infra_label: str = INFRA_LABEL,
) -> None:
self.port = port
self.network = network
# Two distinct images (#384): `image` is the lean control-plane
# runtime this container runs; `_gateway_image` is the heavy egress /
# git-gate / supervise data plane the gateway container runs. They
# were one conflated image before the split.
self.image = image
self._gateway_image = gateway_image
self._gateway_name = gateway_name
self._repo_root = repo_root
self._host_root = host_root or bot_bottle_root()
self._orchestrator_name = orchestrator_name
self._orchestrator_label = orchestrator_label
self._infra_name = infra_name
self._infra_label = infra_label
@property
def url(self) -> str:
"""Host-side control-plane URL (published loopback port)."""
return f"http://127.0.0.1:{self.port}"
@property
def internal_url(self) -> str:
"""Control-plane URL as the gateway container reaches it — by name over
docker DNS on the shared network. This is the gateway's
BOT_BOTTLE_ORCHESTRATOR_URL."""
return f"http://{self._orchestrator_name}:{self.port}"
def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool:
try:
with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp:
@@ -139,139 +138,129 @@ class OrchestratorService:
proc = run_docker(["docker", "ps", "--filter", f"name=^/{name}$", "--format", "{{.Names}}"])
return name in proc.stdout.split()
def _run_orchestrator_container(self, current_hash: str) -> None:
"""Start the control-plane container (idempotent: clears a stale
fixed-name container first). Register-only broker no docker socket.
Labels the container with `current_hash` so a later `ensure_running`
can detect a real code change (see `source_hash`)."""
run_docker(["docker", "rm", "--force", self._orchestrator_name])
proc = run_docker([
"docker", "run", "--detach",
"--name", self._orchestrator_name,
"--label", self._orchestrator_label,
"--label", f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current_hash}",
"--network", self.network,
# Host CLI reaches the control plane here; bound to loopback so it
# is not exposed on the host's external interfaces. NOTE: the
# container is still on `self.network` (the shared gateway network),
# so agents can reach it by container IP — which is exactly why the
# control plane requires the secret below rather than trusting the
# network boundary.
"--publish", f"127.0.0.1:{self.port}:{self.port}",
"--volume", f"{self._repo_root}:{_APP_DIR}:ro",
"--workdir", _APP_DIR,
# Persist the registry DB on the host (sole-owner: only the
# orchestrator opens bot-bottle.db).
"--volume", f"{self._host_root}:{_ROOT_IN_CONTAINER}",
"--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}",
# The control-plane secret it requires on every route but /health.
# Bare `--env NAME` → docker inherits the value from the run env
# below, so the secret never lands on argv / `docker inspect`.
"--env", CONTROL_PLANE_TOKEN_ENV,
"--entrypoint", "python3",
self.image,
"-m", "bot_bottle.orchestrator",
"--host", "0.0.0.0", "--port", str(self.port), "--broker", "stub",
], env={**os.environ, CONTROL_PLANE_TOKEN_ENV: host_control_plane_token()})
if proc.returncode != 0:
raise OrchestratorStartError(
f"orchestrator container failed to start: {proc.stderr.strip()}"
)
def _gateway(self) -> DockerGateway:
return DockerGateway(
self._gateway_image,
name=self._gateway_name,
network=self.network,
orchestrator_url=self.internal_url,
)
def _ensure_orchestrator_image(self) -> None:
"""Build the lean control-plane image from `Dockerfile.orchestrator`
when it's missing (#384). Cheap — a `FROM python:*-slim` base with no
deps to install, so the layer cache makes rebuilds a no-op. Unlike the
gateway image this is build-if-missing, not build-every-time: the
control plane bind-mounts its source, so a code change is caught by the
source-hash recreate (below), not by an image rebuild."""
if run_docker(["docker", "image", "inspect", self.image]).returncode == 0:
return
argv = ["docker", "build", "-t", self.image,
"-f", str(self._repo_root / ORCHESTRATOR_DOCKERFILE),
str(self._repo_root)]
if os.environ.get("BOT_BOTTLE_NO_CACHE"):
argv.insert(2, "--no-cache")
proc = run_docker(argv)
if proc.returncode != 0:
raise GatewayError(
f"orchestrator image build failed: {proc.stderr.strip()}"
)
def _orchestrator_source_current(self, current_hash: str) -> bool:
"""True iff the running orchestrator container was created from the
*current* bind-mounted source. Mirrors `DockerGateway`'s
image-staleness check, but by content hash rather than image id since
the orchestrator runs bind-mounted source, not a built image."""
if not self._container_running(self._orchestrator_name):
def _infra_source_current(self, current_hash: str) -> bool:
"""True iff the running infra container was started from the current
bind-mounted source. Mirrors the macOS backend's `_source_current`."""
if not self._container_running(self._infra_name):
return False
proc = run_docker([
"docker", "inspect", "--format",
"{{ index .Config.Labels \"" + ORCHESTRATOR_SOURCE_HASH_LABEL + "\" }}",
self._orchestrator_name,
"{{ index .Config.Labels \"" + INFRA_SOURCE_HASH_LABEL + "\" }}",
self._infra_name,
])
if proc.returncode != 0:
return True # can't compare -> don't churn a working container
return True # can't compare don't churn a working container
return proc.stdout.strip() == current_hash
def _ensure_network(self) -> None:
if run_docker(["docker", "network", "inspect", self.network]).returncode == 0:
return
proc = run_docker(["docker", "network", "create", self.network])
if proc.returncode != 0 and "already exists" not in proc.stderr:
raise GatewayError(
f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
)
def _build_images(self) -> None:
"""Build the gateway base, the orchestrator intermediate, then the
infra image. All are cache-aware: a no-op when nothing changed."""
for tag, dockerfile in (
(GATEWAY_IMAGE, GATEWAY_DOCKERFILE),
(ORCHESTRATOR_IMAGE, ORCHESTRATOR_DOCKERFILE),
(self.image, INFRA_DOCKERFILE),
):
argv = ["docker", "build", "-t", tag,
"-f", str(self._repo_root / dockerfile),
str(self._repo_root)]
if os.environ.get("BOT_BOTTLE_NO_CACHE"):
argv.insert(2, "--no-cache")
proc = run_docker(argv)
if proc.returncode != 0:
raise GatewayError(f"{dockerfile} build failed: {proc.stderr.strip()}")
def _run_infra_container(self, current_hash: str) -> None:
"""Start the combined infra container (idempotent: clears a stale
fixed-name container first). Labels the container with `current_hash`
so a later `ensure_running` can detect a real code change."""
self._ensure_network()
run_docker(["docker", "rm", "--force", self._infra_name])
proc = run_docker([
"docker", "run", "--detach",
"--name", self._infra_name,
"--label", self._infra_label,
"--label", f"{INFRA_SOURCE_HASH_LABEL}={current_hash}",
"--network", self.network,
# Host CLI reaches the control plane here (loopback only).
# gateway_init always starts the orchestrator on DEFAULT_PORT (8099)
# inside the container; self.port is the host-side published port.
"--publish", f"127.0.0.1:{self.port}:{DEFAULT_PORT}",
# Persist the mitmproxy CA so it survives container recreation.
"--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
# Shared supervise DB (same file the operator reads over HTTP).
"--volume", f"{_host_db_dir()}:{_SUPERVISE_DB_DIR_IN_CONTAINER}",
"--env", f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
# Live control-plane source, mounted to a path that does not
# overlay the gateway's baked /app scripts.
"--volume", f"{self._repo_root}:{_SRC_IN_CONTAINER}:ro",
# PYTHONPATH lets the orchestrator (and other Python daemons)
# import the live source ahead of the installed package.
"--env", f"PYTHONPATH={_SRC_IN_CONTAINER}",
# Orchestrator registry DB on the host (sole writer: control plane).
"--volume", f"{self._host_root}:{_ROOT_IN_CONTAINER}",
"--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}",
# Control-plane secret: required by the orchestrator (to enforce)
# and by the gateway daemons (to present on /resolve calls).
"--env", CONTROL_PLANE_TOKEN_ENV,
# Gateway daemons reach the orchestrator over loopback at its
# fixed internal port (DEFAULT_PORT), independent of self.port.
"--env", f"BOT_BOTTLE_ORCHESTRATOR_URL=http://127.0.0.1:{DEFAULT_PORT}",
# Opt the orchestrator into gateway_init's supervise tree.
"--env", f"BOT_BOTTLE_GATEWAY_DAEMONS={_INFRA_DAEMONS}",
self.image,
], env={**os.environ, CONTROL_PLANE_TOKEN_ENV: host_control_plane_token()})
if proc.returncode != 0:
raise OrchestratorStartError(
f"infra container failed to start: {proc.stderr.strip()}"
)
def ensure_running(
self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
) -> str:
"""Ensure the control plane + shared gateway are up; return the host
control-plane URL. Idempotent a healthy control plane running
current code and a running gateway are left untouched. Raises
`OrchestratorStartError` on timeout."""
gateway = self._gateway()
gateway.ensure_built() # rebuild the bundle image on a source change
gateway.ensure_running() # creates the shared network + (re)starts gateway
"""Ensure the infra container (control plane + gateway) is up; return
the host control-plane URL. Idempotent a healthy container on current
source is left untouched. Raises `OrchestratorStartError` on timeout."""
self._build_images()
# Recreate the orchestrator container only when its bind-mounted
# source has actually changed since it started — its Python process
# loaded that code at startup and won't reload, so a stale container
# would keep running OLD control-plane code. Recreating on *every*
# launch (the prior behaviour) would drop every other active
# bottle's in-memory egress tokens each time a new bottle starts,
# since the orchestrator process holds them only in memory (#381).
current_hash = source_hash(self._repo_root)
if self.is_healthy() and self._orchestrator_source_current(current_hash):
if self.is_healthy() and self._infra_source_current(current_hash):
return self.url
self._ensure_orchestrator_image()
log.info(
"starting orchestrator container",
context={"name": self._orchestrator_name},
)
self._run_orchestrator_container(current_hash)
log.info("starting infra container", context={"name": self._infra_name})
self._run_infra_container(current_hash)
deadline = time.monotonic() + startup_timeout
while time.monotonic() < deadline:
if self.is_healthy():
log.info("orchestrator healthy", context={"url": self.url})
log.info("infra container healthy", context={"url": self.url})
return self.url
time.sleep(_HEALTH_POLL_SECONDS)
raise OrchestratorStartError(
f"orchestrator at {self.url} did not become healthy within {startup_timeout:g}s"
f"infra container at {self.url} did not become healthy within {startup_timeout:g}s"
)
def stop(self) -> None:
"""Remove the orchestrator + gateway containers (idempotent)."""
run_docker(["docker", "rm", "--force", self._orchestrator_name])
self._gateway().stop()
"""Remove the infra container (idempotent)."""
run_docker(["docker", "rm", "--force", self._infra_name])
__all__ = [
"OrchestratorService",
"OrchestratorStartError",
"ORCHESTRATOR_NAME",
"INFRA_NAME",
"INFRA_IMAGE",
"INFRA_SOURCE_HASH_LABEL",
"ORCHESTRATOR_IMAGE",
"DEFAULT_PORT",
"DEFAULT_STARTUP_TIMEOUT_SECONDS",
"source_hash",
]
+72
View File
@@ -32,6 +32,7 @@ import hmac
import secrets
import sqlite3
import time
from collections.abc import Iterable
from dataclasses import dataclass
from pathlib import Path
@@ -42,6 +43,12 @@ from ..paths import host_db_path
# 256 bits of urandom, URL-safe — unguessable per-bottle identity token.
IDENTITY_TOKEN_BYTES = 32
# How recently a row must have been registered to be exempt from
# `reap_absent`. Covers the window between `container run` and the address
# becoming visible to another launch's enumeration, so reconciliation never
# reaps a bottle that is still coming up.
DEFAULT_REAP_GRACE_SECONDS = 120.0
def new_identity_token() -> str:
"""A fresh per-bottle identity token (PRD 0070 attribution defence)."""
@@ -225,6 +232,70 @@ class RegistryStore(DbStore):
).fetchall()
return [_row_to_record(r) for r in rows]
def reap_absent(
self,
live_source_ips: Iterable[str],
*,
grace_seconds: float = DEFAULT_REAP_GRACE_SECONDS,
now: float | None = None,
) -> list[BottleRecord]:
"""Delete active rows whose source IP is not held by a live bottle.
A row only ever leaves the registry two ways: an explicit
`teardown_bottle` (the launcher's cleanup callback) or the supersede
sweep in `register`. Neither runs when the launching CLI dies hard
SIGKILL, a closed terminal, a host sleep/crash so the row outlives
its container. That orphan is not inert: source IPs are recycled by
the backend's DHCP, and `by_source_ip` fail-closes on ambiguity, so a
leftover row at a reused address can brick the *next* bottle that
lands on it (no policy resolved -> every host denied, reported to the
agent as "not in the allowlist"). Reconciling against the live set at
launch keeps the registry from accumulating those landmines.
Restores the invariant the data plane needs: **at most one active row
per live address, and none at all for a dead one.** Two cases, because
a dead bottle's address may already have been handed to a live one:
* no live bottle holds the address every row there is an orphan;
* a live bottle holds it but several rows claim it the newest
registration is authoritative and the rest are orphans, the same
rule `register`'s same-IP supersede sweep applies. Without this
second case a recycled address stays ambiguous, which is exactly
the state that resolves no policy.
`grace_seconds` protects an in-flight launch: registration happens
moments after `container run`, and a concurrent launch's address may
not be visible to the caller's enumeration yet. Rows younger than the
grace window are never reaped, so reconciliation can't race a bottle
that is still coming up. Returns the deleted records."""
live = {ip for ip in live_source_ips if ip}
cutoff = (time.time() if now is None else now) - grace_seconds
with self._connection() as conn:
rows = conn.execute(
"SELECT * FROM orchestrator_bottles WHERE state = 'active'",
).fetchall()
by_ip: dict[str, list[BottleRecord]] = {}
for row in rows:
rec = _row_to_record(row)
by_ip.setdefault(rec.source_ip, []).append(rec)
candidates: list[BottleRecord] = []
for ip, recs in by_ip.items():
if ip not in live:
candidates.extend(recs)
continue
# Keep the newest claim on a live address; supersede the rest.
recs.sort(key=lambda r: r.created_at)
candidates.extend(recs[:-1])
doomed = [r for r in candidates if r.created_at <= cutoff]
for rec in doomed:
conn.execute(
"DELETE FROM orchestrator_bottles WHERE bottle_id = ?",
(rec.bottle_id,),
)
if doomed:
self._chmod()
return doomed
def by_source_ip(self, source_ip: str) -> BottleRecord | None:
"""Network-layer attribution: the single active bottle at this source
IP, or None if unknown or ambiguous (more than one a
@@ -262,4 +333,5 @@ __all__ = [
"new_identity_token",
"default_db_path",
"IDENTITY_TOKEN_BYTES",
"DEFAULT_REAP_GRACE_SECONDS",
]
+30 -1
View File
@@ -13,15 +13,20 @@ Launch lifecycle:
and returns the record. If the broker rejects/fails, the registry entry
is rolled back so a failed launch leaves no orphan.
* `teardown_bottle` sends a signed teardown request, then deregisters.
* `reconcile` sweeps rows whose bottle is no longer running the
self-heal for the teardown paths that never got to run (a hard-killed
launcher), since an orphan row at a recycled source IP bricks the next
bottle that lands on it.
"""
from __future__ import annotations
import json
from collections.abc import Iterable
from datetime import datetime, timezone
from .broker import LaunchBroker, LaunchRequest, sign_request
from .registry import BottleRecord, RegistryStore
from .registry import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore
from .gateway import Gateway
from ..supervise import (
AuditEntry,
@@ -117,6 +122,30 @@ class Orchestrator:
self._tokens.pop(bottle_id, None)
return True
def reconcile(
self,
live_source_ips: Iterable[str],
*,
grace_seconds: float = DEFAULT_REAP_GRACE_SECONDS,
) -> list[str]:
"""Drop registry rows for bottles that are no longer running, and
forget their in-memory egress tokens. Returns the reaped bottle ids.
The caller supplies the live set because only the host can enumerate
its own containers the orchestrator runs *inside* the infra
container and has no view of the backend. Deliberately does not
broker a teardown: the container is already gone, so there is nothing
to stop, and a broker error must not stop the sweep from clearing
the row that would otherwise brick the next bottle at that address.
See `RegistryStore.reap_absent` for why orphans accumulate and why
they are harmful rather than merely untidy."""
reaped = self.registry.reap_absent(
live_source_ips, grace_seconds=grace_seconds)
for rec in reaped:
self._tokens.pop(rec.bottle_id, None)
return [rec.bottle_id for rec in reaped]
def tokens_for(self, bottle_id: str) -> dict[str, str]:
"""The bottle's in-memory egress auth tokens (env_name -> value), or
empty. The gateway injects these per request; they are never
@@ -0,0 +1,156 @@
# PRD prd-new: Consolidate infra backend for Docker
- **Status:** Active
- **Author:** Claude
- **Created:** 2026-07-20
- **Issue:** #431
## Summary
The Docker backend runs two containers — `bot-bottle-orch-gateway` (gateway
data plane) and `bot-bottle-orchestrator` (control plane) — where the
macOS and Firecracker backends already run a single combined infra
unit. This PRD collapses Docker to the same model: one `bot-bottle-infra`
container running both processes under the `gateway_init` supervise tree, a
restructured `Dockerfile.infra` as the shared gateway+orchestrator base,
and a handful of extracted shared utilities (CA cert polling, teardown
sequence, launch skeleton) that are currently duplicated across all three
`consolidated_launch.py` files.
## Goals / success criteria
- Docker backend starts exactly one infra container instead of two.
- `Dockerfile.infra` is the shared base image (gateway + orchestrator, no
buildah); the Firecracker image layers buildah on top of it.
- The orchestrator process runs under the `gateway_init` supervise tree
inside the combined container (one PID-1, one restart/health surface).
- CA cert polling, the teardown sequence, and the shared launch skeleton
(ensure-infra → register → provision → return context) live in a single
shared module; all three backends import from it.
- No functional change to macOS or Firecracker launch paths.
## Non-goals
- Changing per-bottle isolation — agents stay one-VM/container-each.
- Consolidating transport implementations (`DockerGatewayTransport`,
`AppleGatewayTransport`, `SshGatewayTransport`) — these are already the
right abstraction boundary.
- macOS DHCP-inversion of registration order — irreducible backend
difference, stays as-is.
- Any changes to the orchestrator RPC protocol or the attribution model.
## Design
### Dockerfile restructuring
**Current shape:**
- `Dockerfile.gateway` — data plane (mitmproxy, gitleaks, git, openssh,
supervise daemons)
- `Dockerfile.orchestrator` — control plane (python:3.12-slim + bot_bottle
package; stdlib-only, no third-party deps)
- `Dockerfile.infra` — Firecracker only: `FROM bot-bottle-gateway` +
buildah + `COPY --from bot-bottle-orchestrator`
**New shape:**
- `Dockerfile.gateway` — unchanged
- `Dockerfile.orchestrator` — unchanged (single definition of orchestrator
content; both Docker infra and Firecracker infra `COPY --from` it)
- `Dockerfile.infra`**shared base**: `FROM bot-bottle-gateway` + `COPY
--from bot-bottle-orchestrator` (no buildah — Docker infra image)
- `Dockerfile.infra.fc` — Firecracker only: `FROM bot-bottle-infra` +
buildah/crun/netavark/aardvark-dns (layered on the shared base, same net
result as today)
The comment in `Dockerfile.infra` that says "the docker backend keeps
orchestrator + gateway as separate images; this combined image exists only
for the Firecracker single-VM cut" is removed.
### Orchestrator in the supervise tree
`gateway_init` already supervises the data-plane daemons (egress, git-http,
supervise-MCP). The orchestrator control plane is added as another supervised
process: `python3 -m bot_bottle.orchestrator --host 0.0.0.0 --port <port>
--broker stub`.
The orchestrator source is bind-mounted (`/app` → repo root, as today) so
dev live-reload still works. `source_hash`-based container recreation in
`OrchestratorService.ensure_running` continues to apply — a code change
recreates the combined infra container, which bounces both gateway and
orchestrator. This is acceptable: the docker backend is a dev/legacy target
where in-flight egress connections across a code deploy are not a hard
requirement.
### `OrchestratorService` changes
`OrchestratorService` currently starts two containers in sequence: gateway
first (`DockerGateway.ensure_running`), then orchestrator. After this PRD:
- Single `docker run` of `bot-bottle-infra:latest`
- Container name: `bot-bottle-infra` (replaces `bot-bottle-orch-gateway` +
`bot-bottle-orchestrator`)
- Published ports: `127.0.0.1:{host_port}:8099` for the control plane
(`gateway_init` listens on a fixed internal port 8099; the caller-chosen
host port maps to it)
- Bind mounts: repo root + host root (same as today)
- `DockerGateway` becomes an implementation detail of `OrchestratorService`
rather than a separately started container; the gateway image name
(`GATEWAY_IMAGE`) is no longer referenced at runtime, only at build time
for the `Dockerfile.infra` base
The `_gateway()` / `ensure_running` two-step in `OrchestratorService` is
replaced by a single `_run_infra_container()`.
### Shared backend utilities
Three items are duplicated across
`backend/docker/consolidated_launch.py`,
`backend/macos_container/consolidated_launch.py`, and
`backend/firecracker/consolidated_launch.py`:
1. **CA cert polling loop** — `deadline = time.monotonic() + timeout; while
...: try fetch CA; sleep` — extracted to
`backend/consolidated_util.py:poll_ca_cert(transport, *, timeout)`.
2. **Teardown sequence**`OrchestratorClient(url).teardown_bottle(id)` +
`deprovision_git_gate(transport, id)` — extracted to
`backend/consolidated_util.py:teardown_consolidated(url, transport,
bottle_id)`.
3. **Launch skeleton** — all three follow: ensure-infra → allocate/register
→ provision git-gate → fetch CA cert → return launch context. The macOS
inversion (agent starts before registration, source IP from DHCP) is the
only deviation. Extract a shared `_provision_bottle(transport, bottle_id,
plan, orchestrator_url)` helper covering the register → provision →
return-token steps; the backends keep their own `launch_consolidated`
wrappers for the before/after (infra-ensure + agent-start + IP
allocation), calling the shared helper.
The new `backend/consolidated_util.py` module holds only backend-neutral,
transport-agnostic logic. All three backends import from it.
## Implementation chunks
1. **(this PR)** Dockerfile restructuring: rename current `Dockerfile.infra`
content to `Dockerfile.infra.fc`; write new `Dockerfile.infra` as
gateway+orchestrator base. Update Firecracker image-build references from
`Dockerfile.infra``Dockerfile.infra.fc`.
2. Add orchestrator process to `gateway_init` supervise tree.
3. Collapse `OrchestratorService` to a single-container start; rename
container from `bot-bottle-orch-gateway`/`bot-bottle-orchestrator`
`bot-bottle-infra`; update image name constant.
4. Extract `backend/consolidated_util.py` with `poll_ca_cert`,
`teardown_consolidated`, and `_provision_bottle`; update all three
`consolidated_launch.py` files to import from it.
5. Update tests that reference the old container names or two-container
startup sequence.
## Open questions
None — the supervise-tree approach and shared Dockerfile layering were
confirmed in issue #431.
@@ -34,6 +34,7 @@ from tests._docker import skip_unless_docker
# image instead of leaking a new dangling tag on every invocation.
_TEST_ORCHESTRATOR_IMAGE = "bot-bottle-orchestrator:itest"
_TEST_GATEWAY_IMAGE = "bot-bottle-gateway:itest"
_TEST_INFRA_IMAGE = "bot-bottle-infra:itest"
@skip_unless_docker()
@@ -69,20 +70,17 @@ class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
os.environ["BOT_BOTTLE_ROOT"] = cls._tmp.name
cls.addClassCleanup(_restore_root)
orchestrator_name = f"bot-bottle-orch-itest-{suffix}"
gateway_name = f"bot-bottle-gw-itest-{suffix}"
infra_name = f"bot-bottle-infra-itest-{suffix}"
network = f"bot-bottle-net-itest-{suffix}"
host_root = Path(cls._tmp.name)
cls.addClassCleanup(
cls._teardown_docker, orchestrator_name, gateway_name, network, host_root
cls._teardown_docker, infra_name, network, host_root
)
cls.svc = OrchestratorService(
orchestrator_name=orchestrator_name,
gateway_name=gateway_name,
infra_name=infra_name,
network=network,
image=_TEST_ORCHESTRATOR_IMAGE,
gateway_image=_TEST_GATEWAY_IMAGE,
image=_TEST_INFRA_IMAGE,
port=20000 + secrets.randbelow(10000),
host_root=host_root,
)
@@ -91,23 +89,23 @@ class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
@staticmethod
def _teardown_docker(
orchestrator_name: str, gateway_name: str, network: str, host_root: Path
infra_name: str, network: str, host_root: Path
) -> None:
subprocess.run(
["docker", "rm", "--force", orchestrator_name, gateway_name],
["docker", "rm", "--force", infra_name],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
subprocess.run(
["docker", "network", "rm", network],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
# The orchestrator container (no USER directive) wrote the registry
# The infra container (no USER directive) wrote the registry
# DB as root into the throwaway host_root; chown it back so the
# (non-root) tempdir cleanup can remove it. Same workaround
# test_multitenant_isolation.py uses for the identical bind mount.
subprocess.run(
["docker", "run", "--rm", "-v", f"{host_root}:/r",
"--entrypoint", "chown", _TEST_GATEWAY_IMAGE, "-R",
"--entrypoint", "chown", _TEST_INFRA_IMAGE, "-R",
f"{os.getuid()}:{os.getgid()}", "/r"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
+29
View File
@@ -0,0 +1,29 @@
"""Unit: shared cross-backend helpers in backend/util.py."""
from __future__ import annotations
import unittest
from unittest.mock import patch
from bot_bottle.backend import util as backend_util
class TestPollCaCert(unittest.TestCase):
def test_returns_pem_on_first_success(self) -> None:
result = backend_util.poll_ca_cert(lambda: "PEM", timeout=1.0)
self.assertEqual("PEM", result)
def test_raises_timeout_error_when_cert_never_appears(self) -> None:
with self.assertRaises(TimeoutError):
backend_util.poll_ca_cert(lambda: None, timeout=0.0)
def test_polls_until_cert_appears(self) -> None:
responses = iter([None, None, "-----BEGIN CERTIFICATE-----\n"])
with patch("bot_bottle.backend.util.time.sleep") as mock_sleep:
result = backend_util.poll_ca_cert(lambda: next(responses), timeout=5.0)
self.assertTrue(result.startswith("-----BEGIN CERTIFICATE-----"))
self.assertEqual(2, mock_sleep.call_count)
if __name__ == "__main__":
unittest.main()
+4 -3
View File
@@ -15,6 +15,7 @@ from bot_bottle.git_gate import GitGatePlan
from bot_bottle.orchestrator.client import RegisteredBottle
_MOD = "bot_bottle.backend.docker.consolidated_launch"
_UTIL = "bot_bottle.backend.consolidated_util"
def _egress_plan() -> EgressPlan:
@@ -49,7 +50,7 @@ class TestLaunchConsolidated(unittest.TestCase):
patch(f"{_MOD}._container_ip", return_value="172.18.0.2"), \
patch(f"{_MOD}._network_container_ips", return_value=list(on_network)), \
patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate", provision or Mock()):
patch(f"{_UTIL}.provision_git_gate", provision or Mock()):
return launch_consolidated(_egress_plan(), _git_plan(), service=service)
def test_allocates_ip_registers_and_provisions(self) -> None:
@@ -84,8 +85,8 @@ class TestLaunchConsolidated(unittest.TestCase):
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:
with patch(f"{_UTIL}.OrchestratorClient", return_value=client), \
patch(f"{_UTIL}.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()
+60 -1
View File
@@ -4,7 +4,14 @@ from __future__ import annotations
import unittest
from bot_bottle.egress_addon_core import resolve_client_config, resolve_client_context
from bot_bottle.egress_addon_core import (
DENY_RESOLVER_ERROR,
DENY_UNATTRIBUTED,
DENY_UNPARSEABLE,
decide,
resolve_client_config,
resolve_client_context,
)
from bot_bottle.policy_resolver import PolicyResolveError
@@ -108,3 +115,55 @@ class TestResolveClientContext(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestDenyReasonNamesTheRealFault(unittest.TestCase):
"""A deny-all must not masquerade as a missing allowlist entry.
Regression: an unregistered bottle resolves no policy, so *every* host is
denied but the block message said `host X is not in the allowlist`,
which reads as a config problem and sends the operator hunting for a route
that was never missing. The structural reason wins over that wording.
"""
def _reason(self, resolver: object, host: str = "chatgpt.com") -> str:
cfg = resolve_client_config(resolver, "10.243.0.1") # type: ignore[arg-type]
return decide(cfg.routes, host, "/v1/x", {}, deny_reason=cfg.deny_reason).reason
def test_unattributed_says_unattributed_not_allowlist(self) -> None:
reason = self._reason(_FakeResolver(result=None))
self.assertEqual(DENY_UNATTRIBUTED, reason)
# The misleading claim is the one that must be gone: the host was
# never "not in the allowlist" — there was no allowlist at all.
self.assertNotIn("is not in the bottle's egress.routes allowlist", reason)
# Both causes must be named. `/resolve` fail-closes on a missing row
# *and* on a token mismatch, and the message pointing only at the row
# sent us hunting for a deregistered bottle that was registered fine.
self.assertIn("registry row", reason)
self.assertIn("identity token", reason)
def test_resolver_error_says_orchestrator_unreachable(self) -> None:
self.assertEqual(DENY_RESOLVER_ERROR, self._reason(_FakeResolver(raises=True)))
def test_unparseable_policy_says_so(self) -> None:
self.assertEqual(
DENY_UNPARSEABLE, self._reason(_FakeResolver(result="routes: notalist\n")))
def test_a_real_allowlist_miss_keeps_the_allowlist_wording(self) -> None:
"""The message only changes for structural deny-alls — a loaded policy
that genuinely lacks the host still points at the allowlist."""
reason = self._reason(_FakeResolver(result='routes:\n - host: "api.example.com"\n'))
self.assertIn("is not in the bottle's egress.routes allowlist", reason)
self.assertIn("chatgpt.com", reason)
def test_allowed_host_is_still_forwarded(self) -> None:
cfg = resolve_client_config(
_FakeResolver(result='routes:\n - host: "api.example.com"\n'), "10.243.0.1")
decision = decide(
cfg.routes, "api.example.com", "/v1/x", {}, deny_reason=cfg.deny_reason)
self.assertEqual("forward", decision.action)
def test_a_parsed_policy_carries_no_deny_reason(self) -> None:
cfg = resolve_client_config(
_FakeResolver(result='routes:\n - host: "api.example.com"\n'), "10.243.0.1")
self.assertEqual("", cfg.deny_reason)
+10
View File
@@ -71,6 +71,16 @@ class TestSshGatewayTransport(unittest.TestCase):
t.exec(["mkdir", "-p", "/git-gate"])
class TestGatewayCaPem(unittest.TestCase):
def test_dies_when_cert_never_appears(self) -> None:
from subprocess import CompletedProcess
infra = infra_vm.InfraVm(vm=None, guest_ip="10.0.0.1", private_key=Path("/k"))
with patch.object(infra_vm.subprocess, "run",
return_value=CompletedProcess([], 1, stdout="", stderr="")), \
self.assertRaises(SystemExit):
infra.gateway_ca_pem(timeout=0)
class TestRegistryVolume(unittest.TestCase):
def test_reuses_existing_volume(self):
import tempfile
+1 -1
View File
@@ -96,7 +96,7 @@ class TestVersionInputs(unittest.TestCase):
(pkg / "app.py").write_text("print('hi')\n")
(pkg / "egress_entrypoint.sh").write_text("#!/bin/sh\nexec mitmdump\n")
(pkg / "netpool.defaults.env").write_text("FOO=1\n")
for name in ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra"):
for name in ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra", "Dockerfile.infra.fc"):
(root / name).write_text(f"FROM scratch # {name}\n")
(root / "pyproject.toml").write_text("[project]\nname = 'bot-bottle'\n")
+106 -3
View File
@@ -17,6 +17,7 @@ from bot_bottle.git_gate import GitGatePlan
from bot_bottle.orchestrator.client import RegisteredBottle
_MOD = "bot_bottle.backend.macos_container.consolidated_launch"
_UTIL = "bot_bottle.backend.consolidated_util"
def _egress_plan() -> EgressPlan:
@@ -87,7 +88,8 @@ class TestRegisterAgent(unittest.TestCase):
*, source_ip: str = "192.168.128.9",
):
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.provision_git_gate", provision or Mock()):
patch(f"{_UTIL}.provision_git_gate", provision or Mock()), \
patch(f"{_MOD}.live_source_ips", return_value=[]):
return register_agent(
_egress_plan(), _git_plan(),
source_ip=source_ip, endpoint=_endpoint(), image_ref="img:1",
@@ -125,8 +127,8 @@ class TestTeardown(unittest.TestCase):
def test_deregisters_and_deprovisions(self) -> None:
client = Mock()
deprovision = Mock()
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_MOD}.deprovision_git_gate", deprovision):
with patch(f"{_UTIL}.OrchestratorClient", return_value=client), \
patch(f"{_UTIL}.deprovision_git_gate", deprovision):
teardown_consolidated("b1", orchestrator_url="http://o:8099")
client.teardown_bottle.assert_called_once_with("b1")
self.assertEqual("b1", deprovision.call_args.args[1])
@@ -134,3 +136,104 @@ class TestTeardown(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestLiveSourceIps(unittest.TestCase):
"""The reconciliation input: the host enumerates its own bottles because
the orchestrator, inside the infra container, cannot see the backend."""
def _agents(self, *slugs: str) -> list[Mock]:
return [Mock(slug=s) for s in slugs]
def test_maps_slugs_to_container_addresses(self) -> None:
from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.container_mod.inspect_container_network_ip",
side_effect=["10.0.0.1", "10.0.0.2"]) as ip:
got = live_source_ips("net0")
self.assertEqual(["10.0.0.1", "10.0.0.2"], got)
self.assertEqual("bot-bottle-a", ip.call_args_list[0].args[0])
def test_containers_without_an_address_are_skipped(self) -> None:
"""A container that hasn't been given a DHCP address yet contributes
nothing the reap's grace window, not this list, protects it."""
from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.container_mod.inspect_container_network_ip",
side_effect=["", "10.0.0.2"]):
self.assertEqual(["10.0.0.2"], live_source_ips("net0"))
def test_container_list_failure_raises(self) -> None:
"""If container list fails, the live set is not authoritative and
reconciliation must be skipped."""
from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
from bot_bottle.backend.macos_container.enumerate import EnumerationError
with patch(f"{_MOD}.enumerate_active",
side_effect=EnumerationError("container list failed")):
with self.assertRaises(EnumerationError):
live_source_ips("net0")
def test_per_container_inspect_failure_raises(self) -> None:
"""If any individual inspect fails, the live set is not authoritative."""
from bot_bottle.backend.macos_container.consolidated_launch import live_source_ips
from bot_bottle.backend.macos_container.enumerate import EnumerationError
with patch(f"{_MOD}.enumerate_active", return_value=self._agents("a", "b")), \
patch(f"{_MOD}.container_mod.inspect_container_network_ip",
side_effect=["10.0.0.1", None]):
with self.assertRaises(EnumerationError):
live_source_ips("net0")
class TestRegisterAgentReconciles(unittest.TestCase):
"""Registration self-heals the registry first: an orphan row at a recycled
address makes attribution ambiguous, which resolves no policy at all and
denies every host for the bottle being launched."""
def _register(self, client: Mock) -> None:
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_UTIL}.provision_git_gate"), \
patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]):
register_agent(
_egress_plan(), _git_plan(),
source_ip="10.0.0.7", endpoint=_endpoint(),
)
def test_reconciles_before_registering(self) -> None:
client = _client()
calls: list[str] = []
def _reconcile(*_args: object, **_kwargs: object) -> list[str]:
calls.append("reconcile")
return []
def _register_bottle(*_args: object, **_kwargs: object) -> RegisteredBottle:
calls.append("register")
return RegisteredBottle("b1", "tok")
client.reconcile.side_effect = _reconcile
client.register_bottle.side_effect = _register_bottle
self._register(client)
self.assertEqual(["reconcile", "register"], calls)
client.reconcile.assert_called_once_with(["10.0.0.7"])
def test_a_reconcile_failure_does_not_block_the_launch(self) -> None:
from bot_bottle.orchestrator.client import OrchestratorClientError
client = _client()
client.reconcile.side_effect = OrchestratorClientError("unreachable")
self._register(client)
client.register_bottle.assert_called_once()
def test_enumeration_error_does_not_block_the_launch(self) -> None:
"""A partial container listing must not abort the launch — skip
reconciliation and proceed, just as with an unreachable orchestrator."""
from bot_bottle.backend.macos_container.enumerate import EnumerationError
client = _client()
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_UTIL}.provision_git_gate"), \
patch(f"{_MOD}.live_source_ips",
side_effect=EnumerationError("container list failed")):
register_agent(
_egress_plan(), _git_plan(),
source_ip="10.0.0.7", endpoint=_endpoint(),
)
client.register_bottle.assert_called_once()
+4 -2
View File
@@ -66,8 +66,10 @@ class TestMacosContainerEnumerate(unittest.TestCase):
agents = self._enumerate("bot-bottle-mac-infra\nbot-bottle-dev-abc\n")
self.assertEqual(["dev-abc"], [a.slug for a in agents])
def test_empty_when_the_cli_fails(self):
self.assertEqual([], self._enumerate("", returncode=1))
def test_raises_when_the_cli_fails(self):
from bot_bottle.backend.macos_container.enumerate import EnumerationError
with self.assertRaises(EnumerationError):
self._enumerate("", returncode=1)
if __name__ == "__main__":
+44
View File
@@ -334,6 +334,50 @@ class TestInspectDigests(unittest.TestCase):
self.assertEqual({}, util.container_env("x"))
class TestInspectContainerNetworkIp(unittest.TestCase):
"""inspect_container_network_ip must distinguish inspect failure (None)
from 'no DHCP address yet' (""), which is the invariant live_source_ips
relies on to skip reconciliation on partial snapshots."""
_NETWORK = "bot-bottle-mac-gateway"
def _inspect(self, stdout: str, returncode: int = 0) -> str | None:
cp = util.subprocess.CompletedProcess(
args=[], returncode=returncode, stdout=stdout, stderr="",
)
with patch.object(util.subprocess, "run", return_value=cp):
return util.inspect_container_network_ip("bot-bottle-abc", self._NETWORK)
def _entry(self, ip: str = "192.168.128.5") -> str:
return (
f'[{{"status":{{"networks":['
f'{{"network":"{self._NETWORK}","ipv4Address":"{ip}"}}'
f']}}}}]'
)
def test_returns_ip_when_inspect_succeeds(self) -> None:
self.assertEqual("192.168.128.5", self._inspect(self._entry()))
def test_strips_cidr_prefix(self) -> None:
self.assertEqual("192.168.128.5", self._inspect(self._entry("192.168.128.5/24")))
def test_returns_empty_string_when_no_address_assigned_yet(self) -> None:
no_ip = f'[{{"status":{{"networks":[{{"network":"{self._NETWORK}","ipv4Address":""}}]}}}}]'
self.assertEqual("", self._inspect(no_ip))
def test_returns_empty_string_when_network_list_absent(self) -> None:
self.assertEqual("", self._inspect('[{"status":{}}]'))
def test_returns_none_on_nonzero_exit(self) -> None:
self.assertIsNone(self._inspect("", returncode=1))
def test_returns_none_on_malformed_json(self) -> None:
self.assertIsNone(self._inspect("not-json"))
def test_returns_none_on_unexpected_json_shape(self) -> None:
self.assertIsNone(self._inspect("null"))
class TestWaitContainerIpv4(unittest.TestCase):
def test_returns_address_once_dhcp_assigns_it(self):
with patch.object(util, "try_container_ipv4_on_network", side_effect=["", "", "192.168.128.4"]), \
+8
View File
@@ -153,6 +153,14 @@ class TestCaCertPem(unittest.TestCase):
argv = mod.run_container_argv.call_args.args[0]
self.assertEqual(["container", "exec", "bot-bottle-mac-infra", "cat"], argv[:4])
def test_raises_gateway_error_when_cert_never_appears(self) -> None:
from bot_bottle.backend.macos_container.gateway import GatewayError
svc = MacosInfraService(repo_root=Path("/r"))
with patch(f"{_INFRA}.container_mod") as mod:
mod.run_container_argv.return_value = _fail()
with self.assertRaises(GatewayError):
svc.ca_cert_pem(timeout=0)
class TestProbeControlPlane(unittest.TestCase):
def test_returns_url_when_running(self) -> None:
+29
View File
@@ -104,3 +104,32 @@ class TestHealthAndPolicy(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReconcile(unittest.TestCase):
def setUp(self) -> None:
self.c = OrchestratorClient("http://orch:8080")
def test_posts_live_ips_and_returns_reaped(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["b1", "b2"]})) as m:
got = self.c.reconcile(["10.0.0.2", "10.0.0.3"])
self.assertEqual(["b1", "b2"], got)
sent = json.loads(m.call_args.args[0].data)
self.assertEqual(["10.0.0.2", "10.0.0.3"], sent["live_source_ips"])
self.assertNotIn("grace_seconds", sent) # omitted -> server default
def test_grace_seconds_is_forwarded_when_given(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": []})) as m:
self.c.reconcile([], grace_seconds=30)
self.assertEqual(30, json.loads(m.call_args.args[0].data)["grace_seconds"])
def test_malformed_reaped_is_tolerated(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["ok", 5, None]})):
self.assertEqual(["ok"], self.c.reconcile([]))
with patch(_URLOPEN, return_value=_resp(200, {})):
self.assertEqual([], self.c.reconcile([]))
def test_error_status_raises(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(500)):
with self.assertRaises(OrchestratorClientError):
self.c.reconcile([])
@@ -0,0 +1,147 @@
"""Unit: OrchestratorConfigStore and resolve_teardown_timeout.
Also verifies the lifecycle ordering invariant: resolve_teardown_timeout()
must be called before launch_consolidated() / register_agent() so that a
resolver failure cannot leave an orphaned registration with no teardown
callback.
"""
from __future__ import annotations
import inspect
import os
import tempfile
import unittest
from pathlib import Path
from types import ModuleType
from bot_bottle.orchestrator.config_store import (
DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
TEARDOWN_TIMEOUT_ENV,
OrchestratorConfigStore,
resolve_teardown_timeout,
)
class TestOrchestratorConfigStore(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db = Path(self._tmp.name) / "test.db"
self.store = OrchestratorConfigStore(self.db)
self.store.migrate()
def tearDown(self) -> None:
self._tmp.cleanup()
def test_get_returns_none_when_not_set(self) -> None:
self.assertIsNone(self.store.get_teardown_timeout_seconds())
def test_set_and_get_roundtrip(self) -> None:
self.store.set_teardown_timeout_seconds(42.5)
self.assertEqual(42.5, self.store.get_teardown_timeout_seconds())
def test_set_overwrites_existing_value(self) -> None:
self.store.set_teardown_timeout_seconds(10.0)
self.store.set_teardown_timeout_seconds(20.0)
self.assertEqual(20.0, self.store.get_teardown_timeout_seconds())
def test_delete_clears_value_and_returns_true(self) -> None:
self.store.set_teardown_timeout_seconds(30.0)
deleted = self.store.delete_teardown_timeout_seconds()
self.assertTrue(deleted)
self.assertIsNone(self.store.get_teardown_timeout_seconds())
def test_delete_absent_returns_false(self) -> None:
self.assertFalse(self.store.delete_teardown_timeout_seconds())
def test_is_migrated_true_after_migrate(self) -> None:
self.assertTrue(self.store.is_migrated())
def test_is_migrated_false_before_migrate(self) -> None:
store = OrchestratorConfigStore(Path(self._tmp.name) / "new.db")
self.assertFalse(store.is_migrated())
class TestResolveTeardownTimeout(unittest.TestCase):
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db = Path(self._tmp.name) / "cfg.db"
def tearDown(self) -> None:
self._tmp.cleanup()
os.environ.pop(TEARDOWN_TIMEOUT_ENV, None)
def test_returns_default_when_nothing_configured(self) -> None:
self.assertEqual(
DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
resolve_teardown_timeout(self.db),
)
def test_env_var_overrides_default(self) -> None:
os.environ[TEARDOWN_TIMEOUT_ENV] = "99"
self.assertEqual(99.0, resolve_teardown_timeout(self.db))
def test_env_var_overrides_db_value(self) -> None:
store = OrchestratorConfigStore(self.db)
store.migrate()
store.set_teardown_timeout_seconds(55.0)
os.environ[TEARDOWN_TIMEOUT_ENV] = "77"
self.assertEqual(77.0, resolve_teardown_timeout(self.db))
def test_db_value_overrides_default(self) -> None:
store = OrchestratorConfigStore(self.db)
store.migrate()
store.set_teardown_timeout_seconds(42.0)
self.assertEqual(42.0, resolve_teardown_timeout(self.db))
def test_invalid_env_var_falls_through_to_default(self) -> None:
os.environ[TEARDOWN_TIMEOUT_ENV] = "not-a-number"
self.assertEqual(DEFAULT_TEARDOWN_TIMEOUT_SECONDS, resolve_teardown_timeout(self.db))
def test_non_positive_env_var_falls_through_to_default(self) -> None:
os.environ[TEARDOWN_TIMEOUT_ENV] = "0"
self.assertEqual(DEFAULT_TEARDOWN_TIMEOUT_SECONDS, resolve_teardown_timeout(self.db))
def test_non_positive_db_value_falls_through_to_default(self) -> None:
store = OrchestratorConfigStore(self.db)
store.migrate()
store.set_teardown_timeout_seconds(0.0)
self.assertEqual(DEFAULT_TEARDOWN_TIMEOUT_SECONDS, resolve_teardown_timeout(self.db))
def test_migrates_db_on_first_call(self) -> None:
result = resolve_teardown_timeout(self.db)
self.assertEqual(DEFAULT_TEARDOWN_TIMEOUT_SECONDS, result)
self.assertTrue(OrchestratorConfigStore(self.db).is_migrated())
class TestTeardownTimeoutResolvedBeforeRegistration(unittest.TestCase):
"""Ordering invariant: if resolve_teardown_timeout() raises, the bottle
must not yet be registered no orphaned state can result."""
def _src(self, module: ModuleType) -> str:
return inspect.getsource(module)
def test_docker_resolves_timeout_before_launch_consolidated(self) -> None:
from bot_bottle.backend.docker import launch
src = self._src(launch)
resolve_at = src.index("teardown_timeout = resolve_teardown_timeout()")
launch_at = src.index("ctx = launch_consolidated(")
self.assertLess(resolve_at, launch_at)
def test_firecracker_resolves_timeout_before_launch_consolidated(self) -> None:
from bot_bottle.backend.firecracker import launch
src = self._src(launch)
resolve_at = src.index("teardown_timeout = resolve_teardown_timeout()")
launch_at = src.index("ctx = launch_consolidated(")
self.assertLess(resolve_at, launch_at)
def test_macos_resolves_timeout_before_register_agent(self) -> None:
from bot_bottle.backend.macos_container import launch
src = self._src(launch)
resolve_at = src.index("teardown_timeout = resolve_teardown_timeout()")
register_at = src.index("ctx = register_agent(")
self.assertLess(resolve_at, register_at)
if __name__ == "__main__":
unittest.main()
@@ -8,11 +8,13 @@ from __future__ import annotations
import json
import secrets
import sqlite3
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from contextlib import closing
from pathlib import Path
from unittest.mock import patch
@@ -264,6 +266,7 @@ class TestControlPlaneAuth(unittest.TestCase):
("POST", "/bottles", _body({"source_ip": "10.0.0.1"})),
("PUT", "/bottles/x/policy", _body({"policy": "routes: []"})),
("DELETE", "/bottles/x", b""),
("POST", "/reconcile", _body({"live_source_ips": []})),
("POST", "/resolve", _body({"source_ip": "10.0.0.1", "identity_token": "t"})),
("POST", "/attribute", _body({"source_ip": "10.0.0.1", "identity_token": "t"})),
("GET", "/supervise/proposals", b""),
@@ -387,3 +390,56 @@ class TestDispatchSupervise(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReconcileRoute(unittest.TestCase):
"""`POST /reconcile` — the host tells the orchestrator which bottles are
actually up, since the orchestrator can't see the backend from inside the
infra container."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
def tearDown(self) -> None:
self._tmp.cleanup()
def _old(self, source_ip: str) -> str:
rec = self.orch.registry.register(source_ip)
with closing(sqlite3.connect(self.orch.registry.db_path)) as conn:
conn.execute(
"UPDATE orchestrator_bottles SET created_at = 0.0 WHERE bottle_id = ?",
(rec.bottle_id,))
conn.commit()
return rec.bottle_id
def test_reaps_absent_and_reports_ids(self) -> None:
dead = self._old("10.0.0.1")
alive = self._old("10.0.0.2")
status, payload = dispatch(
self.orch, "POST", "/reconcile", _body({"live_source_ips": ["10.0.0.2"]}))
self.assertEqual(200, status)
self.assertEqual([dead], payload["reaped"])
self.assertIsNone(self.orch.registry.get(dead))
self.assertIsNotNone(self.orch.registry.get(alive))
def test_missing_live_source_ips_is_400(self) -> None:
status, _ = dispatch(self.orch, "POST", "/reconcile", _body({}))
self.assertEqual(400, status)
def test_grace_seconds_is_honoured(self) -> None:
"""A grace window wide enough to cover the row protects it."""
self.orch.registry.register("10.0.0.3")
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [], "grace_seconds": 3600}))
self.assertEqual(200, status)
self.assertEqual([], payload["reaped"])
def test_non_string_entries_are_ignored(self) -> None:
dead = self._old("10.0.0.4")
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [None, 7, "10.0.0.9"]}))
self.assertEqual(200, status)
self.assertEqual([dead], payload["reaped"])
+119 -79
View File
@@ -1,4 +1,4 @@
"""Unit: orchestrator+gateway container lifecycle — idempotent singleton (PRD 0070)."""
"""Unit: infra container lifecycle — idempotent singleton (PRD 0070)."""
from __future__ import annotations
@@ -8,10 +8,10 @@ import urllib.error
from pathlib import Path
from unittest.mock import MagicMock, Mock, patch
from bot_bottle.orchestrator.gateway import GatewayError
from bot_bottle.orchestrator.lifecycle import (
ORCHESTRATOR_IMAGE,
ORCHESTRATOR_NAME,
ORCHESTRATOR_SOURCE_HASH_LABEL,
INFRA_NAME,
INFRA_SOURCE_HASH_LABEL,
OrchestratorService,
OrchestratorStartError,
source_hash,
@@ -20,7 +20,6 @@ from tests.unit import use_bottle_root
_URLOPEN = "bot_bottle.orchestrator.lifecycle.urllib.request.urlopen"
_RUN = "bot_bottle.orchestrator.lifecycle.run_docker"
_GATEWAY = "bot_bottle.orchestrator.lifecycle.DockerGateway"
_SLEEP = "bot_bottle.orchestrator.lifecycle.time.sleep"
_MONOTONIC = "bot_bottle.orchestrator.lifecycle.time.monotonic"
@@ -42,10 +41,8 @@ class TestOrchestratorService(unittest.TestCase):
self.addCleanup(use_bottle_root(Path(self._tmp.name)))
self.svc = OrchestratorService(port=8099)
def test_urls(self) -> None:
def test_url(self) -> None:
self.assertEqual("http://127.0.0.1:8099", self.svc.url)
# The gateway reaches the control plane by container name over docker DNS.
self.assertEqual(f"http://{ORCHESTRATOR_NAME}:8099", self.svc.internal_url)
def test_is_healthy(self) -> None:
with patch(_URLOPEN, return_value=_health(200)):
@@ -54,126 +51,169 @@ class TestOrchestratorService(unittest.TestCase):
self.assertFalse(self.svc.is_healthy())
def test_ensure_running_noop_when_healthy_and_source_unchanged(self) -> None:
# A healthy control plane already running the *current* bind-mounted
# source is left alone — recreating it on every launch would drop
# every other active bottle's in-memory egress tokens (#381).
# A healthy container on current source is left alone — recreating it
# on every launch drops in-memory egress tokens (#381).
current = source_hash(self.svc._repo_root)
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
return _proc(stdout=INFRA_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(stdout=current)
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_GATEWAY) as gw_cls, patch(_RUN, side_effect=fake), patch(_SLEEP):
patch(_RUN, side_effect=fake), patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
gw_cls.return_value.ensure_running.assert_called() # gateway kept up
runs = [c for c in calls if c[:2] == ["docker", "run"]]
rms = [c for c in calls if c[:3] == ["docker", "rm", "--force"] and ORCHESTRATOR_NAME in c]
self.assertEqual([], runs) # not recreated
rms = [c for c in calls if c[:3] == ["docker", "rm", "--force"] and INFRA_NAME in c]
self.assertEqual([], runs)
self.assertEqual([], rms)
def test_ensure_running_recreates_when_source_changed(self) -> None:
# Healthy, but the running container's label doesn't match the
# current source hash (a real code change) — recreate so it takes
# effect, same as the gateway's image-staleness check.
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
return _proc(stdout=INFRA_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(stdout="stale-hash")
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_GATEWAY), patch(_RUN, side_effect=fake), patch(_SLEEP):
patch(_RUN, side_effect=fake), patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c for c in calls if c[:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
self.assertIn(ORCHESTRATOR_NAME, runs[0])
# the fresh container is labeled with the current hash, not the stale one
self.assertIn(INFRA_NAME, runs[0])
current = source_hash(self.svc._repo_root)
self.assertIn(f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current}", runs[0])
self.assertIn(f"{INFRA_SOURCE_HASH_LABEL}={current}", runs[0])
def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="") # not running
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_GATEWAY), patch(_RUN, side_effect=fake), patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c for c in calls if c[:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
argv = runs[0]
self.assertIn(ORCHESTRATOR_NAME, argv)
self.assertIn("--broker", argv)
self.assertEqual("stub", argv[argv.index("--broker") + 1]) # register-only, no socket
self.assertIn("bot_bottle.orchestrator", argv)
self.assertEqual("127.0.0.1:8099:8099", argv[argv.index("--publish") + 1])
def test_ensure_running_builds_lean_orchestrator_image_when_missing(self) -> None:
# The control plane runs its own lean image (#384), distinct from the
# gateway data plane — built from Dockerfile.orchestrator when absent.
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="") # orchestrator not running
if argv[:3] == ["docker", "image", "inspect"]:
return _proc(returncode=1) # image absent -> build
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_GATEWAY), patch(_RUN, side_effect=fake), patch(_SLEEP):
self.svc.ensure_running()
builds = [c for c in calls if c[:2] == ["docker", "build"]]
self.assertEqual(1, len(builds))
self.assertIn(ORCHESTRATOR_IMAGE, builds[0])
self.assertTrue(any(a.endswith("Dockerfile.orchestrator") for a in builds[0]))
# It is NOT the gateway image/dockerfile — the split is the point.
self.assertFalse(any("Dockerfile.gateway" in a for a in builds[0]))
def test_ensure_running_skips_orchestrator_image_build_when_present(self) -> None:
def test_ensure_running_starts_infra_container_when_absent(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
if argv[:3] == ["docker", "image", "inspect"]:
return _proc(returncode=0) # image present -> no build
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_GATEWAY), patch(_RUN, side_effect=fake), patch(_SLEEP):
patch(_RUN, side_effect=fake), patch(_SLEEP):
self.assertEqual(self.svc.url, self.svc.ensure_running())
runs = [c for c in calls if c[:2] == ["docker", "run"]]
self.assertEqual(1, len(runs))
argv = runs[0]
self.assertIn(INFRA_NAME, argv)
# Published on loopback — not exposed on external interfaces.
self.assertEqual("127.0.0.1:8099:8099", argv[argv.index("--publish") + 1])
# Both processes in one container — no separate entrypoint override.
self.assertNotIn("--entrypoint", argv)
# Gateway daemons + orchestrator explicitly opted in.
daemons_flag = "BOT_BOTTLE_GATEWAY_DAEMONS=egress,git-http,supervise,orchestrator"
self.assertIn("orchestrator", argv[argv.index(daemons_flag)])
def test_ensure_running_builds_all_images(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake), patch(_SLEEP):
self.svc.ensure_running()
self.assertEqual([], [c for c in calls if c[:2] == ["docker", "build"]])
builds = [c for c in calls if c[:2] == ["docker", "build"]]
# Gateway base + orchestrator intermediate + infra image — all three built.
self.assertEqual(3, len(builds))
dockerfiles = [next(a for a in b if "Dockerfile" in a) for b in builds]
self.assertIn("Dockerfile.gateway", dockerfiles[0])
self.assertIn("Dockerfile.orchestrator", dockerfiles[1])
self.assertIn("Dockerfile.infra", dockerfiles[2])
# All three images are distinct.
tags = [b[b.index("-t") + 1] for b in builds]
self.assertEqual(3, len(set(tags)))
def test_publish_maps_host_port_to_fixed_internal_port(self) -> None:
"""A non-default self.port is published to the fixed internal port 8099,
not to self.port:self.port the orchestrator always listens on 8099."""
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
svc = OrchestratorService(port=20001)
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake), patch(_SLEEP):
svc.ensure_running()
runs = [c for c in calls if c[:2] == ["docker", "run"]]
argv = runs[0]
self.assertEqual("127.0.0.1:20001:8099", argv[argv.index("--publish") + 1])
orch_url = next(a for a in argv if "BOT_BOTTLE_ORCHESTRATOR_URL" in a)
self.assertIn(":8099", orch_url)
def test_ensure_running_raises_on_timeout(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_GATEWAY), patch(_RUN, return_value=Mock(returncode=0, stderr="")), \
patch(_RUN, return_value=Mock(returncode=0, stdout="", stderr="")), \
patch(_SLEEP), patch(_MONOTONIC, side_effect=[0.0, 0.5, 2.0]):
with self.assertRaises(OrchestratorStartError):
self.svc.ensure_running(startup_timeout=1.0)
def test_stop_removes_orchestrator_and_gateway(self) -> None:
with patch(_RUN) as run, patch(_GATEWAY) as gw_cls:
def test_noop_when_healthy_and_inspect_fails(self) -> None:
"""If docker inspect fails (e.g. docker daemon hiccup), leave the
working container alone rather than churning it."""
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=INFRA_NAME)
if argv[:2] == ["docker", "inspect"]:
return _proc(returncode=1, stderr="daemon error")
return _proc()
with patch(_URLOPEN, return_value=_health(200)), \
patch(_RUN, side_effect=fake), patch(_SLEEP):
self.svc.ensure_running()
# no docker run — the working container was left alone
def test_build_failure_raises(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("down")), \
patch(_RUN, return_value=_proc(returncode=1, stderr="no space left on device")):
with self.assertRaises(GatewayError):
self.svc.ensure_running()
def test_ensure_network_creates_if_missing(self) -> None:
"""If the gateway network doesn't exist yet, create it."""
calls: list[list[str]] = []
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:3] == ["docker", "network", "inspect"]:
return _proc(returncode=1, stderr="not found")
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
return _proc()
with patch(_URLOPEN, side_effect=[urllib.error.URLError("down"), _health(200)]), \
patch(_RUN, side_effect=fake), patch(_SLEEP):
self.svc.ensure_running()
creates = [c for c in calls if c[:3] == ["docker", "network", "create"]]
self.assertEqual(1, len(creates))
def test_stop_removes_infra_container(self) -> None:
with patch(_RUN) as run:
self.svc.stop()
rms = [c.args[0] for c in run.call_args_list if c.args[0][:3] == ["docker", "rm", "--force"]]
self.assertTrue(any(ORCHESTRATOR_NAME in a for a in rms))
gw_cls.return_value.stop.assert_called_once()
rms = [
c.args[0] for c in run.call_args_list
if c.args[0][:3] == ["docker", "rm", "--force"]
]
self.assertTrue(any(INFRA_NAME in a for a in rms))
if __name__ == "__main__":
+80
View File
@@ -4,7 +4,9 @@ from __future__ import annotations
import sqlite3
import tempfile
import time
import unittest
from contextlib import closing
from pathlib import Path
from bot_bottle.orchestrator.registry import (
@@ -169,3 +171,81 @@ class TestRegistryStore(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestReapAbsent(unittest.TestCase):
"""`reap_absent` — the self-heal for rows whose bottle is gone.
An orphan is not merely untidy: source IPs get recycled, and
`by_source_ip` fail-closes on ambiguity, so a leftover row at a reused
address resolves *no* policy for the next bottle that lands there and
every host it asks for is denied.
"""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.db = Path(self._tmp.name) / "registry.db"
self.store = RegistryStore(self.db)
self.store.migrate()
def tearDown(self) -> None:
self._tmp.cleanup()
def _aged(self, source_ip: str, *, age: float) -> BottleRecord:
"""Register a bottle and backdate it past the grace window."""
rec = self.store.register(source_ip)
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"UPDATE orchestrator_bottles SET created_at = ? WHERE bottle_id = ?",
(time.time() - age, rec.bottle_id),
)
conn.commit()
return rec
def test_reaps_row_with_no_live_container(self) -> None:
gone = self._aged("10.243.0.9", age=600)
reaped = self.store.reap_absent([])
self.assertEqual([gone.bottle_id], [r.bottle_id for r in reaped])
self.assertIsNone(self.store.get(gone.bottle_id))
def test_keeps_row_whose_ip_is_live(self) -> None:
alive = self._aged("10.243.0.9", age=600)
self.assertEqual([], self.store.reap_absent(["10.243.0.9"]))
self.assertIsNotNone(self.store.get(alive.bottle_id))
def test_grace_window_protects_an_in_flight_launch(self) -> None:
"""A bottle registered moments ago is never reaped, even though the
caller's enumeration didn't see its address yet."""
fresh = self.store.register("10.243.0.10")
self.assertEqual([], self.store.reap_absent([]))
self.assertIsNotNone(self.store.get(fresh.bottle_id))
def test_reaping_the_orphan_unbricks_the_reused_address(self) -> None:
"""The regression this exists for: an orphan at an address that vmnet
later hands to a new bottle makes `by_source_ip` ambiguous, so the new
bottle resolves no policy at all."""
orphan = self._aged("10.243.0.11", age=600)
# A new bottle lands on the recycled address. Force the row in directly
# so `register`'s own supersede sweep doesn't mask the ambiguity.
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"INSERT INTO orchestrator_bottles "
"(bottle_id, source_ip, identity_token, state, created_at, metadata, policy) "
"VALUES ('newbottle', '10.243.0.11', 'tok-new', 'active', ?, '', 'routes: []')",
(time.time(),),
)
conn.commit()
self.assertIsNone(self.store.by_source_ip("10.243.0.11")) # bricked
reaped = self.store.reap_absent(["10.243.0.11"], grace_seconds=60)
self.assertEqual([orphan.bottle_id], [r.bottle_id for r in reaped])
rec = self.store.by_source_ip("10.243.0.11")
assert rec is not None
self.assertEqual("newbottle", rec.bottle_id)
def test_ignores_empty_ips_in_the_live_set(self) -> None:
gone = self._aged("10.243.0.12", age=600)
self.assertEqual(
[gone.bottle_id],
[r.bottle_id for r in self.store.reap_absent(["", "10.243.0.99"])],
)
+54
View File
@@ -4,8 +4,10 @@ from __future__ import annotations
import json
import secrets
import sqlite3
import tempfile
import unittest
from contextlib import closing
from pathlib import Path
from unittest.mock import patch
@@ -304,3 +306,55 @@ class TestOrchestratorSupervise(unittest.TestCase):
if __name__ == "__main__":
unittest.main()
class TestOrchestratorReconcile(unittest.TestCase):
"""`reconcile` — drop rows for bottles that are no longer running."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.secret = secrets.token_bytes(16)
self.db = Path(self._tmp.name) / "r.db"
self.store = RegistryStore(self.db)
self.store.migrate()
self.broker = StubBroker(self.secret)
self.orch = Orchestrator(self.store, self.broker, self.secret)
def tearDown(self) -> None:
self._tmp.cleanup()
def _age_all(self, seconds: float) -> None:
"""Backdate every row past the reap grace window."""
with closing(sqlite3.connect(self.db)) as conn:
conn.execute(
"UPDATE orchestrator_bottles SET created_at = created_at - ?", (seconds,))
conn.commit()
def test_reaps_dead_bottle_and_forgets_its_tokens(self) -> None:
dead = self.orch.launch_bottle("10.243.0.1", tokens={"EGRESS_TOKEN_0": "s3cret"})
live = self.orch.launch_bottle("10.243.0.2", tokens={"EGRESS_TOKEN_0": "keep"})
self._age_all(600)
self.assertEqual([dead.bottle_id], self.orch.reconcile(["10.243.0.2"]))
self.assertIsNone(self.store.get(dead.bottle_id))
self.assertIsNotNone(self.store.get(live.bottle_id))
# The in-memory egress credential goes with the row.
self.assertEqual({}, self.orch.tokens_for(dead.bottle_id))
self.assertEqual({"EGRESS_TOKEN_0": "keep"}, self.orch.tokens_for(live.bottle_id))
def test_reconcile_does_not_broker_a_teardown(self) -> None:
"""The container is already gone — there is nothing to stop, and a
broker error must not stop the sweep clearing the row."""
self.orch.launch_bottle("10.243.0.1")
self._age_all(600)
self.broker.launched.clear()
self.orch.reconcile([])
self.assertEqual([], self.broker.torn_down)
def test_reconcile_keeps_everything_when_all_are_live(self) -> None:
a = self.orch.launch_bottle("10.243.0.1")
b = self.orch.launch_bottle("10.243.0.2")
self._age_all(600)
self.assertEqual([], self.orch.reconcile(["10.243.0.1", "10.243.0.2"]))
self.assertIsNotNone(self.store.get(a.bottle_id))
self.assertIsNotNone(self.store.get(b.bottle_id))