Compare commits

..

9 Commits

Author SHA1 Message Date
didericis fd86e7fa99 fix(egress): redact the per-bottle token in the now-active multi-tenant response log
test / unit (pull_request) Successful in 1m9s
test / integration (pull_request) Successful in 21s
test / coverage (pull_request) Successful in 1m24s
lint / lint (push) Successful in 2m15s
test / unit (push) Successful in 1m8s
test / integration (push) Successful in 24s
test / coverage (push) Successful in 1m20s
Update Quality Badges / update-badges (push) Successful in 1m18s
Self-review of this PR: making `response()` run in the consolidated gateway
also activates its `LOG_FULL` `_log_response` call there — previously
unreachable, since the empty static config made `response()` return early. That
logger redacted with `os.environ`, which in multi-tenant mode does NOT hold the
bottle's per-request `/resolve` tokens (only the resolved `env` overlay does),
so a non-token-shaped provisioned secret appearing in a response could be logged
in the clear.

Thread the resolved per-flow `env` into `_log_request` / `_log_response` so the
LOG_FULL redaction scrubs the calling bottle's secrets. Adds a regression test
(a non-token-shaped `/resolve` secret, absent from os.environ, must not appear
in the response log) and updates the redaction-test helpers for the new arg.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UoEZHDjv84ChoZbozQERhJ
2026-07-17 17:14:52 -04:00
didericis 3bbd839917 fix(egress): scan response + websocket DLP against the resolved per-flow config
In the consolidated (multi-tenant) gateway the addon's static `self.config`
is empty — each request's real policy comes from the per-request `/resolve`.
`response()` and `websocket_message()` still matched routes against that empty
config, so inbound prompt-injection DLP and WebSocket credential/injection DLP
silently skipped every scan (fail-open) whenever the gateway ran multi-tenant.
This is backend-agnostic: the gateway image (and this addon) is shared by the
Firecracker, macOS, and docker consolidated backends.

Resolve the per-flow (config, slug, env) once in `request()`, stash it on
`flow.metadata`, and have both hooks read it back — falling back to the static
single-tenant values for a flow that never passed through `request()`. Reusing
the request's one `/resolve` avoids a round-trip per response and per WebSocket
frame.

Adds multi-tenant regression tests for both hooks that fail against the old
fall-open behaviour.

Refs: audit issue #400 (finding #2)

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UoEZHDjv84ChoZbozQERhJ
2026-07-17 17:14:52 -04:00
didericis 5c526860bc fix(test): resolve the remaining review findings on the control-plane auth test
test / unit (pull_request) Successful in 1m17s
test / integration (pull_request) Successful in 22s
test / coverage (pull_request) Successful in 1m21s
lint / lint (push) Successful in 2m23s
test / unit (push) Successful in 1m24s
test / integration (push) Successful in 30s
test / coverage (push) Successful in 1m26s
Update Quality Badges / update-badges (push) Successful in 1m24s
Addresses the 5 lower-priority findings left as follow-up in the earlier
review, now that each has a concrete answer:

- Add gateway_name: str = GATEWAY_NAME to OrchestratorService.__init__
  (mirrors the existing orchestrator_name param) and thread it through
  _gateway(). Deletes the test's _IsolatedOrchestratorService subclass,
  which existed only to override a private method for this one kwarg —
  any caller needing gateway-name isolation can now use the public
  constructor. Backward compatible: every existing caller constructs
  OrchestratorService with keyword args and a sensible default is kept.

- Give the test its own fixed image tags (bot-bottle-orchestrator:itest,
  bot-bottle-gateway:itest) instead of the production :latest ones.
  _running_image_is_current() keys gateway staleness off the image tag's
  ID, not per-instance identity, so rebuilding the shared :latest tag from
  whatever's on disk during a test run could make a real host's running
  production gateway look stale and get force-recreated. Fixed tags (not
  per-run-suffixed, so they don't accumulate) fully decouple the two.

- setUp -> setUpClass/tearDownClass: all 5 tests are read-only checks
  against the same running control plane, so one shared container
  lifecycle replaces 5 (each of which paid its own container-start +
  image-build + health-poll cycle). Cuts the file's wall-clock roughly
  4x (11.5s -> 2.9-4.3s) and, combined with the network-rm cleanup from
  the previous commit, means one cleanup instead of five.

- Reuse OrchestratorClient (bot_bottle/orchestrator/client.py) instead of
  a hand-rolled urllib helper — the test now exercises the same
  request/response code path the real host CLI uses, rather than a
  private copy that could silently drift from it.

- Add the chown workaround test_multitenant_isolation.py already needed
  for this exact bind-mount: the orchestrator container has no USER
  directive, so it writes the registry DB as root into the throwaway
  host_root; chown it back before tempdir cleanup so that doesn't raise
  PermissionError on native Linux Docker (no UID remap, unlike Docker
  Desktop's macOS VM).

Verified: ran the suite twice in a row (idempotency — fixed image tags
don't accumulate, 5/5 pass both times, 2.99-4.33s each), the real
~/.bot-bottle/control-plane-token is untouched, zero leaked networks or
containers after either run, exactly 2 :itest images (not growing), the
full orchestrator unit suite (93 tests) and the sibling docker
gateway/broker integration tests still pass. pyright clean, pylint
10.00/10 on both changed files.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-17 17:00:31 -04:00
didericis 492669e620 fix(test): skip the control-plane auth test under act_runner (CI)
lint / lint (push) Successful in 2m23s
test / unit (pull_request) Successful in 1m17s
test / integration (pull_request) Successful in 30s
test / coverage (pull_request) Successful in 1m29s
Pushing the previous two fixes surfaced a real, currently-failing CI run
(gitea actions run 2164, jobs "integration" and "coverage"): every one of
the 5 new tests errored in setUp with

  docker: Error response from daemon: error while creating mount source
  path '/workspace/didericis/bot-bottle': mkdir /workspace: read-only
  file system

This is finding #4 from the review of the previous commit, now confirmed
live rather than just plausible: act_runner's job-container topology
can't satisfy the orchestrator's host-path bind mount, the same
constraint test_multitenant_isolation.py, test_gateway_image.py, and
test_sandbox_escape.py already skip around. Add the identical
skip_unless_docker + GITEA_ACTIONS guard so this test degrades the same
way its siblings do instead of failing the job.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-17 16:45:50 -04:00
didericis d32f36bb2b fix(test): actually isolate the docker control-plane auth test's root + network
Two bugs surfaced by a review of the previous commit:

- host_control_plane_token() resolves its path via the ambient
  BOT_BOTTLE_ROOT env var, not the host_root kwarg passed to
  OrchestratorService (that kwarg only controls the DB bind-mount
  destination). The test's isolation claim was false as a result: running
  it read/wrote the developer's real ~/.bot-bottle/control-plane-token
  instead of the throwaway temp dir — confirmed directly on disk. Fixed
  by pointing the env var at the same temp dir for the test's duration
  and restoring it via addCleanup.

- ensure_running() creates a per-bottle Docker network but stop() only
  ever removes containers, never the network — every run of this test
  leaked one bridge network permanently (found and removed 5 from prior
  runs via `docker network ls`). Fixed with an explicit `docker network
  rm` in addCleanup.

Verified: re-ran the suite twice: 5/5 pass both times, the real
~/.bot-bottle/control-plane-token timestamp is unchanged across both runs
(proving isolation), and `docker network ls` shows zero leaked
bot-bottle-net-itest-* networks afterward. pyright clean, pylint 10.00/10.

Remaining findings from the same review (missing GITEA_ACTIONS skip
guard, root-owned bind-mount cleanup on native Linux, no setUpClass,
reinvented OrchestratorClient, gateway_name should be a constructor
param rather than a subclassed private-method override) are left for a
follow-up — each is a real, separate design/scope call, not a
mechanical fix like these two.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-17 16:43:07 -04:00
didericis 4f36b919b2 test(orchestrator): live Docker coverage for the control-plane auth fix (#400)
lint / lint (push) Successful in 2m30s
test / unit (pull_request) Successful in 1m24s
test / integration (pull_request) Failing after 3m8s
test / coverage (pull_request) Failing after 1m59s
ca91fc4 fixed the control plane's missing caller authentication but only
verified it end-to-end on Apple Container; Docker had the same code change
(docker_cmd.py's env injection) backed only by the in-process dispatch()
unit tests, which never exercise the real HTTP server or a real container.

Ran it for real first: brought up the actual orchestrator + gateway via
Docker, hit the published control-plane port directly. Confirms /health
stays open, /bottles and /resolve both 401 with no token or a wrong one
(the enumeration and credential-lift vectors from #400), and /bottles
succeeds with the real per-host secret.

Added as a proper integration test so this stays covered: a subclass of
OrchestratorService giving the gateway its own unique name too (the base
class only parameterizes the orchestrator's), plus a throwaway
BOT_BOTTLE_ROOT and a random port, so a run can never collide with a real
host's orchestrator or gateway. Gated on a reachable Docker daemon, same
as the existing docker gateway/broker integration tests.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-17 16:17:42 -04:00
didericis ca91fc4d91 fix(security): authenticate the orchestrator control plane (issue #400 Critical)
lint / lint (push) Successful in 2m34s
test / unit (pull_request) Successful in 1m21s
test / integration (pull_request) Successful in 24s
test / coverage (pull_request) Successful in 1m26s
The control plane had no caller authentication — dispatch() routed every
request straight through, and the only guards anywhere were the
(source_ip, identity_token) checks inside /resolve and /attribute, which
authenticate the *bottle* a request is about, not the *caller*. On the
consolidated backends the agent shares a network with the control-plane port
(macOS: the infra container runs 8099 on the agent's --internal network;
docker: the orchestrator container sits on the shared gateway network), so a
prompt-injected agent could GET /bottles (enumerate siblings + policies),
PUT /bottles/<id>/policy (rewrite its own egress allowlist to allow-all),
POST /resolve (read the injected upstream credentials it's never meant to
see), and POST /supervise/respond (self-approve its own proposals). Every
README guarantee collapsed to the agent's discretion.

Fix: a per-host control-plane secret required on every route but GET /health,
compared with hmac.compare_digest. It is held only by the trusted callers and
never handed to an agent:
- minted + persisted 0600 at <root>/control-plane-token (paths.host_control_plane_token);
- injected as $BOT_BOTTLE_CONTROL_PLANE_TOKEN into the orchestrator + gateway
  containers via bare `--env NAME` (value inherited from the launch process,
  so it never lands on argv or in `container/docker inspect`);
- presented by the gateway's PolicyResolver (reads the env) on /resolve, and by
  the host CLI's OrchestratorClient (reads the host file) on every call.

The agent container is never given the env var or the host file, so from a
bottle every /bottles*, /resolve, /attribute, and /supervise/* call now
returns 401 — closing the enumeration, allowlist-rewrite, credential-lift, and
self-approval. The existing (source_ip, identity_token) checks stay as
defense-in-depth.

Enforced when configured: macOS + docker inject the secret (→ enforced). With
no secret set the server runs open and warns loudly at startup — a
fail-visible fallback for the unit suite and for Firecracker, whose
port-scoped nft already blocks agents from 8099 (wiring the secret into its
infra-VM init is a clean fast-follow, left out here to avoid churning the
prebuilt-artifact hash).

Verified end-to-end on real Apple Container: infra comes up healthy, the host
CLI (with the secret) lists bottles while an unauthenticated GET /bottles gets
401, all five issue-#400 attacks from inside the agent get 401, and egress
policy still works (200 allowed / 403 denied) — proving the gateway
authenticates to /resolve with the secret. 1829 unit tests pass, pyright
clean, pylint 9.91.

Refs #400.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-17 05:15:16 -04:00
didericis 4a607ad098 refactor(macos): one infra container (control plane + gateway), fixes shared-DB races
lint / lint (push) Successful in 2m15s
test / unit (pull_request) Successful in 1m16s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m17s
Adopts the firecracker infra-VM pattern for macOS: the orchestrator control
plane and the gateway data plane now run in a SINGLE Apple container instead of
two. Apple Containers are lightweight VMs with separate kernels, so the prior
two-container design had both guests writing one bot-bottle.db over virtiofs,
where fcntl locks are not coherent across kernels — concurrent writes (the
orchestrator's registry vs the gateway supervise daemon's queue) could corrupt
it. One container = one kernel = coherent locking.

The DB moves onto a container-only Apple volume (bot-bottle-mac-db), never
bind-mounted from the host, so no host process opens the live file either. The
host CLI already reaches registry + supervise state over the control-plane HTTP
surface (cli/supervise.py uses OrchestratorClient), exactly as firecracker's
VM-only DB requires.

Two simplifications fall out of the single container:
- No DNS dance: the control plane and gateway daemons reach each other over
  127.0.0.1, so the orchestrator-before-gateway ordering (a workaround for
  Apple having no container DNS) is gone, along with the moved-IP recreate
  logic it needed.
- Net -243 lines.

Mechanics: the infra container runs from the gateway image with the
control-plane source bind-mounted read-only (like the docker orchestrator, so a
code change needs no rebuild) and a small sh -c init that starts both processes
(mirrors firecracker's _infra_init). Also implements the macOS backend's
ensure_orchestrator() and adds it to discover_orchestrator_url, so operator
tools (supervise) can bring up / find the control plane on demand — previously
the macOS backend died with "no orchestrator control plane".

Verified end-to-end on real Apple Container 1.0.0: the single infra container
comes up healthy (one address for control plane + gateway), both processes run,
the DB is written on the container-only volume, host-side supervise works over
HTTP, and a registered agent gets 200 for an allowed host / 403 for a denied
one. 1824 unit tests pass with `container` absent (CI parity), pyright clean,
pylint 9.89.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-17 04:14:14 -04:00
didericis e24b62b6b9 fix(macos): review fixes — token on plan, self-heal, symmetric digest, DHCP poll
lint / lint (push) Successful in 2m18s
test / unit (pull_request) Successful in 1m6s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Successful in 1m18s
Addresses findings from a high-effort review of the PRD 0070 macOS backend.

Correctness:
- Stamp identity_token onto MacosContainerBottlePlan after registration. git's
  gitconfig extraHeader and the supervise MCP --header read
  getattr(plan,"identity_token","") at provision time, and both reach the
  gateway on NO_PROXY (bypassing the egress proxy that carries the token). The
  plan never carried it, so /resolve fail-closed and every git fetch/push and
  supervise call from a macOS bottle would have been denied. Registration
  precedes provision(), so — unlike the run-time env — the plan can carry it.
- Self-heal the orchestrator: recreate when it is not (source-current AND
  answering /health), not on the source-hash label alone. A container running
  current code but with a wedged HTTP server was left alone and polled to
  death, failing every launch until manual deletion.
- image_digest and container_image_digest now read the same descriptor.digest
  field; dropped image_digest's id/tag fallback that could yield a value the
  container side can't produce — a permanent mismatch would have recreated the
  shared gateway on every launch (severing every live bottle's egress, since
  the replacement gets a new DHCP address).
- Poll for the agent's and gateway's DHCP address instead of a fatal read
  right after `container run` (there is no --ip; the address can lag start).

Cleanup:
- One _inspect_first + _descriptor_digest behind the four inspect readers.
- Shared bind_mount_spec (util) and host_db_dir (paths) replace per-module
  copies; _GIT_HTTP_PORT now imports git_http_backend.DEFAULT_PORT.
- Drop the dead _url cache / url property and the write-only agent_proxy_url.

Deferred (noted on the PR, not fixed here): the gateway image rebuilding on
every launch (needs source-hash-labeled build), SQLite shared across VM
guests, and the sh -lc profile-override edge — each is design-level or
behavior-risk beyond a review fix.

Verified: real Apple Container bring-up is green and idempotent; 1826 unit
tests pass with `container` absent (CI parity), pyright clean, pylint 9.86.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
2026-07-17 03:26:33 -04:00
31 changed files with 1372 additions and 903 deletions
@@ -89,6 +89,14 @@ class MacosContainerBottleBackend(
with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle
def ensure_orchestrator(self) -> str:
"""Bring up the per-host infra container (control plane + gateway) and
return its control-plane URL — the on-demand entry point operator tools
(`supervise`) call when no control plane is running yet. Mirrors
firecracker's infra-VM bring-up."""
from .infra import MacosInfraService
return MacosInfraService().ensure_running().control_plane_url
def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:
return _cleanup.prepare_cleanup()
@@ -13,9 +13,13 @@ from .. import BottlePlan
class MacosContainerBottlePlan(BottlePlan):
slug: str
forwarded_env: dict[str, str] = field(repr=False)
agent_proxy_url: str = ""
agent_git_gate_url: str = ""
agent_supervise_url: str = ""
# Read by provision-time consumers (git extraHeader, supervise MCP header)
# via getattr(plan, "identity_token", ""); stamped in launch after the
# bottle is registered. See launch.py's stamp for why it lives here and not
# only in the exec-time proxy env.
identity_token: str = ""
@property
def container_name(self) -> str:
@@ -15,6 +15,9 @@ caller has to start the agent in between. `ensure_gateway` runs first because
the agent's proxy env needs the gateway's address at `container run` time; the
agent's *own* address (the attribution key) only exists afterwards.
The control plane and the gateway are one **infra container** here (see
`infra`), so `gateway_ip` and the control-plane host are the same address.
The consequence for the identity token: it is minted by registration, i.e.
*after* the agent container exists, so it cannot be baked into the run-time
env the way docker's compose spec does. It is delivered at `container exec`
@@ -38,7 +41,7 @@ from ...orchestrator.registration import registration_inputs
from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
from .gateway import GATEWAY_NETWORK
from .gateway_provision import AppleGatewayTransport
from .orchestrator_service import MacosOrchestratorService, OrchestratorStartError
from .infra import MacosInfraService, OrchestratorStartError
class ConsolidatedLaunchError(RuntimeError):
@@ -47,7 +50,9 @@ class ConsolidatedLaunchError(RuntimeError):
@dataclass(frozen=True)
class GatewayEndpoint:
"""What the agent `container run` needs to reach the shared gateway."""
"""What the agent `container run` needs to reach the shared gateway (the
infra container). `gateway_ip` is that container's host-only address, the
same host the control-plane URL points at."""
orchestrator_url: str
gateway_ip: str # the gateway's address — the agent's proxy target
@@ -68,19 +73,18 @@ class LaunchContext:
def ensure_gateway(
*, service: MacosOrchestratorService | None = None,
*, service: MacosInfraService | None = None,
) -> GatewayEndpoint:
"""Ensure the orchestrator control plane + shared gateway are up, and
report how to reach them. Idempotent — both are per-host singletons, so N
bottle launches share the one pair. Call before starting the agent
container: the agent's proxy env needs `gateway_ip` at run time."""
service = service or MacosOrchestratorService()
url = service.ensure_running()
gateway = service.gateway(url)
"""Ensure the per-host infra container (control plane + gateway) is up and
report how to reach it. Idempotent — one singleton, so N bottle launches
share it. Call before starting the agent container: the agent's proxy env
needs `gateway_ip` at run time."""
service = service or MacosInfraService()
infra = service.ensure_running()
return GatewayEndpoint(
orchestrator_url=url,
gateway_ip=gateway.ip_on_shared_network(),
gateway_ca_pem=gateway.ca_cert_pem(),
orchestrator_url=infra.control_plane_url,
gateway_ip=infra.gateway_ip,
gateway_ca_pem=service.ca_cert_pem(),
network=service.network,
)
@@ -6,14 +6,13 @@ import subprocess
from ...bottle_state import read_metadata
from .. import ActiveAgent
from .gateway import GATEWAY_NAME
from .orchestrator_service import ORCHESTRATOR_NAME
from .infra import INFRA_NAME
_PREFIX = "bot-bottle-"
# The shared per-host singletons carry the same prefix as agent containers but
# are infrastructure, not bottles one gateway and one control plane serve
# every agent, so listing them as agents would invent one per host.
_INFRA_NAMES = frozenset({GATEWAY_NAME, ORCHESTRATOR_NAME})
# The shared per-host infra container carries the same prefix as agent
# containers but is infrastructure, not a bottle — one control plane + gateway
# serves every agent, so listing it as an agent would invent one per host.
_INFRA_NAMES = frozenset({INFRA_NAME})
def enumerate_active() -> list[ActiveAgent]:
+17 -212
View File
@@ -1,241 +1,46 @@
"""The consolidated per-host gateway as an Apple container (PRD 0070).
"""Shared network/image constants for the macOS consolidated infra container.
The macOS counterpart of `orchestrator.gateway.DockerGateway`: one persistent
gateway per host, shared by every bottle, attributing each request to a bottle
by its source IP on the shared host-only network.
Two Apple Container 1.0.0 constraints shape this and make it *not* a
transliteration of the docker gateway:
- **No container DNS.** Containers cannot resolve each other by name (the
host-only network's resolver refuses the query), so the gateway reaches the
control plane by **IP**, not by name as the docker gateway does. The
orchestrator must therefore be started *before* the gateway — see
`orchestrator_service`.
- **Networks are fixed at `container run`.** There is no `network connect`,
so a network cannot be attached to a running container. The gateway must sit
on one shared, up-front network for the lifetime of the process; per-bottle
networks would mean restarting the gateway on every launch, which defeats the
consolidation.
The gateway is dual-homed, **NAT network first**: Apple Container makes the
first `--network` the default route, so the egress network must lead or the
gateway has no route to the internet.
The gateway data plane no longer runs as its own Apple container — it shares a
single per-host **infra container** with the control plane (see `infra`),
because two Apple-Container guests writing one `bot-bottle.db` over virtiofs
would race incoherent `fcntl` locks. This module holds the pieces both the
infra service and the launch/provision glue need: the network names, the
gateway image, and the network-creation helper.
"""
from __future__ import annotations
import os
import time
from pathlib import Path
from ...orchestrator.gateway import (
GATEWAY_CA_CERT,
GATEWAY_DOCKERFILE,
MITMPROXY_HOME,
Gateway,
GatewayError,
)
from ...paths import host_db_path
from ...supervise import DB_PATH_IN_CONTAINER
from ...orchestrator.gateway import GatewayError
from . import util as container_mod
# Distinct from the docker gateway's names so both backends' gateways can
# coexist on one host (a macOS host can run the docker backend too).
GATEWAY_NAME = "bot-bottle-mac-gateway"
# The shared host-only network the gateway and every agent bottle sit on. The
# agent's address here is the attribution key.
# The shared host-only network the infra container and every agent bottle sit
# on. The agent's address here is the attribution key. Distinct from the docker
# names so both backends can coexist on one host.
GATEWAY_NETWORK = "bot-bottle-mac-gateway"
# The NAT network that gives the gateway (and only the gateway) a route out.
# The NAT network that gives the infra container (and only it) a route out.
GATEWAY_EGRESS_NETWORK = "bot-bottle-mac-egress"
GATEWAY_IMAGE = os.environ.get("BOT_BOTTLE_GATEWAY_IMAGE", "bot-bottle-gateway:latest")
_REPO_ROOT = Path(__file__).resolve().parents[3]
_SUPERVISE_DB_DIR_IN_CONTAINER = os.path.dirname(DB_PATH_IN_CONTAINER)
# mitmproxy writes its CA a beat after start; reads poll rather than assume.
_CA_POLL_SECONDS = 0.5
DEFAULT_CA_TIMEOUT_SECONDS = 30.0
def gateway_ca_dir() -> Path:
"""Host dir bind-mounted as mitmproxy's home, keeping the gateway's
self-generated CA **stable across container recreation** — every agent
installs this one CA to trust the shared gateway's TLS interception, so it
must not rotate when the gateway restarts.
The docker gateway uses a named volume for this; a plain host dir is the
same guarantee with fewer moving parts, and it lets `ca_cert_pem` read the
PEM straight off the host instead of shelling into the container."""
path = host_db_path().parent / "mac-gateway-ca"
path.mkdir(parents=True, exist_ok=True)
return path
def ensure_networks(
network: str = GATEWAY_NETWORK, egress_network: str = GATEWAY_EGRESS_NETWORK,
) -> None:
"""Create the shared host-only network + the gateway's NAT network.
Idempotent — `create_network` tolerates 'already exists'.
Module-level rather than a gateway method because the **orchestrator**
needs the shared network too, and it starts first (Apple has no container
DNS, so the gateway must be handed the control plane's IP). Both callers
ensure the networks; whoever runs first wins."""
"""Create the shared host-only network + the NAT egress network. Idempotent
— `create_network` tolerates 'already exists'."""
container_mod.create_network(egress_network)
container_mod.create_network(network, internal=True)
def _host_db_dir() -> str:
db_dir = host_db_path().parent
db_dir.mkdir(parents=True, exist_ok=True)
return str(db_dir)
def _mount(source: str, target: str, *, readonly: bool = False) -> str:
spec = f"type=bind,source={source},target={target}"
if readonly:
spec += ",readonly"
return spec
class AppleGateway(Gateway):
"""The consolidated gateway as a single, fixed-name Apple container."""
def __init__(
self,
image_ref: str = GATEWAY_IMAGE,
*,
name: str = GATEWAY_NAME,
network: str = GATEWAY_NETWORK,
egress_network: str = GATEWAY_EGRESS_NETWORK,
orchestrator_url: str = "",
build_context: Path | None = None,
dockerfile: str | None = GATEWAY_DOCKERFILE,
) -> None:
self.image_ref = image_ref
self.name = name
self.network = network
self.egress_network = egress_network
# Reached by IP (no container DNS on Apple) — the caller resolves the
# orchestrator's address before constructing this. Empty → single-tenant.
self._orchestrator_url = orchestrator_url
self._build_context = build_context or _REPO_ROOT
self._dockerfile = dockerfile
def ensure_built(self) -> None:
"""Build the gateway data-plane image from its Dockerfile. Builds every
time (cache-aware, so it's cheap when nothing changed): a stale image
silently runs the OLD single-tenant daemons. Mirrors `DockerGateway`."""
if self._dockerfile is None:
return
container_mod.build_image(
self.image_ref, str(self._build_context), dockerfile=self._dockerfile,
)
def is_running(self) -> bool:
return container_mod.container_is_running(self.name)
def _running_image_is_current(self) -> bool:
"""True iff the running gateway was created from the *current*
`image_ref`. `ensure_built` rebuilding the image is not enough on its
own — the running container still holds the OLD image, so this
mismatch check is what makes a rebuild take effect."""
running = container_mod.container_image_digest(self.name)
current = container_mod.image_digest(self.image_ref)
if not running or not current:
return True # can't compare → don't churn a working container
return running == current
def _running_control_plane_is_current(self) -> bool:
"""True iff the running gateway points at the control plane we would
pass today.
Docker gets this for free — it hands the gateway a container *name*,
which survives the orchestrator being recreated. Apple has no container
DNS, so the URL is an **IP baked into the gateway's env at run time**,
and a recreated orchestrator can come back on a different DHCP address.
Without this check the gateway would keep pointing at the old address
and every `/resolve` would fail — denying egress for *every* bottle on
the host until something else happened to recreate the gateway."""
if not self._orchestrator_url:
return True
env = container_mod.container_env(self.name)
if not env:
return True # can't compare → don't churn a working container
return env.get("BOT_BOTTLE_ORCHESTRATOR_URL") == self._orchestrator_url
def ensure_running(self) -> None:
if (self.is_running()
and self._running_image_is_current()
and self._running_control_plane_is_current()):
return
ensure_networks(self.network, self.egress_network)
container_mod.force_remove_container(self.name)
argv = [
"container", "run", "--detach",
"--name", self.name,
"--label", "bot-bottle.backend=macos-container",
"--label", "bot-bottle-mac-gateway=1",
# NAT network FIRST: Apple Container takes the first --network as
# the default route, so this ordering is what gives the gateway a
# route out. Reversing it silently blackholes egress.
"--network", self.egress_network,
"--network", self.network,
# The NAT gateway routes but does not resolve, so DNS is explicit.
"--dns", container_mod.dns_server(),
"--mount", _mount(str(gateway_ca_dir()), MITMPROXY_HOME),
"--mount", _mount(_host_db_dir(), _SUPERVISE_DB_DIR_IN_CONTAINER),
"--env", f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
]
if self._orchestrator_url:
# Makes the data plane multi-tenant: each request resolves
# source-IP → policy against the control plane.
argv += ["--env", f"BOT_BOTTLE_ORCHESTRATOR_URL={self._orchestrator_url}"]
argv.append(self.image_ref)
result = container_mod.run_container_argv(argv)
if result.returncode != 0:
raise GatewayError(
f"gateway failed to start: "
f"{(result.stderr or '').strip() or '<no stderr>'}"
)
def ip_on_shared_network(self) -> str:
"""The gateway's address on the shared host-only network — what agents
point their proxy / git-http / supervise URLs at."""
return container_mod.container_ipv4_on_network(self.name, self.network)
def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
"""The gateway's CA certificate (PEM) that agents install to trust its
TLS interception. Polls: mitmproxy generates it a moment after start.
Read from the host bind-mount, so no exec into the container."""
ca_path = gateway_ca_dir() / os.path.basename(GATEWAY_CA_CERT)
deadline = time.monotonic() + timeout
while True:
try:
pem = ca_path.read_text()
if pem.strip():
return pem
except OSError:
pass
if time.monotonic() >= deadline:
raise GatewayError(
f"gateway CA cert not available at {ca_path} after {timeout:g}s"
)
time.sleep(_CA_POLL_SECONDS)
def stop(self) -> None:
container_mod.force_remove_container(self.name)
__all__ = [
"AppleGateway",
"GatewayError",
"GATEWAY_NAME",
"GATEWAY_NETWORK",
"GATEWAY_EGRESS_NETWORK",
"GATEWAY_IMAGE",
"gateway_ca_dir",
"GatewayError",
"DEFAULT_CA_TIMEOUT_SECONDS",
"ensure_networks",
]
@@ -1,23 +1,23 @@
"""`GatewayTransport` for the Apple gateway container (PRD 0070).
"""`GatewayTransport` for the Apple infra container (PRD 0070).
The provisioning *logic* (per-bottle creds dirs, namespaced repo init) is
backend-neutral and lives in `backend.docker.gateway_provision`; this is only
the transport — how files and commands reach the running gateway. Docker uses
`docker exec`/`docker cp` and Firecracker uses SSH; Apple uses the `container`
CLI's equivalents.
CLI's equivalents against the infra container that hosts the gateway daemons.
"""
from __future__ import annotations
from ..docker.gateway_provision import GatewayProvisionError
from . import util as container_mod
from .gateway import GATEWAY_NAME
from .infra import INFRA_NAME
class AppleGatewayTransport:
"""`GatewayTransport` for the gateway as an Apple container."""
"""`GatewayTransport` for the gateway daemons in the Apple infra container."""
def __init__(self, gateway: str = GATEWAY_NAME) -> None:
def __init__(self, gateway: str = INFRA_NAME) -> None:
self.gateway = gateway
def exec(self, argv: list[str]) -> None:
+305
View File
@@ -0,0 +1,305 @@
"""The per-host infra container for the macOS backend (PRD 0070).
A single persistent Apple container that runs BOTH the orchestrator control
plane and the gateway data plane — the macOS analogue of the Firecracker infra
VM (`backend/firecracker/infra_vm.py`), not the docker backend's two separate
containers.
Why one container, not two: Apple Containers are lightweight VMs, each with its
own kernel. The docker backend runs the orchestrator and gateway as two
containers safely because they share the host kernel, so their concurrent
writes to the one `bot-bottle.db` (the orchestrator's registry + the gateway
supervise daemon's queue) are serialized by coherent `fcntl` locks. Across two
*guest* kernels sharing a virtiofs-mounted DB those locks are not coherent, and
concurrent writers can corrupt the file. Firecracker solved this by putting
both services in one guest with the DB on a device only that guest mounts; this
does the same with Apple primitives.
Two consequences fall out of the single container, both simplifications:
- **No DNS dance.** The control plane and the gateway daemons reach each other
over `127.0.0.1`, so nothing depends on Apple's (absent) container DNS and
there is no orchestrator-before-gateway ordering to get right.
- **The DB is never host-shared.** It lives on a container-only volume, so no
host process opens the live file. The host CLI reaches registry + supervise
state through the control-plane HTTP surface (`cli/supervise.py` already uses
`OrchestratorClient`), exactly as it does for firecracker.
The control-plane source is bind-mounted (like the docker orchestrator), so a
code change takes effect on the next launch without an image rebuild; the
gateway daemons are baked in the gateway image and rebuild through its own
digest check.
"""
from __future__ import annotations
import os
import time
import urllib.error
import urllib.request
from dataclasses import dataclass
from pathlib import Path
from ... import log
from ...orchestrator.gateway import GATEWAY_CA_CERT
from ...orchestrator.lifecycle import (
DEFAULT_PORT,
DEFAULT_STARTUP_TIMEOUT_SECONDS,
OrchestratorStartError,
source_hash,
)
from ...paths import (
CONTROL_PLANE_TOKEN_ENV,
HOST_DB_FILENAME,
host_control_plane_token,
)
from . import util as container_mod
from .gateway import (
DEFAULT_CA_TIMEOUT_SECONDS,
GATEWAY_EGRESS_NETWORK,
GATEWAY_IMAGE,
GATEWAY_NETWORK,
GatewayError,
ensure_networks,
)
# The one per-host infra container: control plane + gateway data plane.
INFRA_NAME = "bot-bottle-mac-infra"
INFRA_LABEL = "bot-bottle-mac-infra=1"
# Container-only volume holding bot-bottle.db. No host bind-mount, so the DB is
# written by exactly one kernel (this container's). Survives recreation.
INFRA_DB_VOLUME = "bot-bottle-mac-db"
# BOT_BOTTLE_ROOT inside the container; host_db_path() resolves the DB to
# <root>/db/<filename> and the supervise daemon writes the same file.
_DB_ROOT_IN_CONTAINER = "/var/lib/bot-bottle"
_DB_PATH_IN_CONTAINER = f"{_DB_ROOT_IN_CONTAINER}/db/{HOST_DB_FILENAME}"
_SRC_IN_CONTAINER = "/bot-bottle-src"
_REPO_ROOT = Path(__file__).resolve().parents[3]
_HEALTH_POLL_SECONDS = 0.25
_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
_CA_POLL_SECONDS = 0.5
# The gateway subset the consolidated model runs (no per-bottle git:// daemon).
_GATEWAY_DAEMONS = "egress,git-http,supervise"
def _init_script(port: int) -> str:
"""PID-1 init: start the control plane and the gateway daemons, both in
this container, reaching each other over loopback. Backgrounded so `wait`
reaps as PID 1. No `set -e` — a transient daemon failure must not kill the
whole container (gateway_init applies the same 'stay up' policy)."""
return (
"export PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin\n"
f"mkdir -p $(dirname {_DB_PATH_IN_CONTAINER})\n"
# Control plane, from the bind-mounted source (stdlib-only package).
f"( cd {_SRC_IN_CONTAINER} && BOT_BOTTLE_ROOT={_DB_ROOT_IN_CONTAINER} "
f"python3 -m bot_bottle.orchestrator --host 0.0.0.0 --port {port} "
"--broker stub ) &\n"
# Gateway data plane, multi-tenant against the local control plane.
f"( cd /app && BOT_BOTTLE_GATEWAY_DAEMONS={_GATEWAY_DAEMONS} "
f"BOT_BOTTLE_ORCHESTRATOR_URL=http://127.0.0.1:{port} "
f"SUPERVISE_DB_PATH={_DB_PATH_IN_CONTAINER} python3 /app/gateway_init.py ) &\n"
"while : ; do wait ; done\n"
)
@dataclass(frozen=True)
class InfraEndpoint:
"""How to reach the running infra container. The control plane and the
gateway are the same container, so one address serves both."""
control_plane_url: str # http://<infra ip>:8099 — host CLI + registration
gateway_ip: str # same container; agents' proxy / git-http / MCP target
class MacosInfraService:
"""Manages the single per-host infra container. Callers use
`ensure_running()` (returns the endpoint) and `ca_cert_pem()`."""
def __init__(
self,
*,
port: int = DEFAULT_PORT,
network: str = GATEWAY_NETWORK,
egress_network: str = GATEWAY_EGRESS_NETWORK,
image: str = GATEWAY_IMAGE,
repo_root: Path = _REPO_ROOT,
name: str = INFRA_NAME,
db_volume: str = INFRA_DB_VOLUME,
) -> None:
self.port = port
self.network = network
self.egress_network = egress_network
self.image = image
self._repo_root = repo_root
self._name = name
self._db_volume = db_volume
def _resolve_url(self) -> str:
"""The control-plane URL, or "" while the container has no address."""
ip = container_mod.try_container_ipv4_on_network(self._name, self.network)
return f"http://{ip}:{self.port}" if ip else ""
def is_healthy(
self, url: str, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS,
) -> bool:
if not url:
return False
try:
with urllib.request.urlopen(f"{url}/health", timeout=timeout) as resp:
return resp.status == 200
except (urllib.error.URLError, TimeoutError, OSError):
return False
def _source_current(self, current_hash: str) -> bool:
"""True iff the running infra container was created from the current
bind-mounted control-plane source. The control-plane process loads that
code at startup and won't reload it, so a stale container keeps serving
OLD code."""
if not container_mod.container_is_running(self._name):
return False
env = container_mod.container_env(self._name)
if not env:
return True # can't compare → don't churn a working container
return env.get("BOT_BOTTLE_SOURCE_HASH") == current_hash
def _running_healthy_endpoint(self, current_hash: str) -> InfraEndpoint | None:
"""The endpoint if the running container is BOTH source-current and
answering /health, else None (→ recreate). Health, not just the source
label, is what lets a wedged-but-current container self-heal instead of
being polled to death forever."""
if not self._source_current(current_hash):
return None
url = self._resolve_url()
if url and self.is_healthy(url):
return InfraEndpoint(control_plane_url=url, gateway_ip=_ip_of(url))
return None
def ensure_built(self) -> None:
"""Ensure the gateway data-plane image exists. The control-plane source
is bind-mounted, not baked, so only the gateway image needs building."""
container_mod.build_image(
self.image, str(self._repo_root), dockerfile="Dockerfile.gateway",
)
def ensure_running(
self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
) -> InfraEndpoint:
"""Ensure the single infra container is up; return how to reach it.
Idempotent per-host singleton — a healthy container on current source
is left untouched, so N launches share the one control plane + gateway.
Raises `OrchestratorStartError` on startup timeout."""
current_hash = source_hash(self._repo_root)
endpoint = self._running_healthy_endpoint(current_hash)
if endpoint is not None:
return endpoint
self.ensure_built()
log.info("starting infra container", context={"name": self._name})
self._run_container(current_hash)
return self._wait_healthy(startup_timeout)
def _run_container(self, current_hash: str) -> None:
ensure_networks(self.network, self.egress_network)
container_mod.force_remove_container(self._name)
argv = [
"container", "run", "--detach",
"--name", self._name,
"--label", "bot-bottle.backend=macos-container",
"--label", INFRA_LABEL,
# NAT network FIRST so the gateway's egress has a default route;
# the host-only network is where agents (and the host CLI) reach it.
"--network", self.egress_network,
"--network", self.network,
"--dns", container_mod.dns_server(),
# Container-only DB volume: one kernel writes bot-bottle.db, never
# shared with the host or another guest.
"--volume", f"{self._db_volume}:{_DB_ROOT_IN_CONTAINER}",
# Bind-mount the control-plane source (read-only); a code change
# takes effect on relaunch with no image rebuild.
"--mount",
container_mod.bind_mount_spec(
str(self._repo_root), _SRC_IN_CONTAINER, readonly=True),
# Baked onto the container so `_source_current` can detect a real
# control-plane code change and recreate.
"--env", f"BOT_BOTTLE_SOURCE_HASH={current_hash}",
# The control-plane secret, for BOTH the control plane (to require
# it) and the gateway's PolicyResolver (to present it) — they share
# this one container. Bare `--env NAME` inherits the value from the
# run process below, so the secret never lands on argv or in
# `container inspect`'s command line. The agent runs in a SEPARATE
# container that is never given this var, which is the whole point.
"--env", CONTROL_PLANE_TOKEN_ENV,
"--entrypoint", "sh",
self.image,
"-c", _init_script(self.port),
]
run_env = {**os.environ, CONTROL_PLANE_TOKEN_ENV: host_control_plane_token()}
result = container_mod.run_container_argv(argv, env=run_env)
if result.returncode != 0:
raise OrchestratorStartError(
f"infra container failed to start: "
f"{(result.stderr or '').strip() or '<no stderr>'}"
)
def _wait_healthy(self, startup_timeout: float) -> InfraEndpoint:
deadline = time.monotonic() + startup_timeout
while True:
url = self._resolve_url()
if url and self.is_healthy(url):
log.info("infra container healthy", context={"url": url})
return InfraEndpoint(control_plane_url=url, gateway_ip=_ip_of(url))
if time.monotonic() >= deadline:
raise OrchestratorStartError(
f"infra container did not become healthy within "
f"{startup_timeout:g}s"
)
time.sleep(_HEALTH_POLL_SECONDS)
def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
"""The gateway's mitmproxy CA (PEM) agents install to trust its TLS
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:
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)
def stop(self) -> None:
"""Remove the infra container (idempotent). The DB volume persists."""
container_mod.force_remove_container(self._name)
def _ip_of(url: str) -> str:
"""The host from an http://host:port URL."""
return url.split("://", 1)[-1].rsplit(":", 1)[0]
def probe_control_plane_url(port: int = DEFAULT_PORT) -> str:
"""The running infra container's control-plane URL, or "" if it isn't up.
Used by host-side control-plane discovery (`discover_orchestrator_url`);
safe to call on any host — returns "" when the container or the `container`
CLI isn't present."""
ip = container_mod.try_container_ipv4_on_network(INFRA_NAME, GATEWAY_NETWORK)
return f"http://{ip}:{port}" if ip else ""
__all__ = [
"MacosInfraService",
"InfraEndpoint",
"OrchestratorStartError",
"GatewayError",
"INFRA_NAME",
"INFRA_DB_VOLUME",
]
+18 -4
View File
@@ -52,6 +52,7 @@ from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...git_http_backend import DEFAULT_PORT as _GIT_HTTP_PORT
from ...log import die, info, warn
from ...supervise import SUPERVISE_PORT
from ..docker.egress import EGRESS_PORT
@@ -68,7 +69,6 @@ from .consolidated_launch import (
_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
_AGENT_SLEEP_SECONDS = "2147483647"
_GIT_HTTP_PORT = 9420
@contextmanager
@@ -115,10 +115,16 @@ def launch(
# Step 4: read the assigned address and register by it. This is the
# attribution key; `--cap-drop CAP_NET_RAW` at run is what makes it
# unforgeable.
source_ip = container_mod.container_ipv4_on_network(
# unforgeable. Poll: `container run --detach` can return before vmnet's
# DHCP has assigned the address.
source_ip = container_mod.wait_container_ipv4_on_network(
plan.container_name, endpoint.network,
)
if not source_ip:
die(
f"agent {plan.container_name} never got an address on "
f"{endpoint.network}"
)
effective_env = {**os.environ, **plan.agent_provision.provisioned_env}
token_values = egress_resolve_token_values(
plan.egress_plan.token_env_map, effective_env,
@@ -140,6 +146,15 @@ def launch(
f"(gateway {endpoint.gateway_ip}, ip {source_ip})"
)
# Stamp the token onto the plan so provision-time consumers can read it,
# not only the exec-time egress proxy. git-gate's gitconfig extraHeader
# and the supervise MCP --header both reach the gateway on NO_PROXY (they
# bypass the egress proxy that carries the token), so without this the
# gateway's /resolve fail-closes and every git fetch/push and supervise
# call from the bottle is denied. Registration already produced the
# token above, so — unlike the run-time env — the plan CAN carry it.
plan = dataclasses.replace(plan, identity_token=ctx.identity_token)
bottle = MacosContainerBottle(
plan.container_name,
teardown,
@@ -227,7 +242,6 @@ def _stamp_agent_urls(
)
return dataclasses.replace(
plan,
agent_proxy_url=f"http://{endpoint.gateway_ip}:{EGRESS_PORT}",
agent_git_gate_url=git_gate_url,
agent_supervise_url=supervise_url,
)
@@ -1,241 +0,0 @@
"""Orchestrator + gateway lifecycle for the macOS backend (PRD 0070).
The macOS counterpart of `orchestrator.lifecycle.OrchestratorService`. Same
shape — an idempotent singleton that brings up the control plane and the shared
gateway and hands back the control-plane URL — but the startup **order is
reversed**, and the reason is a real Apple Container constraint rather than a
stylistic choice:
- Docker starts the gateway first and lets it find the control plane by
*container name* over docker DNS.
- Apple Container 1.0.0 has **no container DNS** (see `gateway`), so the
gateway can only be handed an **IP**. That IP does not exist until the
orchestrator container is running — hence: orchestrator first, read its
address, then start the gateway pointed at it.
The second difference: no published port. Apple Container puts every container
on a host-reachable address, and the host can reach the host-only network
directly, so the host CLI and the gateway use the **same** URL — the
orchestrator's address on the shared network. Docker needs a
`--publish 127.0.0.1:…` hop plus a separate `internal_url` for the same job.
Like the docker service this runs the **register-only broker**: the backend
launches agent containers, so the control plane needs no privileged socket.
"""
from __future__ import annotations
import time
import urllib.error
import urllib.request
from pathlib import Path
from ... import log
from ...orchestrator.lifecycle import (
DEFAULT_PORT,
DEFAULT_STARTUP_TIMEOUT_SECONDS,
ORCHESTRATOR_DOCKERFILE,
ORCHESTRATOR_IMAGE,
ORCHESTRATOR_SOURCE_HASH_LABEL,
OrchestratorStartError,
source_hash,
)
from ...paths import bot_bottle_root
from . import util as container_mod
from .gateway import (
GATEWAY_EGRESS_NETWORK,
GATEWAY_IMAGE,
GATEWAY_NETWORK,
AppleGateway,
ensure_networks,
)
# Distinct from the docker backend's container name so both can run on one host.
ORCHESTRATOR_NAME = "bot-bottle-mac-orchestrator"
_REPO_ROOT = Path(__file__).resolve().parents[3]
_APP_DIR = "/app"
_ROOT_IN_CONTAINER = "/bot-bottle-root"
_HEALTH_POLL_SECONDS = 0.25
_HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
def _mount(source: str, target: str, *, readonly: bool = False) -> str:
spec = f"type=bind,source={source},target={target}"
if readonly:
spec += ",readonly"
return spec
class MacosOrchestratorService:
"""Manages the orchestrator control-plane container + the shared Apple
gateway. Callers only need `ensure_running()`, which returns the URL."""
def __init__(
self,
*,
port: int = DEFAULT_PORT,
network: str = GATEWAY_NETWORK,
egress_network: str = GATEWAY_EGRESS_NETWORK,
image: str = ORCHESTRATOR_IMAGE,
gateway_image: str = GATEWAY_IMAGE,
repo_root: Path = _REPO_ROOT,
host_root: Path | None = None,
orchestrator_name: str = ORCHESTRATOR_NAME,
) -> None:
self.port = port
self.network = network
self.egress_network = egress_network
self.image = image
self._gateway_image = gateway_image
self._repo_root = repo_root
self._host_root = host_root or bot_bottle_root()
self._orchestrator_name = orchestrator_name
# Resolved once the container is up — there is no name to fall back on.
self._url = ""
@property
def url(self) -> str:
"""The control-plane URL, or "" before the container is up. One URL for
both the host CLI and the gateway (see the module docstring)."""
return self._url
def _resolve_url(self) -> str:
"""The control-plane URL, or "" while the container has no address."""
ip = container_mod.try_container_ipv4_on_network(
self._orchestrator_name, self.network,
)
return f"http://{ip}:{self.port}" if ip else ""
def is_healthy(
self, url: str = "", *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS,
) -> bool:
target = url or self._url
if not target:
return False
try:
with urllib.request.urlopen(f"{target}/health", timeout=timeout) as resp:
return resp.status == 200
except (urllib.error.URLError, TimeoutError, OSError):
return False
def _ensure_orchestrator_image(self) -> None:
"""Build the lean control-plane image when missing. 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, not a rebuild."""
if container_mod.image_exists(self.image):
return
container_mod.build_image(
self.image, str(self._repo_root), dockerfile=ORCHESTRATOR_DOCKERFILE,
)
def _orchestrator_source_current(self, current_hash: str) -> bool:
"""True iff the running orchestrator was created from the *current*
bind-mounted source. The process loaded that code at startup and won't
reload it, so a stale container would keep serving OLD control-plane
code."""
if not container_mod.container_is_running(self._orchestrator_name):
return False
data = container_mod.inspect_container(self._orchestrator_name)
config = data.get("configuration")
labels = config.get("labels") if isinstance(config, dict) else None
if not isinstance(labels, dict):
return True # can't compare → don't churn a working container
return labels.get(ORCHESTRATOR_SOURCE_HASH_LABEL) == current_hash
def _run_orchestrator_container(self, current_hash: str) -> None:
# The orchestrator is the first thing on the shared network, so it —
# not the gateway — is what has to bring the network into existence.
ensure_networks(self.network, self.egress_network)
container_mod.force_remove_container(self._orchestrator_name)
argv = [
"container", "run", "--detach",
"--name", self._orchestrator_name,
"--label", "bot-bottle.backend=macos-container",
"--label", f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current_hash}",
# Host-only network only: the control plane needs no route out, and
# the host reaches it here directly (no --publish needed).
"--network", self.network,
"--mount", _mount(str(self._repo_root), _APP_DIR, readonly=True),
"--workdir", _APP_DIR,
# Persist the registry DB on the host (sole-owner: only the
# orchestrator opens bot-bottle.db).
"--mount", _mount(str(self._host_root), _ROOT_IN_CONTAINER),
"--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}",
"--entrypoint", "python3",
self.image,
"-m", "bot_bottle.orchestrator",
"--host", "0.0.0.0", "--port", str(self.port), "--broker", "stub",
]
result = container_mod.run_container_argv(argv)
if result.returncode != 0:
raise OrchestratorStartError(
f"orchestrator container failed to start: "
f"{(result.stderr or '').strip() or '<no stderr>'}"
)
def gateway(self, orchestrator_url: str) -> AppleGateway:
"""The shared gateway, pointed at the control plane at
`orchestrator_url` (by IP — Apple has no container DNS)."""
return AppleGateway(
self._gateway_image,
network=self.network,
egress_network=self.egress_network,
orchestrator_url=orchestrator_url,
)
def ensure_running(
self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
) -> str:
"""Ensure the control plane + shared gateway are up; return the
control-plane URL. Idempotent — a healthy control plane running current
code and a running gateway are left untouched."""
current_hash = source_hash(self._repo_root)
if not self._orchestrator_source_current(current_hash):
self._ensure_orchestrator_image()
log.info(
"starting orchestrator container",
context={"name": self._orchestrator_name},
)
self._run_orchestrator_container(current_hash)
url = self._wait_healthy(startup_timeout)
self._url = url
# Gateway second: it can only reach the control plane by IP, which does
# not exist until the orchestrator container is up (see the docstring).
gateway = self.gateway(url)
gateway.ensure_built()
gateway.ensure_running()
return url
def _wait_healthy(self, startup_timeout: float) -> str:
"""Poll until the control plane answers /health, resolving its address
each time: the container is up before it has an IP, and it has an IP
before the server binds."""
deadline = time.monotonic() + startup_timeout
while True:
url = self._resolve_url()
if url and self.is_healthy(url):
log.info("orchestrator healthy", context={"url": url})
return url
if time.monotonic() >= deadline:
raise OrchestratorStartError(
f"orchestrator did not become healthy within "
f"{startup_timeout:g}s"
)
time.sleep(_HEALTH_POLL_SECONDS)
def stop(self) -> None:
"""Remove the orchestrator + gateway containers (idempotent)."""
container_mod.force_remove_container(self._orchestrator_name)
self.gateway("").stop()
self._url = ""
__all__ = [
"MacosOrchestratorService",
"OrchestratorStartError",
"ORCHESTRATOR_NAME",
]
+80 -66
View File
@@ -445,85 +445,89 @@ def container_ipv4_on_network(name: str, network: str) -> str:
return ip
def run_container_argv(argv: list[str]) -> subprocess.CompletedProcess[str]:
def run_container_argv(
argv: list[str], *, env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
"""Run a `container` command, returning the result for the caller to
interpret. Unlike the `die`-on-failure helpers above, this lets callers
that raise their own typed errors (the gateway / orchestrator lifecycle)
keep control of the failure path."""
return subprocess.run(argv, capture_output=True, text=True, check=False)
keep control of the failure path.
`env` sets the child process environment — used to hand a secret to a bare
`--env NAME` flag (Apple's "just key → inherit from host" form) so the
value is inherited from this process, never written onto argv or into
`container inspect`'s recorded command line."""
return subprocess.run(
argv, capture_output=True, text=True, check=False, env=env)
def bind_mount_spec(source: str, target: str, *, readonly: bool = False) -> str:
"""A `container run --mount` bind spec. One definition so the gateway and
orchestrator emit an identical string — a divergence here would silently
break one backend's mounts while the other kept working."""
spec = f"type=bind,source={source},target={target}"
if readonly:
spec += ",readonly"
return spec
def _normalize_digest(value: str) -> str:
return value.split(":", 1)[1] if ":" in value else value
def image_digest(ref: str) -> str:
"""The digest of image `ref`, or "" if it can't be read. Empty is a
'don't know' signal — callers treat it as "don't churn a working
container" rather than as a mismatch."""
result = run_container_argv([_CONTAINER, "image", "inspect", ref])
def _inspect_first(argv: list[str]) -> dict[str, object]:
"""Run an inspect command and return its first JSON object, or {} on any
failure (non-zero exit, malformed JSON, unexpected shape). {} is the shared
'don't know' signal all the non-fatal inspect readers below build on — a
caller comparing against it treats it as 'leave the working container
alone', never as a mismatch."""
result = run_container_argv(argv)
if result.returncode != 0:
return ""
return {}
try:
data = json.loads(result.stdout or "{}")
data = json.loads(result.stdout or "[]")
except json.JSONDecodeError:
return {}
if isinstance(data, list):
data = data[0] if data else {}
return data if isinstance(data, dict) else {}
def _descriptor_digest(node: object) -> str:
"""The normalized digest under a `{... "descriptor": {"digest": ...}}`
node, or "". Both the image and container inspect shapes nest the image's
identity this way, so the digest readers stay symmetric — a difference
between them is what would spuriously recreate a container."""
if not isinstance(node, dict):
return ""
if isinstance(data, list) and data:
data = data[0]
if not isinstance(data, dict):
return ""
config = data.get("configuration")
if isinstance(config, dict):
descriptor = config.get("descriptor")
if isinstance(descriptor, dict) and descriptor.get("digest"):
return _normalize_digest(str(descriptor["digest"]))
value = data.get("id")
return _normalize_digest(str(value)) if value else ""
descriptor = node.get("descriptor")
if isinstance(descriptor, dict) and descriptor.get("digest"):
return _normalize_digest(str(descriptor["digest"]))
return ""
def image_digest(ref: str) -> str:
"""The digest of image `ref`, or "" if it can't be read. Reads exactly the
field `container_image_digest` reads (`configuration.descriptor.digest`) so
the two are comparable; "" means 'don't know' → callers don't churn."""
data = _inspect_first([_CONTAINER, "image", "inspect", ref])
return _descriptor_digest(data.get("configuration"))
def container_image_digest(name: str) -> str:
"""The digest of the image container `name` was created from, or "" if it
can't be read. Compare with `image_digest(ref)` to tell whether a running
container predates an image rebuild."""
result = run_container_argv([_CONTAINER, "inspect", name])
if result.returncode != 0:
return ""
try:
data = json.loads(result.stdout or "[]")
except json.JSONDecodeError:
return ""
if isinstance(data, list) and data:
data = data[0]
if not isinstance(data, dict):
return ""
config = data.get("configuration")
if not isinstance(config, dict):
return ""
image = config.get("image")
if not isinstance(image, dict):
return ""
descriptor = image.get("descriptor")
if isinstance(descriptor, dict) and descriptor.get("digest"):
return _normalize_digest(str(descriptor["digest"]))
return ""
config = _inspect_first([_CONTAINER, "inspect", name]).get("configuration")
image = config.get("image") if isinstance(config, dict) else None
return _descriptor_digest(image)
def container_env(name: str) -> dict[str, str]:
"""The env container `name` was started with, or {} if unreadable. Lets a
caller tell whether a running container's baked-in configuration still
matches what it would pass today."""
result = run_container_argv([_CONTAINER, "inspect", name])
if result.returncode != 0:
return {}
try:
data = json.loads(result.stdout or "[]")
except json.JSONDecodeError:
return {}
if isinstance(data, list) and data:
data = data[0]
if not isinstance(data, dict):
return {}
config = data.get("configuration")
config = _inspect_first([_CONTAINER, "inspect", name]).get("configuration")
init = config.get("initProcess") if isinstance(config, dict) else None
entries = init.get("environment") if isinstance(init, dict) else None
if not isinstance(entries, list):
@@ -540,18 +544,7 @@ def try_container_ipv4_on_network(name: str, network: str) -> str:
"""`container_ipv4_on_network` without the fatal exit: "" when the address
isn't readable yet. For pollers — a container is created before it has an
address, so "not yet" is an expected state there, not an error."""
result = run_container_argv([_CONTAINER, "inspect", name])
if result.returncode != 0:
return ""
try:
data = json.loads(result.stdout or "[]")
except json.JSONDecodeError:
return ""
if isinstance(data, list) and data:
data = data[0]
if not isinstance(data, dict):
return ""
status = data.get("status")
status = _inspect_first([_CONTAINER, "inspect", name]).get("status")
networks = status.get("networks") if isinstance(status, dict) else None
if not isinstance(networks, list):
return ""
@@ -564,6 +557,27 @@ def try_container_ipv4_on_network(name: str, network: str) -> str:
return ""
def wait_container_ipv4_on_network(
name: str, network: str, *, timeout: float = 15.0, poll: float = 0.25,
) -> str:
"""Poll for the container's DHCP-assigned address on `network`, returning
it once available or "" on timeout.
Apple Container has no `--ip`: `container run --detach` can return before
vmnet's DHCP has populated `status.networks[].ipv4Address`, so a bare read
right after start races the assignment. Callers that need the address (the
attribution key, the gateway's proxy target) poll through here instead of
the fatal `container_ipv4_on_network`."""
deadline = time.monotonic() + timeout
while True:
ip = try_container_ipv4_on_network(name, network)
if ip:
return ip
if time.monotonic() >= deadline:
return ""
time.sleep(poll)
def image_id(ref: str) -> str:
"""Return the image digest/ID from `container image inspect`.
+10 -3
View File
@@ -14,13 +14,20 @@ from __future__ import annotations
import subprocess
def run_docker(argv: list[str]) -> subprocess.CompletedProcess[str]:
def run_docker(
argv: list[str], *, env: dict[str, str] | None = None,
) -> subprocess.CompletedProcess[str]:
"""Run a `docker` command, capturing stdout/stderr as text. Never raises
on a non-zero exit callers inspect `returncode` / `stderr` so they can
stay fail-closed or tolerate idempotent no-ops (e.g. removing an
already-absent container)."""
already-absent container).
`env` sets the child process environment used to hand a secret to a bare
`--env NAME` flag (docker inherits its value from this process) so the
value never lands on argv or in `docker inspect`'s recorded command line."""
return subprocess.run(
argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True, check=False,
argv, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True,
check=False, env=env,
)
+76 -21
View File
@@ -78,6 +78,16 @@ ORCHESTRATOR_URL_ENV = "BOT_BOTTLE_ORCHESTRATOR_URL"
# is still stripped defensively (git-http uses that header on its own port).
IDENTITY_HEADER = "x-bot-bottle-identity"
# Per-flow key under which `request()` stashes the resolved (Config, supervise
# slug, env) so the later `response()` and `websocket_message()` hooks scan
# against the *calling bottle's* policy. In the consolidated (multi-tenant)
# gateway the static `self.config` is empty — every request's real policy comes
# from the per-request `/resolve` — so a hook that fell back to `self.config`
# would find no route and silently skip its DLP scan (fail-open). Resolving once
# at the request and reusing it also avoids a `/resolve` round-trip per response
# and per WebSocket frame.
_FLOW_CTX_KEY = "bot_bottle_egress_ctx"
def _token_from_proxy_auth(header: str) -> str:
"""Extract the identity token (the password) from a `Proxy-Authorization:
@@ -223,31 +233,39 @@ class EgressAddon:
{"Content-Type": "text/plain; charset=utf-8"},
)
def _log_request(self, flow: http.HTTPFlow) -> None:
def _log_request(
self, flow: http.HTTPFlow, env: "typing.Mapping[str, str]",
) -> None:
# `env` is the per-flow resolved overlay (process env + this bottle's
# /resolve tokens), so the log redaction scrubs the calling bottle's
# provisioned secrets — not just the process-level ones in os.environ.
headers = {
k: redact_tokens(v, env=os.environ)
k: redact_tokens(v, env=env)
for k, v in flow.request.headers.items()
if k.lower() != "authorization"
}
body = redact_tokens(flow.request.get_text(strict=False) or "", env=os.environ)
body = redact_tokens(flow.request.get_text(strict=False) or "", env=env)
sys.stderr.write(
json.dumps({
"event": "egress_request",
"host": redact_tokens(flow.request.pretty_host, env=os.environ),
"host": redact_tokens(flow.request.pretty_host, env=env),
"method": flow.request.method,
"path": redact_tokens(flow.request.path, env=os.environ),
"path": redact_tokens(flow.request.path, env=env),
"headers": headers,
"body": body,
})
+ "\n"
)
def _log_response(self, flow: http.HTTPFlow) -> None:
def _log_response(
self, flow: http.HTTPFlow, env: "typing.Mapping[str, str]",
) -> None:
# Per-flow env overlay (see _log_request): redact this bottle's tokens.
headers = {
k: redact_tokens(v, env=os.environ)
k: redact_tokens(v, env=env)
for k, v in flow.response.headers.items()
}
body = redact_tokens(flow.response.get_text(strict=False) or "", env=os.environ)
body = redact_tokens(flow.response.get_text(strict=False) or "", env=env)
sys.stderr.write(
json.dumps({
"event": "egress_response",
@@ -280,6 +298,36 @@ class EgressAddon:
env = {**os.environ, **tokens} if tokens else os.environ
return config, slug, env
def _stash_flow_ctx(
self,
flow: http.HTTPFlow,
config: Config,
slug: str,
env: "typing.Mapping[str, str]",
) -> None:
"""Remember the per-flow context `request()` resolved, so the later
`response()` / `websocket_message()` hooks reuse it scanning against
the same bottle's policy the request was decided on, with one `/resolve`
per flow rather than one per frame."""
meta = getattr(flow, "metadata", None)
if isinstance(meta, dict):
meta[_FLOW_CTX_KEY] = (config, slug, env)
def _flow_ctx(
self, flow: http.HTTPFlow,
) -> "tuple[Config, str, typing.Mapping[str, str]]":
"""The `(Config, supervise slug, env)` `request()` resolved for this
flow, so a later hook scans against the calling bottle's policy — not the
empty static config the consolidated gateway carries. Falls back to the
single-tenant static values for a flow that never passed through
`request()` (or a flow object without metadata)."""
meta = getattr(flow, "metadata", None)
if isinstance(meta, dict):
ctx = meta.get(_FLOW_CTX_KEY)
if ctx is not None:
return ctx
return self.config, self._supervise_slug, os.environ
def _request_token(self, flow: http.HTTPFlow) -> str:
"""The per-bottle identity token for this request, from the proxy
credentials the delivery mechanism (`HTTPS_PROXY=http://id:token@gw`)
@@ -319,6 +367,9 @@ class EgressAddon:
return
config, slug, env = self._resolve_flow(flow)
# Stash for the response / websocket hooks so their DLP scans use this
# bottle's resolved policy, not the empty static config (see _flow_ctx).
self._stash_flow_ctx(flow, config, slug, env)
# DLP outbound scan BEFORE stripping auth — catches tokens the
# agent tried to smuggle in any header, path, query param, or body.
@@ -377,7 +428,7 @@ class EgressAddon:
flow.request.headers["authorization"] = decision.inject_authorization
if config.log >= LOG_FULL:
self._log_request(flow)
self._log_request(flow, env)
def _block_dlp(self, flow: http.HTTPFlow, result: ScanResult) -> None:
ctx = self._req_ctx(flow)
@@ -586,14 +637,17 @@ class EgressAddon:
await asyncio.sleep(TOKEN_ALLOW_POLL_INTERVAL_SECONDS)
def response(self, flow: http.HTTPFlow) -> None:
"""DLP inbound scan on response headers and body."""
route = match_route(self.config.routes, flow.request.pretty_host)
"""DLP inbound scan on response headers and body, against the calling
bottle's resolved config (multi-tenant) or the static config
(single-tenant) see `_flow_ctx`."""
config, _slug, env = self._flow_ctx(flow)
route = match_route(config.routes, flow.request.pretty_host)
if route is None:
return
if flow.response is None:
return
if self.config.log >= LOG_FULL:
self._log_response(flow)
if config.log >= LOG_FULL:
self._log_response(flow, env)
resp_headers = {k.lower(): v for k, v in flow.response.headers.items()}
body = flow.response.get_text(strict=False) or ""
scan_text = build_inbound_scan_text(resp_headers, body)
@@ -610,7 +664,7 @@ class EgressAddon:
resp_ctx = {**resp_ctx, "context": result.context}
if result.severity == "block":
self._block(flow, f"egress DLP: {result.reason}", ctx=resp_ctx)
elif result.severity == "warn" and self.config.log >= LOG_BLOCKS:
elif result.severity == "warn" and config.log >= LOG_BLOCKS:
sys.stderr.write(
json.dumps({
"event": "egress_warn",
@@ -621,7 +675,10 @@ class EgressAddon:
)
def websocket_message(self, flow: http.HTTPFlow) -> None:
"""DLP scan on WebSocket frames.
"""DLP scan on WebSocket frames, against the calling bottle's resolved
config (see `_flow_ctx`). `request()` resolves and stashes the per-flow
(config, slug, env) at the upgrade, so both the multi-tenant and
single-tenant gateways scan here.
Outbound frames (from_client) are scanned for credential leakage;
inbound frames are scanned for prompt injection. On a block the
@@ -630,10 +687,8 @@ class EgressAddon:
"""
if flow.websocket is None: # type: ignore[union-attr]
return
# WebSocket DLP runs against the static config only (single-tenant); in
# the consolidated gateway self.config has no routes, so this is inert
# until websocket routing is made source-IP-aware (a separate slice).
route = match_route(self.config.routes, flow.request.pretty_host)
config, slug, env = self._flow_ctx(flow)
route = match_route(config.routes, flow.request.pretty_host)
if route is None:
return
message = flow.websocket.messages[-1] # type: ignore[union-attr]
@@ -642,8 +697,8 @@ class EgressAddon:
# A WebSocket data frame is not an HTTP request line, so CRLF is
# not an injection vector here — scan only for credential leakage.
result = scan_outbound(
route, content, os.environ,
safe_tokens=self._safe_tokens_for(self._supervise_slug), crlf_text="",
route, content, env,
safe_tokens=self._safe_tokens_for(slug), crlf_text="",
)
if result is not None and result.severity == "block":
sys.stderr.write(f"egress DLP: {result.reason}\n")
+37 -2
View File
@@ -17,9 +17,22 @@ import urllib.error
import urllib.request
from dataclasses import dataclass
from ..paths import host_control_plane_token
from .control_plane import CONTROL_AUTH_HEADER
DEFAULT_TIMEOUT_SECONDS = 5.0
def _host_auth_token() -> str:
"""The per-host control-plane secret, or "" if it can't be read. "" means
'send no auth header' correct against an open (unconfigured) control
plane, and harmlessly rejected by a secured one."""
try:
return host_control_plane_token()
except OSError:
return ""
class OrchestratorClientError(RuntimeError):
"""A control-plane call failed (unreachable, or an unexpected status)."""
@@ -34,11 +47,24 @@ class RegisteredBottle:
class OrchestratorClient:
"""Trusted host-side client for the orchestrator control plane."""
"""Trusted host-side client for the orchestrator control plane.
def __init__(self, base_url: str, *, timeout: float = DEFAULT_TIMEOUT_SECONDS) -> None:
Presents the per-host control-plane secret on every call (the header the
control plane requires on all routes but `/health`). The secret is read
from the host file this client only ever runs host-side (CLI, launcher,
discovery), so it can read what an agent can't. `auth_token` is overridable
for tests; the default reads the host file, minting it on first use."""
def __init__(
self,
base_url: str,
*,
timeout: float = DEFAULT_TIMEOUT_SECONDS,
auth_token: str | None = None,
) -> None:
self._base = base_url.rstrip("/")
self._timeout = timeout
self._auth_token = auth_token if auth_token is not None else _host_auth_token()
def _request(
self, method: str, path: str, body: dict[str, object] | None = None,
@@ -49,6 +75,8 @@ class OrchestratorClient:
callers can treat 404 as a meaningful "no such bottle"."""
data = json.dumps(body).encode() if body is not None else None
headers = {"Content-Type": "application/json"} if data is not None else {}
if self._auth_token:
headers[CONTROL_AUTH_HEADER] = self._auth_token
req = urllib.request.Request(
f"{self._base}{path}", data=data, method=method, headers=headers,
)
@@ -186,6 +214,13 @@ def discover_orchestrator_url(*, timeout: float = 2.0) -> str:
f"http://{netpool.orch_slot().guest_ip}:{CONTROL_PLANE_PORT}")
except Exception: # noqa: BLE001 — backend optional / not firecracker
pass
try: # macOS: infra container control plane on its host-only address
from ..backend.macos_container.infra import probe_control_plane_url
url = probe_control_plane_url()
if url:
candidates.append(url)
except Exception: # noqa: BLE001 — backend optional / not macOS
pass
for url in candidates:
if OrchestratorClient(url, timeout=timeout).health():
return url
+59 -5
View File
@@ -34,6 +34,7 @@ returned only once, to the caller that launches the bottle.
from __future__ import annotations
import hmac
import http.server
import json
import os
@@ -42,11 +43,20 @@ import sys
import typing
from urllib.parse import urlsplit
from ..paths import CONTROL_PLANE_TOKEN_ENV
from .service import Orchestrator
# JSON body payload type (parsed request / rendered response).
Json = dict[str, object]
# The request header carrying the per-host control-plane secret. Every route
# except `GET /health` requires it (see `dispatch`). The trusted callers hold
# the secret (the gateway's PolicyResolver, the host CLI's OrchestratorClient);
# an agent that can merely *reach* the port cannot present it, so it can't
# enumerate bottles, rewrite policy, read injected upstream tokens, or approve
# its own supervise proposals.
CONTROL_AUTH_HEADER = "x-bot-bottle-control-auth"
def _parse_json_object(body: bytes) -> Json:
"""Parse a JSON object body. Raises ValueError for non-objects / bad JSON."""
@@ -59,15 +69,29 @@ def _parse_json_object(body: bytes) -> Json:
def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
orch: Orchestrator, method: str, path: str, body: bytes
orch: Orchestrator, method: str, path: str, body: bytes, *, authorized: bool = True,
) -> tuple[int, Json]:
"""Route one control-plane request to a (status, payload) pair. Pure —
no I/O beyond the orchestrator so it is fully testable without a socket."""
no I/O beyond the orchestrator so it is fully testable without a socket.
`authorized` is whether the request presented the control-plane secret (or
no secret is configured see `ControlPlaneServer`). Every route except
`GET /health` requires it: the source-IP + identity-token checks inside
`/resolve` and `/attribute` authenticate the *bottle* a request is about,
not the *caller*, so without this gate any agent that can reach the port
could rewrite another bottle's policy, read the injected upstream tokens,
or approve its own supervise proposals. Defaults True so unit tests of the
routing logic don't have to thread it through."""
route = urlsplit(path).path.rstrip("/") or "/"
if method == "GET" and route == "/health":
return 200, {"status": "ok"}
if not authorized:
# Everything below is a trusted-caller operation. Deny before touching
# the registry / broker / supervise store.
return 401, {"error": "control-plane authentication required"}
if method == "GET" and route == "/gateway":
return 200, orch.gateway_status()
@@ -209,8 +233,10 @@ class Handler(http.server.BaseHTTPRequestHandler):
assert isinstance(server, ControlPlaneServer)
length = int(self.headers.get("Content-Length") or 0)
body = self.rfile.read(length) if length > 0 else b""
authorized = server.is_authorized(self.headers.get(CONTROL_AUTH_HEADER, ""))
try:
status, payload = dispatch(server.orchestrator, method, self.path, body)
status, payload = dispatch(
server.orchestrator, method, self.path, body, authorized=authorized)
except Exception as e: # noqa: BLE001 — the control plane must stay up
sys.stderr.write(f"orchestrator: {method} {self.path} failed: {e!r}\n")
sys.stderr.flush()
@@ -236,15 +262,40 @@ class Handler(http.server.BaseHTTPRequestHandler):
class ControlPlaneServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
"""Threading HTTP server that carries the orchestrator for its handlers."""
"""Threading HTTP server that carries the orchestrator for its handlers.
Holds the per-host control-plane secret (from `$BOT_BOTTLE_CONTROL_PLANE_TOKEN`,
injected by the launcher into this container only). When a secret is set,
every route but `/health` requires it; when it is unset the server runs
**open** and says so loudly at startup a fail-visible fallback for tests
and any backend that hasn't wired the secret yet (e.g. Firecracker, whose
nft boundary already blocks agents from the control-plane port)."""
daemon_threads = True
allow_reuse_address = True
def __init__(self, address: tuple[str, int], orchestrator: Orchestrator) -> None:
self.orchestrator = orchestrator
self._auth_token = os.environ.get(CONTROL_PLANE_TOKEN_ENV, "").strip()
if not self._auth_token:
sys.stderr.write(
"orchestrator: WARNING — no control-plane secret "
f"(${CONTROL_PLANE_TOKEN_ENV}); running WITHOUT caller "
"authentication. Any client that can reach this port can drive "
"it. Backends that put the control plane on an agent-reachable "
"network MUST set this.\n"
)
sys.stderr.flush()
super().__init__(address, Handler)
def is_authorized(self, presented: str) -> bool:
"""True iff the request may proceed past `/health`: either no secret is
configured (open mode) or the presented header matches it. Constant-time
compare so a wrong token leaks nothing timing-wise."""
if not self._auth_token:
return True
return hmac.compare_digest(presented, self._auth_token)
def make_server(
orchestrator: Orchestrator, host: str = "127.0.0.1", port: int = 0
@@ -254,4 +305,7 @@ def make_server(
return ControlPlaneServer((host, port), orchestrator)
__all__ = ["dispatch", "Handler", "ControlPlaneServer", "make_server", "Json"]
__all__ = [
"dispatch", "Handler", "ControlPlaneServer", "make_server", "Json",
"CONTROL_AUTH_HEADER",
]
+13 -2
View File
@@ -23,7 +23,11 @@ import time
from pathlib import Path
from ..docker_cmd import run_docker
from ..paths import host_db_path
from ..paths import (
CONTROL_PLANE_TOKEN_ENV,
host_control_plane_token,
host_db_path,
)
from ..supervise import DB_PATH_IN_CONTAINER
# The host DB dir is bind-mounted here so the gateway's supervise daemon
@@ -215,12 +219,19 @@ class DockerGateway(Gateway):
]
for port in self._host_port_bindings:
argv += ["--publish", f"0.0.0.0:{port}:{port}"]
run_env = dict(os.environ)
if self._orchestrator_url:
# Makes the gateway's egress / git / supervise daemons multi-tenant:
# each request resolves source-IP -> policy against the control plane.
argv += ["--env", f"BOT_BOTTLE_ORCHESTRATOR_URL={self._orchestrator_url}"]
# ...and presents the control-plane secret on those /resolve calls
# (the control plane requires it). Bare `--env NAME` keeps the value
# off argv / `docker inspect`; only the gateway (not the agent) is
# given it. Only needed in multi-tenant mode, where /resolve is used.
argv += ["--env", CONTROL_PLANE_TOKEN_ENV]
run_env[CONTROL_PLANE_TOKEN_ENV] = host_control_plane_token()
argv.append(self.image_ref)
proc = run_docker(argv)
proc = run_docker(argv, env=run_env)
if proc.returncode != 0:
raise GatewayError(f"gateway failed to start: {proc.stderr.strip()}")
+23 -7
View File
@@ -25,8 +25,8 @@ from pathlib import Path
from .. import log
from ..docker_cmd import run_docker
from ..paths import bot_bottle_root
from .gateway import GATEWAY_IMAGE, GATEWAY_NETWORK, DockerGateway, GatewayError
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
DEFAULT_PORT = 8099
ORCHESTRATOR_NAME = "bot-bottle-orchestrator"
@@ -83,8 +83,11 @@ class OrchestratorService:
`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`). Subclass and override `_gateway()`
to supply a backend-specific gateway variant."""
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)."""
def __init__(
self,
@@ -93,6 +96,7 @@ class OrchestratorService:
network: str = GATEWAY_NETWORK,
image: str = ORCHESTRATOR_IMAGE,
gateway_image: str = GATEWAY_IMAGE,
gateway_name: str = GATEWAY_NAME,
repo_root: Path = _REPO_ROOT,
host_root: Path | None = None,
orchestrator_name: str = ORCHESTRATOR_NAME,
@@ -106,6 +110,7 @@ class OrchestratorService:
# 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
@@ -147,7 +152,11 @@ class OrchestratorService:
"--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.
# 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,
@@ -155,11 +164,15 @@ class OrchestratorService:
# 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()}"
@@ -167,7 +180,10 @@ class OrchestratorService:
def _gateway(self) -> DockerGateway:
return DockerGateway(
self._gateway_image, network=self.network, orchestrator_url=self.internal_url
self._gateway_image,
name=self._gateway_name,
network=self.network,
orchestrator_url=self.internal_url,
)
def _ensure_orchestrator_image(self) -> None:
+59 -1
View File
@@ -16,6 +16,8 @@ layer (and to COPY flat into the gateway).
from __future__ import annotations
import os
import secrets
import stat
from pathlib import Path
# The single shared host state DB. All bot-bottle SQLite stores (supervise
@@ -23,6 +25,14 @@ from pathlib import Path
# TableMigrations schema_key namespaces each store's tables.
HOST_DB_FILENAME = "bot-bottle.db"
# The per-host control-plane secret file, and the env var the launchers inject
# its value into. The control plane requires this secret on every mutating /
# reading route (see orchestrator/control_plane.py); it is held only by the
# trusted callers (control plane, gateway, host CLI) and never handed to an
# agent, so an agent that can reach the control-plane port still can't drive it.
CONTROL_PLANE_TOKEN_FILENAME = "control-plane-token"
CONTROL_PLANE_TOKEN_ENV = "BOT_BOTTLE_CONTROL_PLANE_TOKEN"
def bot_bottle_root() -> Path:
"""The app data root — `$BOT_BOTTLE_ROOT` if set, else `~/.bot-bottle`."""
@@ -40,4 +50,52 @@ def host_db_path() -> Path:
return bot_bottle_root() / "db" / HOST_DB_FILENAME
__all__ = ["HOST_DB_FILENAME", "bot_bottle_root", "host_db_path"]
def host_db_dir() -> Path:
"""The directory holding the shared host state DB, created if missing.
Backends bind-mount this into their gateway so the supervise daemon writes
to the one DB the orchestrator (and the operator over HTTP) reads."""
db_dir = host_db_path().parent
db_dir.mkdir(parents=True, exist_ok=True)
return db_dir
def host_control_plane_token() -> str:
"""The per-host control-plane secret, minted (256-bit, url-safe) and
persisted 0600 on first use, then reused.
This is the shared secret the launchers inject into the control-plane and
gateway containers and that the host CLI presents on every call. It is a
*host* artifact the file lives under the root the agent never mounts, and
the env var is set only on the trusted containers so reading it here is
safe on the host launch path but the value never reaches a bottle."""
path = bot_bottle_root() / CONTROL_PLANE_TOKEN_FILENAME
try:
existing = path.read_text().strip()
if existing:
return existing
except OSError:
pass
path.parent.mkdir(parents=True, exist_ok=True)
token = secrets.token_urlsafe(32)
# Create 0600 up front (O_EXCL loses a concurrent race harmlessly — we
# re-read the winner's token below) so the secret is never briefly world-
# readable between write and chmod.
try:
fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
except FileExistsError:
return path.read_text().strip()
with os.fdopen(fd, "w") as f:
f.write(token)
os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
return token
__all__ = [
"HOST_DB_FILENAME",
"CONTROL_PLANE_TOKEN_FILENAME",
"CONTROL_PLANE_TOKEN_ENV",
"bot_bottle_root",
"host_db_path",
"host_db_dir",
"host_control_plane_token",
]
+19 -1
View File
@@ -29,11 +29,29 @@ the gateway.
from __future__ import annotations
import json
import os
import urllib.error
import urllib.request
DEFAULT_TIMEOUT_SECONDS = 2.0
# The control-plane secret this gateway presents on every /resolve call, read
# from the env the launcher injects into the gateway container. The control
# plane requires it (orchestrator/control_plane.py). Constant + env-var name are
# duplicated here rather than imported because this module is COPYed flat into
# the gateway image, free of bot-bottle imports — same rationale as
# IDENTITY_HEADER in egress_addon / git_http_backend.
CONTROL_AUTH_HEADER = "x-bot-bottle-control-auth"
CONTROL_PLANE_TOKEN_ENV = "BOT_BOTTLE_CONTROL_PLANE_TOKEN"
def _control_auth_headers() -> dict[str, str]:
"""The auth header to send, or {} when no secret is configured (an open
control plane, e.g. Firecracker behind its nft boundary sending nothing
is correct there and harmlessly ignored)."""
token = os.environ.get(CONTROL_PLANE_TOKEN_ENV, "").strip()
return {CONTROL_AUTH_HEADER: token} if token else {}
class PolicyResolveError(RuntimeError):
"""The orchestrator was unreachable or returned an unexpected status —
@@ -57,7 +75,7 @@ class PolicyResolver:
).encode()
req = urllib.request.Request(
f"{self._base}/resolve", data=body, method="POST",
headers={"Content-Type": "application/json"},
headers={"Content-Type": "application/json", **_control_auth_headers()},
)
try:
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
@@ -0,0 +1,155 @@
"""Integration: the Docker orchestrator's control plane enforces the
per-host auth secret against a real container (issue #400).
Unit tests exercise `dispatch()` in-process, socket-free. This starts the
actual orchestrator + gateway as Docker containers and drives the real HTTP
server over its published loopback port, the same path an agent sharing the
gateway network or the trusted host CLI would use.
Gated on a reachable Docker daemon. Uses unique container/network names,
test-only image tags (never the production `:latest` ones, so a rebuild
here can't make `_running_image_is_current()` see a real host's running
gateway as stale and force-recreate it), and a throwaway `BOT_BOTTLE_ROOT`
so a run never touches or collides with a real per-host orchestrator or
gateway. The whole stack is brought up once for the class (`setUpClass`),
not per test method every test here is a read-only check against the
same running control plane.
"""
from __future__ import annotations
import os
import secrets
import subprocess
import tempfile
import unittest
from pathlib import Path
from bot_bottle.orchestrator.client import OrchestratorClient
from bot_bottle.orchestrator.lifecycle import OrchestratorService
from bot_bottle.paths import host_control_plane_token
from tests._docker import skip_unless_docker
# Fixed (not per-run-suffixed) so repeated runs reuse the same layer-cached
# 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"
@skip_unless_docker()
@unittest.skipIf(
os.environ.get("GITEA_ACTIONS") == "true",
"skipped under act_runner: the orchestrator container bind-mounts the repo "
"path into a container on the socket-shared host daemon, which can't see the "
"runner's /workspace — same host-bind-mount constraint as the other "
"bottle-bringup integration tests",
)
class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
@classmethod
def setUpClass(cls) -> None:
suffix = secrets.token_hex(4)
cls._tmp = tempfile.TemporaryDirectory() # pylint: disable=consider-using-with
cls.addClassCleanup(cls._tmp.cleanup)
# host_control_plane_token() — both the token read below and the one
# OrchestratorService injects into the container's env — resolves its
# path via the *ambient* BOT_BOTTLE_ROOT env var, not the host_root
# kwarg passed to the constructor (that kwarg only controls the DB
# bind-mount destination). Without pointing the env var at the same
# throwaway dir, this "isolated" test would read/write the developer's
# real ~/.bot-bottle/control-plane-token.
previous_root = os.environ.get("BOT_BOTTLE_ROOT")
def _restore_root() -> None:
if previous_root is None:
os.environ.pop("BOT_BOTTLE_ROOT", None)
else:
os.environ["BOT_BOTTLE_ROOT"] = previous_root
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}"
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.svc = OrchestratorService(
orchestrator_name=orchestrator_name,
gateway_name=gateway_name,
network=network,
image=_TEST_ORCHESTRATOR_IMAGE,
gateway_image=_TEST_GATEWAY_IMAGE,
port=20000 + secrets.randbelow(10000),
host_root=host_root,
)
cls.svc.ensure_running()
cls.token = host_control_plane_token()
@staticmethod
def _teardown_docker(
orchestrator_name: str, gateway_name: str, network: str, host_root: Path
) -> None:
subprocess.run(
["docker", "rm", "--force", orchestrator_name, gateway_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
# 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",
f"{os.getuid()}:{os.getgid()}", "/r"],
stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
)
def _request(
self, method: str, path: str, *, token: str = ""
) -> tuple[int, dict[str, object]]:
# Reuses the real host-side client's request/response handling rather
# than hand-rolling urllib here; _request (not one of the named
# wrapper methods) is what exposes raw status codes for arbitrary
# paths/tokens, which is exactly what these auth-boundary tests need.
client = OrchestratorClient(self.svc.url, auth_token=token)
return client._request(method, path) # pylint: disable=protected-access
def test_health_is_open_without_a_token(self) -> None:
status, payload = self._request("GET", "/health")
self.assertEqual(200, status)
self.assertEqual("ok", payload["status"])
def test_bottles_rejects_a_caller_with_no_token(self) -> None:
"""The enumeration attack from issue #400: an agent sharing the
gateway network could list every bottle + its policy with no
credential at all."""
status, _ = self._request("GET", "/bottles")
self.assertEqual(401, status)
def test_bottles_rejects_a_wrong_token(self) -> None:
status, _ = self._request("GET", "/bottles", token="not-the-real-secret")
self.assertEqual(401, status)
def test_bottles_accepts_the_real_token(self) -> None:
status, payload = self._request("GET", "/bottles", token=self.token)
self.assertEqual(200, status)
self.assertEqual([], payload["bottles"])
def test_resolve_rejects_a_caller_with_no_token(self) -> None:
"""The credential-lift attack from issue #400: an agent could POST
/resolve directly and read back the upstream tokens it's never meant
to see."""
status, _ = self._request("POST", "/resolve")
self.assertEqual(401, status)
if __name__ == "__main__":
unittest.main()
+10 -5
View File
@@ -247,7 +247,7 @@ class TestHasBackend(unittest.TestCase):
class TestEnsureOrchestrator(unittest.TestCase):
"""The backend-agnostic orchestrator bring-up entry point. Docker starts
the orchestrator + gateway containers; firecracker boots the infra VM;
backends without one (macos-container) die with a pointer."""
macos-container starts the infra container."""
def test_docker_delegates_to_orchestrator_service(self):
b = get_bottle_backend("docker")
@@ -272,11 +272,16 @@ class TestEnsureOrchestrator(unittest.TestCase):
url = b.ensure_orchestrator()
self.assertEqual(url, "http://10.243.255.1:8099")
def test_macos_default_dies(self):
from bot_bottle.log import Die
def test_macos_delegates_to_infra_container(self):
b = get_bottle_backend("macos-container")
with self.assertRaises(Die):
b.ensure_orchestrator()
with patch(
"bot_bottle.backend.macos_container.infra.MacosInfraService"
) as service_cls:
service_cls.return_value.ensure_running.return_value.control_plane_url = (
"http://192.168.128.2:8099"
)
url = b.ensure_orchestrator()
self.assertEqual(url, "http://192.168.128.2:8099")
if __name__ == "__main__":
@@ -8,6 +8,7 @@ real mitmproxy package."""
from __future__ import annotations
import json
import os
import sys
import types
import unittest
@@ -107,14 +108,14 @@ class _Flow:
def _log_request(addon: EgressAddon, flow: _Flow) -> dict[str, Any]:
buf = StringIO()
with patch("sys.stderr", buf):
addon._log_request(flow) # type: ignore[arg-type]
addon._log_request(flow, os.environ) # type: ignore[arg-type]
return json.loads(buf.getvalue())
def _log_response(addon: EgressAddon, flow: _Flow) -> dict[str, Any]:
buf = StringIO()
with patch("sys.stderr", buf):
addon._log_response(flow) # type: ignore[arg-type]
addon._log_response(flow, os.environ) # type: ignore[arg-type]
return json.loads(buf.getvalue())
@@ -141,6 +141,10 @@ class _Flow:
self.response = response
self.websocket: Any = None
self.killed = False
# mitmproxy flows carry a per-flow `metadata` dict for addon use; the
# egress addon stashes the resolved (config, slug, env) there in
# request() so the response/websocket hooks reuse it.
self.metadata: dict[str, Any] = {}
def kill(self) -> None:
self.killed = True
@@ -859,5 +863,91 @@ class TestSuperviseMultiTenant(unittest.TestCase):
self.assertNotIn(_OPENAI_KEY, addon._safe_tokens_for(""))
class TestMultiTenantInboundDlp(unittest.TestCase):
"""Consolidated gateway: the response + websocket DLP hooks must scan
against the *calling bottle's* config, resolved by source IP in request()
and reused here. The static `self.config` is empty in this mode, so before
the flow-context stash these hooks silently skipped every scan (fail-open).
"""
def _consolidated_addon(self) -> EgressAddon:
addon = _addon(Config(routes=())) # empty static config, as in prod
addon._resolver = cast(Any, _CtxResolver({"10.0.0.1": "bottle-a"}))
return addon
def test_response_injection_blocked_after_request_resolves(self) -> None:
addon = self._consolidated_addon()
flow = _with_client_ip(_Flow(_Request(host="api.example.com")), "10.0.0.1")
_run_request(addon, flow) # resolves + stashes bottle-a's allowlist
self.assertIsNone(flow.response) # request forwarded
flow.response = _Response(200, content=_INJECTION_BLOCK)
addon.response(flow) # type: ignore[arg-type]
assert flow.response is not None
# Empty static config would have found no route and left this 200.
self.assertEqual(403, flow.response.status_code)
def test_websocket_outbound_token_killed_after_request_resolves(self) -> None:
addon = self._consolidated_addon()
flow = _with_client_ip(_Flow(_Request(host="api.example.com")), "10.0.0.1")
_run_request(addon, flow) # the ws upgrade resolves + stashes the config
flow.websocket = _WebSocketData(
[_Message(f"k={_OPENAI_KEY}".encode(), from_client=True)]
)
addon.websocket_message(flow) # type: ignore[arg-type]
self.assertTrue(flow.killed) # scanned against bottle-a's route now
def test_websocket_inbound_injection_killed_after_request_resolves(self) -> None:
addon = self._consolidated_addon()
flow = _with_client_ip(_Flow(_Request(host="api.example.com")), "10.0.0.1")
_run_request(addon, flow)
flow.websocket = _WebSocketData(
[_Message(_INJECTION_BLOCK.encode(), from_client=False)]
)
addon.websocket_message(flow) # type: ignore[arg-type]
self.assertTrue(flow.killed)
def test_response_log_redacts_per_bottle_resolve_token(self) -> None:
# LOG_FULL response logging now runs in multi-tenant mode, so it must
# scrub the calling bottle's /resolve token — which lives only in the
# resolved env overlay, never in the gateway's os.environ. A
# non-token-shaped secret is caught only via that env, so os.environ
# redaction (the pre-fix behaviour) would leak it into the log.
secret = "bottle-a-provisioned-secret-value"
policy = "log: 2\nroutes:\n - host: api.example.com\n"
class _TokenResolver:
def resolve_policy_and_bottle_id(
self, ip: str, identity_token: str = "",
) -> tuple[str | None, str | None, dict[str, str]]:
del ip, identity_token
return policy, "bottle-a", {"EGRESS_TOKEN_0": secret}
addon = _addon(Config(routes=()))
addon._resolver = cast(Any, _TokenResolver())
flow = _with_client_ip(_Flow(_Request(host="api.example.com")), "10.0.0.1")
_run_request(addon, flow)
flow.response = _Response(200, content=f"echo {secret} back")
buf = StringIO()
with patch("sys.stderr", buf):
addon.response(flow) # type: ignore[arg-type]
logged = buf.getvalue()
self.assertIn("egress_response", logged) # LOG_FULL logged the response
self.assertNotIn(secret, logged) # redacted via the resolved env overlay
def test_unattributed_flow_has_no_route_so_frame_passes(self) -> None:
# An unattributed source resolves to a deny-all (empty) config, so the
# request is blocked at the upgrade and any later frame has no route to
# scan against — it passes rather than being attributed to a bottle.
addon = self._consolidated_addon()
flow = _with_client_ip(_Flow(_Request(host="api.example.com")), "10.9.9.9")
_run_request(addon, flow)
self.assertIsNotNone(flow.response) # blocked at the upgrade
flow.websocket = _WebSocketData(
[_Message(f"k={_OPENAI_KEY}".encode(), from_client=True)]
)
addon.websocket_message(flow) # type: ignore[arg-type]
self.assertFalse(flow.killed)
if __name__ == "__main__":
unittest.main()
+16 -11
View File
@@ -50,30 +50,35 @@ def _client() -> Mock:
class TestEnsureGateway(unittest.TestCase):
def _run(self, service: MagicMock) -> GatewayEndpoint:
with patch(f"{_MOD}.MacosOrchestratorService", return_value=service):
with patch(f"{_MOD}.MacosInfraService", return_value=service):
return ensure_gateway()
def _service(self) -> MagicMock:
from bot_bottle.backend.macos_container.infra import InfraEndpoint
service = MagicMock()
service.ensure_running.return_value = "http://192.168.128.2:8099"
service.ensure_running.return_value = InfraEndpoint(
control_plane_url="http://192.168.128.2:8099",
gateway_ip="192.168.128.2",
)
service.network = "bot-bottle-mac-gateway"
service.gateway.return_value.ip_on_shared_network.return_value = "192.168.128.3"
service.gateway.return_value.ca_cert_pem.return_value = "PEM"
service.ca_cert_pem.return_value = "PEM"
return service
def test_reports_gateway_endpoint(self) -> None:
endpoint = self._run(self._service())
self.assertEqual("http://192.168.128.2:8099", endpoint.orchestrator_url)
self.assertEqual("192.168.128.3", endpoint.gateway_ip)
self.assertEqual("192.168.128.2", endpoint.gateway_ip)
self.assertEqual("PEM", endpoint.gateway_ca_pem)
self.assertEqual("bot-bottle-mac-gateway", endpoint.network)
def test_gateway_is_pointed_at_the_resolved_control_plane(self) -> None:
"""Apple has no container DNS, so the gateway must be handed the
control plane's *resolved URL* rather than a container name."""
service = self._service()
self._run(service)
service.gateway.assert_called_with("http://192.168.128.2:8099")
def test_control_plane_and_gateway_share_one_address(self) -> None:
"""One infra container hosts both, so the gateway IP and the
control-plane host are the same."""
endpoint = self._run(self._service())
self.assertEqual(
endpoint.gateway_ip,
endpoint.orchestrator_url.split("://")[1].split(":")[0],
)
class TestRegisterAgent(unittest.TestCase):
+4 -7
View File
@@ -60,13 +60,10 @@ class TestMacosContainerEnumerate(unittest.TestCase):
self.assertEqual(["dev-abc"], [a.slug for a in agents])
self.assertEqual(["macos-container"], [a.backend_name for a in agents])
def test_excludes_the_shared_singletons(self):
"""The gateway and control plane share the bot-bottle- prefix but are
infrastructure listing them would invent an agent per host."""
agents = self._enumerate(
"bot-bottle-mac-gateway\nbot-bottle-mac-orchestrator\n"
"bot-bottle-dev-abc\n"
)
def test_excludes_the_infra_singleton(self):
"""The infra container shares the bot-bottle- prefix but is
infrastructure listing it would invent an agent per host."""
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):
@@ -158,6 +158,36 @@ class TestIdentityTokenDelivery(unittest.TestCase):
argv = bottle.agent_argv(["--help"], tty=False)
self.assertNotIn("--env", argv)
class TestPlanIdentityToken(unittest.TestCase):
"""git-gate's gitconfig extraHeader and the supervise MCP --header read
`getattr(plan, "identity_token", "")` at provision time and both bypass the
egress proxy (NO_PROXY), so the exec-time proxy token never reaches them
the plan must carry the token or /resolve fail-closes and git + supervise
are denied on macOS."""
def test_macos_plan_has_the_identity_token_field(self) -> None:
from dataclasses import fields
from bot_bottle.backend.macos_container.bottle_plan import (
MacosContainerBottlePlan,
)
names = {f.name for f in fields(MacosContainerBottlePlan)}
self.assertIn("identity_token", names)
def test_matches_the_docker_plan(self) -> None:
"""Both consolidated backends must expose identity_token so a shared
provision-time consumer degrades to neither backend silently."""
from dataclasses import fields
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.backend.macos_container.bottle_plan import (
MacosContainerBottlePlan,
)
docker = {f.name for f in fields(DockerBottlePlan)}
macos = {f.name for f in fields(MacosContainerBottlePlan)}
self.assertIn("identity_token", docker & macos)
def test_provisioning_exec_also_carries_the_token(self) -> None:
"""`provision` runs through `exec`; a provider whose provision step
fetches anything would otherwise egress token-less and be denied."""
+50
View File
@@ -273,5 +273,55 @@ resolver #2
)
def _completed(stdout: str, returncode: int = 0):
return util.subprocess.CompletedProcess(args=[], returncode=returncode, stdout=stdout, stderr="")
class TestInspectDigests(unittest.TestCase):
"""image_digest and container_image_digest must read the SAME field shape
(a `descriptor.digest`) so a running container and its image are
comparable. An asymmetry recreates the gateway on every launch."""
_IMAGE = '{"configuration": {"descriptor": {"digest": "sha256:abc123"}}, "id": "abc123"}'
_CONTAINER = '[{"configuration": {"image": {"descriptor": {"digest": "sha256:abc123"}}}}]'
def test_image_and_container_digests_agree_for_the_same_image(self):
with patch.object(util, "run_container_argv") as run:
run.return_value = _completed(self._IMAGE)
img = util.image_digest("bot-bottle-gateway:latest")
run.return_value = _completed(self._CONTAINER)
ctr = util.container_image_digest("bot-bottle-mac-gateway")
self.assertEqual("abc123", img)
self.assertEqual(img, ctr)
def test_image_digest_never_falls_back_to_a_tag_or_id(self):
"""The old `id` fallback could yield a value container_image_digest
can't produce, permanently mismatching. Missing descriptor → '' (don't
churn), never a stray id/tag."""
with patch.object(util, "run_container_argv") as run:
run.return_value = _completed('{"id": "bot-bottle-gateway:latest"}')
self.assertEqual("", util.image_digest("bot-bottle-gateway:latest"))
def test_unreadable_inspect_returns_empty(self):
with patch.object(util, "run_container_argv") as run:
run.return_value = _completed("", returncode=1)
self.assertEqual("", util.image_digest("x"))
self.assertEqual("", util.container_image_digest("x"))
self.assertEqual({}, util.container_env("x"))
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"]), \
patch.object(util.time, "sleep"):
ip = util.wait_container_ipv4_on_network("c", "net", timeout=5, poll=0)
self.assertEqual("192.168.128.4", ip)
def test_returns_empty_on_timeout(self):
with patch.object(util, "try_container_ipv4_on_network", return_value=""), \
patch.object(util.time, "sleep"):
self.assertEqual("", util.wait_container_ipv4_on_network("c", "net", timeout=-1))
if __name__ == "__main__":
unittest.main()
-275
View File
@@ -1,275 +0,0 @@
"""Unit: the Apple gateway + orchestrator lifecycle (PRD 0070)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from bot_bottle.backend.macos_container.gateway import AppleGateway, GatewayError
from bot_bottle.backend.macos_container.orchestrator_service import (
MacosOrchestratorService,
OrchestratorStartError,
)
_GW = "bot_bottle.backend.macos_container.gateway"
_ORCH = "bot_bottle.backend.macos_container.orchestrator_service"
def _ok(stdout: str = "") -> Mock:
return Mock(returncode=0, stdout=stdout, stderr="")
def _fail(stderr: str = "boom") -> Mock:
return Mock(returncode=1, stdout="", stderr=stderr)
class TestAppleGatewayRun(unittest.TestCase):
def _argv(self, run: Mock) -> list[str]:
return run.call_args.args[0]
def _start(self, run: Mock) -> None:
with patch(f"{_GW}.container_mod") as mod:
mod.container_is_running.return_value = False
mod.dns_server.return_value = "1.1.1.1"
mod.run_container_argv = run
AppleGateway(orchestrator_url="http://192.168.128.2:8099").ensure_running()
def test_nat_network_precedes_the_host_only_network(self) -> None:
"""Apple Container makes the FIRST --network the default route, so the
NAT network must lead or the gateway has no route out."""
run = Mock(return_value=_ok())
self._start(run)
argv = self._argv(run)
networks = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
self.assertEqual(["bot-bottle-mac-egress", "bot-bottle-mac-gateway"], networks)
def test_control_plane_url_is_passed_for_multi_tenancy(self) -> None:
run = Mock(return_value=_ok())
self._start(run)
self.assertIn(
"BOT_BOTTLE_ORCHESTRATOR_URL=http://192.168.128.2:8099", self._argv(run),
)
def test_dns_is_explicit(self) -> None:
"""The NAT gateway routes but does not resolve."""
run = Mock(return_value=_ok())
self._start(run)
argv = self._argv(run)
self.assertEqual("1.1.1.1", argv[argv.index("--dns") + 1])
def test_start_failure_raises(self) -> None:
with self.assertRaises(GatewayError):
self._start(Mock(return_value=_fail()))
def test_running_current_gateway_is_left_alone(self) -> None:
"""Idempotent singleton: N launches must not restart the gateway and
drop every other bottle's data plane."""
run = Mock(return_value=_ok())
with patch(f"{_GW}.container_mod") as mod:
mod.container_is_running.return_value = True
mod.container_image_digest.return_value = "abc"
mod.image_digest.return_value = "abc"
mod.run_container_argv = run
AppleGateway().ensure_running()
run.assert_not_called()
def test_stale_image_forces_a_recreate(self) -> None:
"""A rebuilt image only takes effect if the running container is
replaced otherwise it keeps serving the OLD daemons."""
run = Mock(return_value=_ok())
with patch(f"{_GW}.container_mod") as mod:
mod.container_is_running.return_value = True
mod.container_image_digest.return_value = "old"
mod.image_digest.return_value = "new"
mod.dns_server.return_value = "1.1.1.1"
mod.run_container_argv = run
AppleGateway().ensure_running()
run.assert_called_once()
def test_unreadable_digest_does_not_churn(self) -> None:
run = Mock(return_value=_ok())
with patch(f"{_GW}.container_mod") as mod:
mod.container_is_running.return_value = True
mod.container_image_digest.return_value = ""
mod.image_digest.return_value = ""
mod.run_container_argv = run
AppleGateway().ensure_running()
run.assert_not_called()
def _start_with_running_env(self, env: dict[str, str], url: str) -> Mock:
run = Mock(return_value=_ok())
with patch(f"{_GW}.container_mod") as mod:
mod.container_is_running.return_value = True
mod.container_image_digest.return_value = "abc"
mod.image_digest.return_value = "abc"
mod.container_env.return_value = env
mod.dns_server.return_value = "1.1.1.1"
mod.run_container_argv = run
AppleGateway(orchestrator_url=url).ensure_running()
return run
def test_moved_control_plane_forces_a_recreate(self) -> None:
"""Docker hands the gateway a container *name*, stable across an
orchestrator recreate. Apple has no DNS, so the URL is an IP baked into
the gateway's env — if the orchestrator comes back on a new address and
the gateway isn't recreated, every /resolve fails and every bottle on
the host loses egress."""
run = self._start_with_running_env(
{"BOT_BOTTLE_ORCHESTRATOR_URL": "http://192.168.128.2:8099"},
"http://192.168.128.7:8099",
)
run.assert_called_once()
self.assertIn(
"BOT_BOTTLE_ORCHESTRATOR_URL=http://192.168.128.7:8099",
run.call_args.args[0],
)
def test_unmoved_control_plane_does_not_churn(self) -> None:
run = self._start_with_running_env(
{"BOT_BOTTLE_ORCHESTRATOR_URL": "http://192.168.128.2:8099"},
"http://192.168.128.2:8099",
)
run.assert_not_called()
def test_unreadable_env_does_not_churn(self) -> None:
run = self._start_with_running_env({}, "http://192.168.128.2:8099")
run.assert_not_called()
class TestMacosOrchestratorService(unittest.TestCase):
def test_orchestrator_starts_before_the_gateway(self) -> None:
"""Apple has no container DNS, so the gateway can only be handed the
control plane's IP — which does not exist until it is running. This
ordering is the whole reason the macOS service diverges from docker's."""
order: list[str] = []
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
gateway = Mock()
gateway.ensure_running.side_effect = lambda: order.append("gateway")
def _record_orchestrator(_hash: str) -> None:
order.append("orchestrator")
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.source_hash", return_value="h1"), \
patch.object(svc, "_run_orchestrator_container",
side_effect=_record_orchestrator), \
patch.object(svc, "gateway", return_value=gateway), \
patch.object(svc, "is_healthy", return_value=True):
mod.container_is_running.return_value = False
mod.image_exists.return_value = True
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
url = svc.ensure_running()
self.assertEqual(["orchestrator", "gateway"], order)
self.assertEqual("http://192.168.128.2:8099", url)
def test_gateway_is_handed_the_resolved_url(self) -> None:
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.source_hash", return_value="h1"), \
patch.object(svc, "_run_orchestrator_container"), \
patch.object(svc, "gateway") as gw, \
patch.object(svc, "is_healthy", return_value=True):
mod.container_is_running.return_value = False
mod.image_exists.return_value = True
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
svc.ensure_running()
gw.assert_called_with("http://192.168.128.2:8099")
def test_current_source_leaves_a_healthy_orchestrator_alone(self) -> None:
"""Recreating on every launch would drop every other live bottle's
in-memory egress tokens (#381)."""
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
run = Mock()
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.source_hash", return_value="h1"), \
patch.object(svc, "_run_orchestrator_container", run), \
patch.object(svc, "gateway"), \
patch.object(svc, "is_healthy", return_value=True):
mod.container_is_running.return_value = True
mod.inspect_container.return_value = {
"configuration": {"labels": {"bot-bottle-orchestrator-source-hash": "h1"}}
}
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
svc.ensure_running()
run.assert_not_called()
def test_changed_source_recreates_the_orchestrator(self) -> None:
"""The control-plane process loaded its bind-mounted source at startup
and won't reload it."""
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
run = Mock()
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.source_hash", return_value="h2"), \
patch.object(svc, "_run_orchestrator_container", run), \
patch.object(svc, "gateway"), \
patch.object(svc, "is_healthy", return_value=True):
mod.container_is_running.return_value = True
mod.inspect_container.return_value = {
"configuration": {"labels": {"bot-bottle-orchestrator-source-hash": "h1"}}
}
mod.image_exists.return_value = True
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
svc.ensure_running()
run.assert_called_once()
def test_never_healthy_raises(self) -> None:
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.source_hash", return_value="h1"), \
patch.object(svc, "_run_orchestrator_container"), \
patch.object(svc, "is_healthy", return_value=False):
mod.container_is_running.return_value = False
mod.image_exists.return_value = True
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
with self.assertRaises(OrchestratorStartError):
svc.ensure_running(startup_timeout=0.01)
def test_control_plane_needs_no_route_out(self) -> None:
"""The orchestrator sits only on the host-only network: the host
reaches it there directly, so there is no --publish and no NAT leg."""
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
run = Mock(return_value=_ok())
# `ensure_networks` lives in the gateway module and shells out to the
# `container` CLI, which does not exist on the Linux CI host — patch it
# here, not gateway.container_mod, since it is called through this
# module's imported name.
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.ensure_networks"):
mod.run_container_argv = run
svc._run_orchestrator_container("h1")
argv = run.call_args.args[0]
networks = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
self.assertEqual(["bot-bottle-mac-gateway"], networks)
self.assertNotIn("--publish", argv)
def test_networks_exist_before_the_orchestrator_runs(self) -> None:
"""The orchestrator is the first container on the shared network, so it
has to create it the gateway that used to do so now starts second."""
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
order: list[str] = []
def _networks(*_args: str) -> None:
order.append("networks")
def _run(*_args: list[str]) -> Mock:
order.append("run")
return _ok()
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.ensure_networks", side_effect=_networks):
mod.run_container_argv = Mock(side_effect=_run)
svc._run_orchestrator_container("h1")
self.assertEqual(["networks", "run"], order)
def test_orchestrator_start_failure_raises(self) -> None:
svc = MacosOrchestratorService(repo_root=Path("/r"), host_root=Path("/h"))
with patch(f"{_ORCH}.container_mod") as mod, \
patch(f"{_ORCH}.ensure_networks"):
mod.run_container_argv = Mock(return_value=_fail())
with self.assertRaises(OrchestratorStartError):
svc._run_orchestrator_container("h1")
if __name__ == "__main__":
unittest.main()
+168
View File
@@ -0,0 +1,168 @@
"""Unit: the single macOS infra container (control plane + gateway, PRD 0070)."""
from __future__ import annotations
import unittest
from pathlib import Path
from unittest.mock import Mock, patch
from bot_bottle.backend.macos_container.infra import (
INFRA_DB_VOLUME,
MacosInfraService,
OrchestratorStartError,
probe_control_plane_url,
)
_INFRA = "bot_bottle.backend.macos_container.infra"
def _ok(stdout: str = "") -> Mock:
return Mock(returncode=0, stdout=stdout, stderr="")
def _fail(stderr: str = "boom") -> Mock:
return Mock(returncode=1, stdout="", stderr=stderr)
class TestInfraRun(unittest.TestCase):
def _run_container(self, svc: MacosInfraService) -> list[str]:
run = Mock(return_value=_ok())
def _spec(src: str, tgt: str, readonly: bool = False) -> str:
return f"type=bind,source={src},target={tgt}" + (
",readonly" if readonly else "")
with patch(f"{_INFRA}.container_mod") as mod, \
patch(f"{_INFRA}.ensure_networks"):
mod.dns_server.return_value = "1.1.1.1"
mod.bind_mount_spec.side_effect = _spec
mod.run_container_argv = run
svc._run_container("h1")
return run.call_args.args[0]
def test_single_container_runs_both_processes(self) -> None:
"""The whole point: one container starts the control plane AND the
gateway daemons, so one kernel owns the DB."""
argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
script = argv[-1]
self.assertIn("bot_bottle.orchestrator", script)
self.assertIn("gateway_init.py", script)
self.assertIn("127.0.0.1", script) # they reach each other on loopback
def test_db_is_a_container_only_volume(self) -> None:
"""No host bind-mount of the DB — a named volume only this container
mounts, so the DB is never written by two kernels."""
argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
vols = [argv[i + 1] for i, a in enumerate(argv) if a == "--volume"]
self.assertTrue(any(v.startswith(f"{INFRA_DB_VOLUME}:") for v in vols))
# The repo source is bind-mounted read-only; the DB is not a bind mount.
mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
self.assertTrue(all("bot-bottle.db" not in m for m in mounts))
def test_nat_network_precedes_the_host_only_network(self) -> None:
argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
self.assertEqual(["bot-bottle-mac-egress", "bot-bottle-mac-gateway"], nets)
def test_source_hash_is_labelled_for_recreate(self) -> None:
argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
self.assertIn("BOT_BOTTLE_SOURCE_HASH=h1", argv)
def test_start_failure_raises(self) -> None:
svc = MacosInfraService(repo_root=Path("/r"))
with patch(f"{_INFRA}.container_mod") as mod, \
patch(f"{_INFRA}.ensure_networks"):
mod.dns_server.return_value = "1.1.1.1"
mod.bind_mount_spec.return_value = "m"
mod.run_container_argv = Mock(return_value=_fail())
with self.assertRaises(OrchestratorStartError):
svc._run_container("h1")
class TestInfraEnsureRunning(unittest.TestCase):
def test_current_healthy_container_is_left_alone(self) -> None:
"""Idempotent singleton: N launches must not churn the infra container
and drop every live bottle's control plane."""
svc = MacosInfraService(repo_root=Path("/r"))
run = Mock()
with patch(f"{_INFRA}.container_mod") as mod, \
patch(f"{_INFRA}.source_hash", return_value="h1"), \
patch.object(svc, "_run_container", run), \
patch.object(svc, "is_healthy", return_value=True):
mod.container_is_running.return_value = True
mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
endpoint = svc.ensure_running()
run.assert_not_called()
self.assertEqual("http://192.168.128.2:8099", endpoint.control_plane_url)
self.assertEqual("192.168.128.2", endpoint.gateway_ip)
def test_changed_source_recreates(self) -> None:
svc = MacosInfraService(repo_root=Path("/r"))
run = Mock()
with patch(f"{_INFRA}.container_mod") as mod, \
patch(f"{_INFRA}.source_hash", return_value="h2"), \
patch.object(svc, "ensure_built"), \
patch.object(svc, "_run_container", run), \
patch.object(svc, "is_healthy", return_value=True):
mod.container_is_running.return_value = True
mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
svc.ensure_running()
run.assert_called_once()
def test_wedged_but_current_container_is_recreated(self) -> None:
"""Current source but a dead HTTP server must be recreated, not polled
to death forever health, not just the source label, gates reuse."""
svc = MacosInfraService(repo_root=Path("/r"))
run = Mock()
health = Mock(side_effect=[False, True])
with patch(f"{_INFRA}.container_mod") as mod, \
patch(f"{_INFRA}.source_hash", return_value="h1"), \
patch.object(svc, "ensure_built"), \
patch.object(svc, "_run_container", run), \
patch.object(svc, "is_healthy", health):
mod.container_is_running.return_value = True
mod.container_env.return_value = {"BOT_BOTTLE_SOURCE_HASH": "h1"}
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
svc.ensure_running()
run.assert_called_once()
def test_never_healthy_raises(self) -> None:
svc = MacosInfraService(repo_root=Path("/r"))
with patch(f"{_INFRA}.container_mod") as mod, \
patch(f"{_INFRA}.source_hash", return_value="h1"), \
patch.object(svc, "ensure_built"), \
patch.object(svc, "_run_container"), \
patch.object(svc, "is_healthy", return_value=False):
mod.container_is_running.return_value = False
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
with self.assertRaises(OrchestratorStartError):
svc.ensure_running(startup_timeout=0.01)
class TestCaCertPem(unittest.TestCase):
def test_reads_ca_out_of_the_container(self) -> None:
svc = MacosInfraService(repo_root=Path("/r"))
with patch(f"{_INFRA}.container_mod") as mod:
mod.run_container_argv.return_value = _ok("-----BEGIN CERTIFICATE-----\n")
pem = svc.ca_cert_pem()
self.assertTrue(pem.startswith("-----BEGIN CERTIFICATE-----"))
argv = mod.run_container_argv.call_args.args[0]
self.assertEqual(["container", "exec", "bot-bottle-mac-infra", "cat"], argv[:4])
class TestProbeControlPlane(unittest.TestCase):
def test_returns_url_when_running(self) -> None:
with patch(f"{_INFRA}.container_mod") as mod:
mod.try_container_ipv4_on_network.return_value = "192.168.128.2"
self.assertEqual("http://192.168.128.2:8099", probe_control_plane_url())
def test_empty_when_absent(self) -> None:
with patch(f"{_INFRA}.container_mod") as mod:
mod.try_container_ipv4_on_network.return_value = ""
self.assertEqual("", probe_control_plane_url())
if __name__ == "__main__":
unittest.main()
@@ -11,6 +11,7 @@ import secrets
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path
from unittest.mock import patch
@@ -243,6 +244,82 @@ class TestServerRoundTrip(unittest.TestCase):
self.assertEqual(reg["bottle_id"], attr["bottle_id"])
class TestControlPlaneAuth(unittest.TestCase):
"""The per-host control-plane secret (issue #400): every route but /health
is a trusted-caller op an agent must not be able to drive just because it
can reach the port."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
def test_health_is_public_even_unauthorized(self) -> None:
status, _ = dispatch(self.orch, "GET", "/health", b"", authorized=False)
self.assertEqual(200, status)
def test_unauthorized_denies_every_other_route(self) -> None:
for method, path, body in [
("GET", "/bottles", b""),
("POST", "/bottles", _body({"source_ip": "10.0.0.1"})),
("PUT", "/bottles/x/policy", _body({"policy": "routes: []"})),
("DELETE", "/bottles/x", b""),
("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""),
("POST", "/supervise/respond", _body({"proposal_id": "p", "bottle_slug": "s", "decision": "approve"})),
]:
status, _ = dispatch(self.orch, method, path, body, authorized=False)
self.assertEqual(401, status, f"{method} {path} should be 401 unauthorized")
def test_deny_happens_before_the_registry_is_touched(self) -> None:
"""An unauthorized DELETE must not tear a bottle down. 401, and the
bottle is still there."""
rec = self.orch.registry.register("10.0.0.9", policy="", metadata="")
status, _ = dispatch(
self.orch, "DELETE", f"/bottles/{rec.bottle_id}", b"", authorized=False)
self.assertEqual(401, status)
self.assertIsNotNone(self.orch.registry.get(rec.bottle_id))
def _server_with_secret(self, secret: str):
with patch.dict("os.environ", {"BOT_BOTTLE_CONTROL_PLANE_TOKEN": secret}):
server = make_server(self.orch, "127.0.0.1", 0)
self.addCleanup(server.server_close)
threading.Thread(target=server.serve_forever, daemon=True).start()
self.addCleanup(server.shutdown)
host, port = server.server_address[0], server.server_address[1]
return f"http://{host}:{port}"
def _status(self, url: str, *, header: str | None = None) -> int:
req = urllib.request.Request(url)
if header is not None:
req.add_header("x-bot-bottle-control-auth", header)
try:
return urllib.request.urlopen(req, timeout=5).status
except urllib.error.HTTPError as e:
return e.code
def test_configured_server_enforces_the_header_over_http(self) -> None:
base = self._server_with_secret("s3cret-admin")
# /health is public — no header needed.
self.assertEqual(200, self._status(f"{base}/health"))
# /bottles requires the secret.
self.assertEqual(401, self._status(f"{base}/bottles"))
self.assertEqual(401, self._status(f"{base}/bottles", header="wrong"))
self.assertEqual(200, self._status(f"{base}/bottles", header="s3cret-admin"))
def test_unconfigured_server_runs_open(self) -> None:
"""No secret set (tests / nft-protected Firecracker): open mode, so the
existing round-trip and unit behavior are unchanged."""
with patch.dict("os.environ", {}, clear=False):
import os
os.environ.pop("BOT_BOTTLE_CONTROL_PLANE_TOKEN", None)
server = make_server(self.orch, "127.0.0.1", 0)
self.addCleanup(server.server_close)
self.assertTrue(server.is_authorized(""))
self.assertTrue(server.is_authorized("anything"))
class TestDispatchSupervise(unittest.TestCase):
"""The /supervise/* routes over the pure dispatch()."""
+8 -8
View File
@@ -38,7 +38,7 @@ class TestDockerGateway(unittest.TestCase):
def test_ensure_running_noop_when_up_and_image_current(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=self.sc.name) # running
@@ -58,7 +58,7 @@ class TestDockerGateway(unittest.TestCase):
# a rebuild's new flat daemons take effect.
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=self.sc.name)
@@ -76,7 +76,7 @@ class TestDockerGateway(unittest.TestCase):
def test_ensure_running_starts_the_singleton_when_absent(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
@@ -102,7 +102,7 @@ class TestDockerGateway(unittest.TestCase):
def test_ensure_running_creates_network_when_missing(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:3] == ["docker", "network", "inspect"]:
return _proc(returncode=1, stderr="No such network")
@@ -135,7 +135,7 @@ class TestDockerGateway(unittest.TestCase):
def test_ensure_running_reuses_existing_network(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
return _proc(stdout="") if argv[:2] == ["docker", "ps"] else _proc()
@@ -144,7 +144,7 @@ class TestDockerGateway(unittest.TestCase):
self.assertEqual([], [c for c in calls if c[:3] == ["docker", "network", "create"]])
def test_ensure_running_raises_on_docker_failure(self) -> None:
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")
if argv[:2] == ["docker", "run"]:
@@ -181,7 +181,7 @@ class TestDockerGatewayBuild(unittest.TestCase):
# build-if-missing silently ran a stale single-tenant image.
calls: list[list[str]] = []
def rec(argv: list[str]) -> Mock:
def rec(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
return _proc() # image present, build succeeds
@@ -196,7 +196,7 @@ class TestDockerGatewayBuild(unittest.TestCase):
def test_ensure_built_no_cache_env_forces_full_rebuild(self) -> None:
calls: list[list[str]] = []
def rec(argv: list[str]) -> Mock:
def rec(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
return _proc()
+5 -5
View File
@@ -60,7 +60,7 @@ class TestOrchestratorService(unittest.TestCase):
current = source_hash(self.svc._repo_root)
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
@@ -83,7 +83,7 @@ class TestOrchestratorService(unittest.TestCase):
# effect, same as the gateway's image-staleness check.
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout=ORCHESTRATOR_NAME)
@@ -104,7 +104,7 @@ class TestOrchestratorService(unittest.TestCase):
def test_ensure_running_starts_orchestrator_container_when_absent(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="") # not running
@@ -127,7 +127,7 @@ class TestOrchestratorService(unittest.TestCase):
# gateway data plane — built from Dockerfile.orchestrator when absent.
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="") # orchestrator not running
@@ -148,7 +148,7 @@ class TestOrchestratorService(unittest.TestCase):
def test_ensure_running_skips_orchestrator_image_build_when_present(self) -> None:
calls: list[list[str]] = []
def fake(argv: list[str]) -> Mock:
def fake(argv: list[str], **_kw: object) -> Mock:
calls.append(argv)
if argv[:2] == ["docker", "ps"]:
return _proc(stdout="")