Compare commits
9 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| fd86e7fa99 | |||
| 3bbd839917 | |||
| 5c526860bc | |||
| 492669e620 | |||
| d32f36bb2b | |||
| 4f36b919b2 | |||
| ca91fc4d91 | |||
| 4a607ad098 | |||
| e24b62b6b9 |
@@ -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]:
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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",
|
||||
]
|
||||
@@ -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`.
|
||||
|
||||
|
||||
@@ -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
@@ -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")
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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()}")
|
||||
|
||||
|
||||
@@ -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
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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()
|
||||
@@ -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()
|
||||
|
||||
@@ -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):
|
||||
|
||||
@@ -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."""
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
@@ -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()."""
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
|
||||
@@ -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="")
|
||||
|
||||
Reference in New Issue
Block a user