Compare commits

...

2 Commits

Author SHA1 Message Date
didericis-claude 76037f36a6 feat(orchestrator): list_live broker op + reconcile via broker (#468)
lint / lint (push) Successful in 1m5s
refresh-image-locks / refresh (push) Successful in 34s
test / unit (pull_request) Has started running
test / image-input-builds (pull_request) Has started running
test / coverage (pull_request) Blocked by required conditions
test / integration-docker (pull_request) Has been cancelled
tracker-policy-pr / check-pr (pull_request) Successful in 7s
Chunk 3 of the host-control-server stack: grow the broker op vocabulary
(PRD gap 3), starting with `list_live`, and invert `reconcile` onto it.

- broker: split the closed op vocabulary into mutation (`launch`/`teardown`,
  carry a bottle id + static flags) and query (`list_live`, carries nothing
  but its op name) kinds. `verify_request` now enforces a **strict schema**
  (open question 1, resolved yes): unknown claim keys are rejected, a mutation
  must name its bottle, and a query that smuggles any id/flag is refused.
- broker verb: `LaunchBroker.list_live` / `SubmitBroker.list_live` return the
  backend's live source IPs; a backend enumeration failure is converted to the
  single `BrokerUnavailableError` "live set unknown" signal. `DockerBroker`
  enumerates its labelled containers; `StubBroker` derives from launches (or a
  test override).
- host controller: `POST /broker/live` verifies a signed `list_live` token and
  returns `{source_ips}`; `BrokerClient.list_live` is its drop-in client.
- reconcile: `OrchestratorCore.reconcile()` drops the `live_source_ips`
  parameter and pulls the live set from the broker itself — the tell that the
  orchestrator couldn't see the backend goes away. **Fail-safe**: if the broker
  can't return an authoritative set the sweep is skipped, never run against an
  empty/partial set (which would reap healthy rows). The `/reconcile` HTTP
  contract + `OrchestratorClient.reconcile` become a bare trigger.

The macOS launcher's Apple-container enumeration stays for now; it becomes the
host controller's `list_live` when launch itself moves behind the broker (the
pulled-forward chunk 5, next in the stack).

Tests + pyright clean; pylint 10.0 on broker.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-26 22:35:26 +00:00
didericis-claude 4b17e6d683 feat(orchestrator): durable launch-broker secret via TrustDomain (#468)
refresh-image-locks / refresh (push) Successful in 33s
lint / lint (push) Successful in 1m7s
test / image-input-builds (pull_request) Failing after 13m0s
test / integration-docker (pull_request) Has been cancelled
test / unit (pull_request) Failing after 10m48s
test / coverage (pull_request) Has been skipped
tracker-policy-pr / check-pr (pull_request) Successful in 13s
Chunk 2 of the host-control-server stack: close the PRD's **durable
secret** gap and replace chunk 1's BOT_BOTTLE_BROKER_SECRET stopgap.

- trust_domain.py: two new domains. LAUNCH_BROKER holds the durable
  HS256 key both the orchestrator (signer) and the host control server
  (verifier) share for the broker's launch JWT — a host-canonical key
  file minted 0600 on first use, so a restarted orchestrator re-verifies
  against the same key. HOST_CONTROLLER is the separate domain for the
  controller's own lifecycle endpoints, keyed by a key the orchestrator
  never holds (its role is `host`, deliberately outside control-plane
  ROLES). LaunchBrokerProvisioning is the fail-closed seam.
- orchestrator_auth.py: ROLE_HOST, outside ROLES.
- paths.py: key-file + env-var constants for both domains.

Key resolution is split by owner (addresses codex review on #497):
broker_secret(allow_host_file=...) — the host controller / dev-harness
(True) may mint/read the durable host key file it owns; the GUEST
orchestrator (--broker http, default False) must be *injected* the key
and fails closed if it isn't. A guest that fell back to the host file
would mint a process-local key unrelated to the host controller's, so
startup would succeed but every launch would 401 — this prevents that
silent divergence.

Tested: domain boundary + separation, provisioning fail-closed, and
broker_secret env-only (guest) vs host-file (host) resolution. pyright
clean; pylint 9.86.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
2026-07-26 22:35:26 +00:00
22 changed files with 841 additions and 173 deletions
@@ -166,9 +166,17 @@ def register_agent(
# collide with this bottle's — and `by_source_ip` fail-closes on ambiguity,
# which would resolve no policy at all and deny every host. Best-effort: a
# reconciliation failure must not block an otherwise-fine launch.
#
# The orchestrator now pulls the live set from the broker itself, so this is
# a bare trigger. Until macOS moves onto the host controller (the follow-up
# that routes launch through the broker), the orchestrator's broker is the
# stub, whose live set is the bottles it recorded launches for — so the sweep
# only under-reaps (leaves an orphan a cycle longer), never reaps a healthy
# bottle. The Apple-container enumeration below (`live_source_ips`) becomes
# that host controller's `list_live` there.
try:
client.reconcile(live_source_ips(endpoint.network))
except (OrchestratorClientError, EnumerationError) as e:
client.reconcile()
except OrchestratorClientError as e:
info(f"registry reconciliation skipped: {e}")
reg = provision_bottle(
client, source_ip, egress_plan, git_gate_plan, MacosGatewayTransport(),
+11 -8
View File
@@ -17,9 +17,10 @@ from pathlib import Path
from .. import log
from .store.store_manager import StoreManager
from ..paths import LAUNCH_BROKER_KEY_ENV
from .broker import StubBroker, SubmitBroker
from .broker_client import BrokerClient
from .host_server import BROKER_SECRET_ENV, DEFAULT_PORT, broker_secret_from_env
from .host_server import DEFAULT_PORT, broker_secret
from .server import make_server
from .docker_broker import DockerBroker
from .store.registry_store import RegistryStore, default_db_path
@@ -57,17 +58,19 @@ def main(argv: list[str] | None = None) -> int:
# A signing secret ties the orchestrator (signer) to its broker (verifier).
# 'stub' records launches instead of starting anything; 'docker' runs real
# containers in-process; 'http' relays signed requests to a separate host
# control server, which verifies and launches. For 'stub'/'docker' the
# secret is ephemeral (signer and verifier share this process); for 'http'
# it must be the SAME secret the host controller holds, so it is read from
# the shared env var (the chunk-1 stand-in for out-of-band provisioning).
# control server, which verifies and launches. For 'stub'/'docker' the secret
# is ephemeral (signer and verifier share this process). For 'http' it must be
# the SAME key the host controller holds — and this process is the *guest*
# (signer), so it must be given that key by injection, NOT mint its own
# process-local one (which would diverge from the host's and 401 every launch).
broker: SubmitBroker
if args.broker == "http":
secret = broker_secret_from_env()
secret = broker_secret() # env-injected only; no host-file fallback here
if secret is None:
parser.error(
f"--broker http requires a shared signing secret in "
f"${BROKER_SECRET_ENV} (hex), matching the host control server"
f"--broker http requires the launch-broker key injected as "
f"${LAUNCH_BROKER_KEY_ENV} (the host controller owns/mints it); the "
"orchestrator must not mint its own or it would diverge from the host's"
)
broker = BrokerClient(args.host_controller_url)
else:
+109 -19
View File
@@ -32,7 +32,23 @@ from dataclasses import dataclass
from typing import Protocol
_JWT_HEADER = {"alg": "HS256", "typ": "JWT"}
_ALLOWED_OPS = ("launch", "teardown")
# The closed op vocabulary, split by kind (PRD "ids + static flags" rule):
# * mutation ops act on ONE bottle — they carry its id (+ static launch flags);
# * query ops enumerate host state — they carry NO ids or flags at all.
# Each new host-privileged op is added here deliberately; `verify_request`
# enforces the per-kind shape so the privileged surface can't drift.
_MUTATION_OPS = ("launch", "teardown")
_QUERY_OPS = ("list_live",)
_ALLOWED_OPS = _MUTATION_OPS + _QUERY_OPS
# Every claim key a signed request may carry — the request fields plus the two
# per-signature envelope claims. A strict allow-list (open question 1, resolved
# "yes"): an unknown key is rejected outright, so the schema can't silently widen
# as ops are added.
_ALLOWED_CLAIMS = frozenset(
{"op", "bottle_id", "source_ip", "image_ref", "slot", "jti", "iat"}
)
class BrokerAuthError(Exception):
@@ -56,17 +72,28 @@ class BrokerUnavailableError(Exception):
@dataclass(frozen=True)
class LaunchRequest:
"""The structured, un-coercible launch/teardown request. Ids + flags
only — `image_ref` is a content-addressed id, never a path/argv."""
class BrokerRequest:
"""A structured, un-coercible broker request. Ids + static flags only —
`image_ref` is a content-addressed id, never a path/argv.
A **mutation** op (`launch`/`teardown`) names the bottle it acts on in
`bottle_id` (plus launch's static flags). A **query** op (`list_live`) acts
on no single bottle, so it carries nothing but its `op` name — `bottle_id`
and every flag stay empty, and `verify_request` rejects a query that smuggles
any in. `LaunchRequest` is kept as an alias for the launch/teardown callers."""
op: str # one of _ALLOWED_OPS
bottle_id: str
bottle_id: str = ""
source_ip: str = ""
image_ref: str = ""
slot: int | None = None
# Historical name — the request type predates the query ops. Kept so the launch
# path (`OrchestratorCore`, `DockerBroker`) reads unchanged.
LaunchRequest = BrokerRequest
# --- minimal JWS/JWT (HS256), stdlib only ----------------------------------
def _b64url(data: bytes) -> str:
@@ -118,20 +145,35 @@ def verify_request(token: str, secret: bytes) -> LaunchRequest:
raise BrokerAuthError("unexpected header/alg")
if not isinstance(claims, dict):
raise BrokerAuthError("claims are not an object")
# Strict schema (open question 1): a signed request may carry ONLY the known
# claim keys, so the privileged surface can't widen by smuggling an extra
# claim past the fixed field checks below.
unknown = set(claims) - _ALLOWED_CLAIMS
if unknown:
raise BrokerAuthError(f"unknown claim(s): {', '.join(sorted(unknown))}")
op = claims.get("op")
bottle_id = claims.get("bottle_id")
if not isinstance(op, str) or op not in _ALLOWED_OPS:
raise BrokerAuthError("request does not match the broker-request schema")
bottle_id = claims.get("bottle_id", "")
source_ip = claims.get("source_ip", "")
image_ref = claims.get("image_ref", "")
slot = claims.get("slot")
if (
not isinstance(op, str) or op not in _ALLOWED_OPS
or not isinstance(bottle_id, str) or not bottle_id
not isinstance(bottle_id, str)
or not isinstance(source_ip, str) or not isinstance(image_ref, str)
or not (slot is None or isinstance(slot, int))
):
raise BrokerAuthError("request does not match the launch-request schema")
return LaunchRequest(
raise BrokerAuthError("request does not match the broker-request schema")
# Per-kind shape: a mutation names exactly one bottle; a query names none and
# carries no flags (the "no arguments" rule for `list_live`). Enforcing both
# halves keeps `bottle_id`/flags from being a coercible field on a query op.
if op in _MUTATION_OPS:
if not bottle_id:
raise BrokerAuthError(f"{op} requires a bottle_id")
elif bottle_id or source_ip or image_ref or slot is not None:
raise BrokerAuthError(f"{op} is a query op and takes no ids or flags")
return BrokerRequest(
op=op, bottle_id=bottle_id, source_ip=source_ip, image_ref=image_ref, slot=slot
)
@@ -139,14 +181,18 @@ def verify_request(token: str, secret: bytes) -> LaunchRequest:
# --- the broker itself ------------------------------------------------------
class SubmitBroker(Protocol):
"""The single method `OrchestratorCore` depends on: verify a signed token and
perform its op, returning the verified request. Both the in-process
`LaunchBroker` and the out-of-process `BrokerClient` (which relays the token
to the host control server) satisfy it structurally, so the core is unchanged
whether the backend is local or a real host service."""
"""The broker surface `OrchestratorCore` depends on. `submit` verifies a
signed *mutation* token and performs its op, returning the verified request;
`list_live` verifies a signed `list_live` token and returns the source IPs of
the bottles the backend currently has running (reconcile's live set). Both the
in-process `LaunchBroker` and the out-of-process `BrokerClient` (which relays
the token to the host control server) satisfy it structurally, so the core is
unchanged whether the backend is local or a real host service."""
def submit(self, token: str) -> LaunchRequest: ...
def list_live(self, token: str) -> list[str]: ...
class LaunchBroker(abc.ABC):
"""Verifies a signed request came from the orchestrator, then performs
@@ -157,15 +203,35 @@ class LaunchBroker(abc.ABC):
self._secret = secret
def submit(self, token: str) -> LaunchRequest:
"""Verify `token` and perform its op. Returns the verified request;
raises `BrokerAuthError` if provenance/shape fails."""
"""Verify `token` and perform its mutation op. Returns the verified
request; raises `BrokerAuthError` if provenance/shape fails, or if the
token is a non-mutation op (a `list_live` token routed here by mistake)."""
req = verify_request(token, self._secret)
if req.op == "launch":
self._launch(req)
else:
elif req.op == "teardown":
self._teardown(req)
else:
raise BrokerAuthError(f"{req.op} is not a submit op")
return req
def list_live(self, token: str) -> list[str]:
"""Verify a `list_live` `token` and return the source IPs of the bottles
the backend currently has running. Raises `BrokerAuthError` on bad
provenance/shape or a non-`list_live` op; a backend enumeration failure
is converted to `BrokerUnavailableError` — the single "live set could not
be determined" signal reconcile catches to skip the sweep rather than
reaping healthy rows against a partial set (the out-of-process
`BrokerClient` raises the same on an enumeration 502 / no response)."""
req = verify_request(token, self._secret)
if req.op != "list_live":
raise BrokerAuthError(f"{req.op} is not a list_live op")
try:
return list(self._list_live())
except Exception as e: # noqa: BLE001 — any enumeration failure means the
# live set is unknown; reconcile must skip, never reap against it.
raise BrokerUnavailableError(f"list_live enumeration failed: {e}") from e
@abc.abstractmethod
def _launch(self, req: LaunchRequest) -> None:
...
@@ -174,15 +240,31 @@ class LaunchBroker(abc.ABC):
def _teardown(self, req: LaunchRequest) -> None:
...
@abc.abstractmethod
def _list_live(self) -> list[str]:
"""Every running bottle's source IP, backend-native. Must raise (not
return a partial list) if the live set can't be determined
authoritatively, so reconcile skips rather than reaping healthy rows."""
...
class StubBroker(LaunchBroker):
"""Dev-harness broker: records verified requests without launching
anything. Exercises the full sign -> verify -> act contract in-process."""
anything. Exercises the full sign -> verify -> act contract in-process.
Its live set (`_list_live`) is, by default, every bottle it recorded a
launch for and no teardown since — the in-process analogue of enumerating
the backend. Tests that need to simulate a bottle dying out from under the
registry (the case reconcile exists for) set `live_source_ips` explicitly to
override that derived set."""
def __init__(self, secret: bytes) -> None:
super().__init__(secret)
self.launched: list[LaunchRequest] = []
self.torn_down: list[LaunchRequest] = []
# When not None, the exact live set `_list_live` reports — lets a test
# say "only these IPs are still up" regardless of what was launched.
self.live_source_ips: list[str] | None = None
def _launch(self, req: LaunchRequest) -> None:
self.launched.append(req)
@@ -190,10 +272,18 @@ class StubBroker(LaunchBroker):
def _teardown(self, req: LaunchRequest) -> None:
self.torn_down.append(req)
def _list_live(self) -> list[str]:
if self.live_source_ips is not None:
return list(self.live_source_ips)
torn = {r.bottle_id for r in self.torn_down}
return [r.source_ip for r in self.launched
if r.bottle_id not in torn and r.source_ip]
__all__ = [
"BrokerAuthError",
"BrokerUnavailableError",
"BrokerRequest",
"LaunchRequest",
"SubmitBroker",
"LaunchBroker",
+42
View File
@@ -75,6 +75,37 @@ class BrokerClient:
# launched the bottle, so this is NOT a definite failure.
raise BrokerUnavailableError(f"POST /broker: {e}") from e
def list_live(self, token: str) -> list[str]:
"""POST the signed `list_live` token to the host controller and return
the source IPs of the bottles it enumerated as running.
Raises `BrokerAuthError` on a fail-closed 401 (bad provenance/schema);
`BrokerClientError` on any other non-success status or a malformed body
(a backend-enumeration 502 included); or `BrokerUnavailableError` if no
response is obtained. A query has no backend side effect, so — unlike
`submit` — every one of these is a *definite* "no live set"; reconcile
catches all three and skips the sweep rather than reaping against an
unknown or partial set."""
data = json.dumps({"token": token}).encode()
req = urllib.request.Request(
f"{self._base}/broker/live", data=data, method="POST",
headers={"Content-Type": "application/json"},
)
try:
with urllib.request.urlopen(req, timeout=self._timeout) as resp:
return _source_ips_from(_json_object(resp.read()))
except urllib.error.HTTPError as e:
detail = _error_detail(e)
if e.code == 401:
raise BrokerAuthError(
detail or "host controller rejected the request"
) from e
raise BrokerClientError(
f"POST /broker/live: HTTP {e.code} {detail}".rstrip()
) from e
except (urllib.error.URLError, TimeoutError, OSError) as e:
raise BrokerUnavailableError(f"POST /broker/live: {e}") from e
def _json_object(raw: bytes) -> dict[str, object]:
"""Parse a JSON object, tolerating an empty or malformed body (→ {}), like
@@ -99,6 +130,17 @@ def _error_detail(e: urllib.error.HTTPError) -> str:
return detail if isinstance(detail, str) else ""
def _source_ips_from(payload: dict[str, object]) -> list[str]:
"""The `source_ips` list from a `/broker/live` response — every string
entry, ignoring any non-string the host controller should never send. A
missing/!list field is a malformed response (the query is meaningless
without it), so it fails rather than silently reconciling against []."""
raw = payload.get("source_ips")
if not isinstance(raw, list):
raise BrokerClientError("host controller response missing source_ips")
return [ip for ip in raw if isinstance(ip, str) and ip]
def _request_from(payload: dict[str, object]) -> LaunchRequest:
"""Reconstruct the verified `LaunchRequest` the controller echoed, so the
returned value matches the in-process broker's (which returns the request it
+7 -9
View File
@@ -15,7 +15,6 @@ from __future__ import annotations
import json
import urllib.error
import urllib.request
from collections.abc import Iterable
from dataclasses import dataclass
from ..log import debug
@@ -194,14 +193,13 @@ class OrchestratorClient:
raise OrchestratorClientError(f"teardown {bottle_id}: HTTP {status}")
return True
def reconcile(
self, live_source_ips: Iterable[str], *, grace_seconds: float | None = None,
) -> list[str]:
"""Drop registry rows for bottles that are no longer running
(`POST /reconcile`), returning the reaped bottle ids. `live_source_ips`
is the caller's enumeration of its live bottles — the orchestrator
can't see the backend from inside the orchestrator container/VM."""
body: dict[str, object] = {"live_source_ips": list(live_source_ips)}
def reconcile(self, *, grace_seconds: float | None = None) -> list[str]:
"""Trigger the orchestrator's self-heal sweep (`POST /reconcile`),
returning the reaped bottle ids. The caller no longer enumerates the
live set: the orchestrator pulls it from the host controller
(`list_live`) itself, so this is a bare trigger with an optional
`grace_seconds`."""
body: dict[str, object] = {}
if grace_seconds is not None:
body["grace_seconds"] = grace_seconds
payload = self._ok("POST", "/reconcile", body)
+25
View File
@@ -71,6 +71,31 @@ class DockerBroker(LaunchBroker):
f"docker rm failed for {req.bottle_id}: {proc.stderr.strip()}"
)
def _list_live(self) -> list[str]:
"""Every running bottle container's network IP, from `docker ps` filtered
to this broker's label. Raises `DockerBrokerError` if the listing or any
inspect fails, so reconcile treats the live set as unknown and skips the
sweep rather than reaping healthy rows against a partial enumeration."""
listing = self._docker([
"docker", "ps", "--filter", f"label={BOTTLE_ID_LABEL}",
"--format", "{{.ID}}",
])
if listing.returncode != 0:
raise DockerBrokerError(f"docker ps failed: {listing.stderr.strip()}")
ips: list[str] = []
for cid in (line for line in listing.stdout.splitlines() if line):
inspected = self._docker([
"docker", "inspect", "--format",
"{{range .NetworkSettings.Networks}}{{.IPAddress}} {{end}}", cid,
])
if inspected.returncode != 0:
raise DockerBrokerError(
f"docker inspect {cid} failed; live set is not authoritative: "
f"{inspected.stderr.strip()}"
)
ips.extend(ip for ip in inspected.stdout.split() if ip)
return ips
__all__ = [
"DockerBroker",
+81 -35
View File
@@ -8,10 +8,19 @@ This module is that service — the single privileged host component — reached
plane's shape (`orchestrator/server.py`): a pure `dispatch()` for socket-free
testing, wrapped by a thin stdlib `http.server` adapter.
GET /health -> 200 {"status": "ok"}
POST /broker -> 200 {"op", "bottle_id", "source_ip", "image_ref", "slot"}
400 (bad body) | 401 (bad provenance/schema) | 502 (backend)
body: {"token": "<signed launch/teardown JWT>"}
GET /health -> 200 {"status": "ok"}
POST /broker -> 200 {"op", "bottle_id", "source_ip", "image_ref", "slot"}
400 (bad body) | 401 (bad provenance/schema) | 502 (backend)
body: {"token": "<signed launch/teardown JWT>"}
POST /broker/live -> 200 {"source_ips": [...]}
400 (bad body) | 401 (bad provenance/schema) | 502 (backend)
body: {"token": "<signed list_live JWT>"}
The op vocabulary grows one signed op at a time (PRD gap 3); each verb reaches
the backend only through `verify_request`, so nothing free-form ever crosses the
wire. `/broker/live` is the first query op reconcile's live-bottle enumeration,
which the orchestrator (blind to the backend from inside its container) now pulls
from the host controller instead of being handed by the CLI.
Only the **signed token** crosses the wire; the server holds the shared HS256
secret and a real `LaunchBroker` (e.g. `DockerBroker`) and runs the existing
@@ -23,12 +32,15 @@ controller down.
The signed launch token *is* the endpoint's authentication (its provenance is the
whole point of the JWS), so `/broker` needs no separate caller credential; the
host controller's own lifecycle endpoints, which do, arrive with the durable
`TrustDomain` key in a later chunk.
host controller's own lifecycle endpoints, which do, arrive with the `host`-role
tokens of the separate `HOST_CONTROLLER` trust domain in a later chunk.
The shared signing secret is read from `$BOT_BOTTLE_BROKER_SECRET` (hex). That is
a **chunk-1 stopgap**: it must be provisioned to signer and verifier out of band,
which is exactly what the durable `TrustDomain` key in chunk 2 (#476) replaces.
The shared signing secret is the durable **launch-broker `TrustDomain` key**
(#468/#476): a host-canonical key file minted 0600 on first use, provisioned to
the orchestrator (signer) and this server (verifier). A backend launcher injects
it via `$BOT_BOTTLE_LAUNCH_BROKER_KEY`; a host-side dev-harness process reads the
key file directly. Durability is the point a restarted orchestrator re-verifies
against the same key, so re-adoption works.
"""
from __future__ import annotations
@@ -43,16 +55,14 @@ import typing
from urllib.parse import urlsplit
from .. import log
from ..paths import LAUNCH_BROKER_KEY_ENV
from ..trust_domain import LAUNCH_BROKER
from .broker import BrokerAuthError, LaunchBroker
from .docker_broker import DockerBroker
# JSON body payload type (parsed request / rendered response).
Json = dict[str, object]
# The hex-encoded HS256 secret shared with the request signer (the orchestrator).
# Chunk-1 stopgap for the durable, out-of-band `TrustDomain` key of chunk 2.
BROKER_SECRET_ENV = "BOT_BOTTLE_BROKER_SECRET"
# Default host-controller port. Distinct from the orchestrator control plane
# (8099) — a separate privileged component listening on its own socket.
DEFAULT_PORT = 8091
@@ -78,19 +88,34 @@ def _parse_json_object(body: bytes) -> Json:
return obj
def broker_secret_from_env(environ: typing.Mapping[str, str] | None = None) -> bytes | None:
"""The shared HS256 secret from `$BOT_BOTTLE_BROKER_SECRET` (hex), or None
when unset or not valid hex. The signer (orchestrator, `--broker http`) and
the verifier (this server) read the same env var so both hold the same key
the chunk-1 stand-in for out-of-band provisioning."""
env = os.environ if environ is None else environ
raw = env.get(BROKER_SECRET_ENV, "").strip()
if not raw:
return None
try:
return bytes.fromhex(raw)
except ValueError:
return None
def broker_secret(
environ: typing.Mapping[str, str] | None = None, *, allow_host_file: bool = False,
) -> bytes | None:
"""The shared launch-broker HS256 secret, as this process should use it.
Always prefers the key injected into this process's env
(`$BOT_BOTTLE_LAUNCH_BROKER_KEY`). `allow_host_file` decides the fallback when
it is absent, and the distinction is a security boundary:
- **Host-side** processes the host controller and the host dev-harness pass
``allow_host_file=True`` to read (minting on first use) the durable host key
file (``bot_bottle_root()/launch-broker-key``) they legitimately own.
- The **guest orchestrator** (``--broker http``) keeps the default ``False``.
It runs inside a container/VM whose ``bot_bottle_root()`` is process-local,
so minting a file there would silently create a key UNRELATED to the host
controller's — startup would succeed but every launch would be rejected 401.
It must instead be *given* the key by its launcher, and fail closed (None)
if it wasn't, rather than diverge.
None when no key is available (a guest with no injection, or an unwritable
host root)."""
key = LAUNCH_BROKER.key_from_env(environ)
if not key and allow_host_file:
try:
key = LAUNCH_BROKER.signing_key() # host-canonical, minted on first use
except OSError:
return None
return key.encode("utf-8") if key else None
def dispatch( # pylint: disable=too-many-return-statements
@@ -134,6 +159,26 @@ def dispatch( # pylint: disable=too-many-return-statements
"slot": req.slot,
}
if method == "POST" and route == "/broker/live":
try:
data = _parse_json_object(body)
except ValueError as e:
return 400, {"error": f"invalid JSON: {e}"}
token = data.get("token")
if not isinstance(token, str) or not token:
return 400, {"error": "token (string) is required"}
try:
source_ips = broker.list_live(token)
except BrokerAuthError as e:
# Fail-closed: bad signature, malformed token, or a non-list_live op.
return 401, {"error": f"broker auth failed: {e}"}
except Exception as e: # noqa: BLE001 — a backend enumeration failure
# (docker down, inspect failed) is operational, not a control-plane
# bug; surface it as a 502 so the caller skips reconcile rather than
# reaping against a partial set, and keep the controller up.
return 502, {"error": f"backend enumeration failed: {e}"}
return 200, {"source_ips": source_ips}
return 404, {"error": "not found"}
@@ -219,20 +264,22 @@ def main(argv: list[str] | None = None) -> int:
python -m bot_bottle.orchestrator.host_server [--host H] [--port P]
Fail-closed: without a shared `$BOT_BOTTLE_BROKER_SECRET` the server can
verify no request's provenance, so it refuses to start rather than run a
launcher that accepts unsigned input."""
Fail-closed: without the launch-broker key the server can verify no request's
provenance, so it refuses to start rather than run a launcher that accepts
unsigned input. As the host-side owner of the key, it may mint/read the host
key file (`allow_host_file=True`)."""
parser = argparse.ArgumentParser(prog="bot_bottle.orchestrator.host_server")
parser.add_argument("--host", default="127.0.0.1", help="bind address")
parser.add_argument("--port", type=int, default=DEFAULT_PORT, help="bind port (0 = ephemeral)")
args = parser.parse_args(argv)
secret = broker_secret_from_env()
secret = broker_secret(allow_host_file=True)
if secret is None:
sys.stderr.write(
f"host controller: refusing to start without a shared signing secret "
f"(${BROKER_SECRET_ENV}, hex) — it could verify no request's "
"provenance and would relay unsigned launches to the backend\n"
f"host controller: refusing to start without the launch-broker key "
f"(${LAUNCH_BROKER_KEY_ENV}, or a writable host root to mint it) — it "
"could verify no request's provenance and would relay unsigned "
"launches to the backend\n"
)
sys.stderr.flush()
return 2
@@ -258,10 +305,9 @@ __all__ = [
"Handler",
"HostControlServer",
"make_host_server",
"broker_secret_from_env",
"broker_secret",
"main",
"Json",
"BROKER_SECRET_ENV",
"DEFAULT_PORT",
]
+8 -13
View File
@@ -18,8 +18,8 @@ vsock / unix-socket portability caveats):
body: {"env_var_secret"}
DELETE /bottles/<bottle_id> -> 200 {"torn_down": true} | 404 (teardown)
POST /reconcile -> 200 {"reaped": [bottle_id, ...]}
body: {"live_source_ips": [...],
["grace_seconds"]}
body: {["grace_seconds"]} (live set is
pulled from the host controller, not sent)
POST /attribute -> 200 {"bottle_id"} | 403
POST /resolve -> 200 {"bottle_id","policy"} | 403
body: {"source_ip","identity_token"}
@@ -207,20 +207,15 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
return 404, {"error": "no such bottle"}
if method == "POST" and route == "/reconcile":
# Host-driven self-heal: the caller enumerates its live bottles (only
# the host can see the backend) and the orchestrator drops rows for
# every other active bottle. Trusted-caller only — an agent that could
# reach this would be able to unregister its neighbours.
# Host-driven self-heal trigger: the orchestrator pulls its own live set
# from the host controller (`list_live`) and drops rows for every other
# active bottle — the caller no longer supplies the live IPs. Trusted-
# caller only: an agent that could reach this would trigger a sweep that
# unregisters its neighbours. Body is optional (just `grace_seconds`).
try:
data = _parse_json_object(body)
except ValueError as e:
return 400, {"error": f"invalid JSON: {e}"}
raw_ips = data.get("live_source_ips")
if not isinstance(raw_ips, list):
return 400, {"error": "live_source_ips (list of strings) is required"}
if any(not isinstance(ip, str) or not ip for ip in raw_ips):
return 400, {"error": "live_source_ips must contain non-empty strings"}
live = raw_ips
grace = data.get("grace_seconds")
kwargs: dict[str, float] = {}
if grace is not None:
@@ -230,7 +225,7 @@ def dispatch( # pylint: disable=too-many-return-statements,too-many-branches
if not math.isfinite(parsed_grace) or parsed_grace < 0:
return 400, {"error": "grace_seconds must be a non-negative finite number"}
kwargs["grace_seconds"] = parsed_grace
return 200, {"reaped": orch.reconcile(live, **kwargs)}
return 200, {"reaped": orch.reconcile(**kwargs)}
if method == "POST" and route == "/attribute":
try:
+30 -9
View File
@@ -22,10 +22,17 @@ Launch lifecycle:
from __future__ import annotations
import json
from collections.abc import Iterable
from datetime import datetime, timezone
from .broker import BrokerUnavailableError, LaunchRequest, SubmitBroker, sign_request
from .. import log
from .broker import (
BrokerAuthError,
BrokerUnavailableError,
LaunchRequest,
SubmitBroker,
sign_request,
)
from .broker_client import BrokerClientError
from .store.registry_store import DEFAULT_REAP_GRACE_SECONDS, BottleRecord, RegistryStore
from .supervisor import (
AuditEntry,
@@ -145,22 +152,36 @@ class OrchestratorCore:
def reconcile(
self,
live_source_ips: Iterable[str],
*,
grace_seconds: float = DEFAULT_REAP_GRACE_SECONDS,
) -> list[str]:
"""Drop registry rows for bottles that are no longer running, and
forget their in-memory egress tokens. Returns the reaped bottle ids.
The caller supplies the live set because only the host can enumerate
its own containers the orchestrator runs *inside* the infra
container and has no view of the backend. Deliberately does not
broker a teardown: the container is already gone, so there is nothing
to stop, and a broker error must not stop the sweep from clearing
the row that would otherwise brick the next bottle at that address.
The live set is pulled from the **broker** (`list_live`), not passed in:
only the host can enumerate its own containers, and the host controller
is now that host component, so the orchestrator blind to the backend
from inside its container asks it over the same signed seam it launches
through. (Before the host control server this was a `live_source_ips`
argument the CLI handed in; the tell that the orchestrator couldn't see
the backend goes away with it PRD gap 3.)
Fail-safe: if the broker can't return an authoritative live set
(unreachable, timed out, enumeration failed), the sweep is **skipped**,
never run against an empty/partial set reaping a healthy bottle is far
worse than leaving an orphan one more cycle. Deliberately does not broker
a teardown: a reaped container is already gone, so there is nothing to
stop.
See `RegistryStore.reap_absent` for why orphans accumulate and why
they are harmful rather than merely untidy."""
token = sign_request(LaunchRequest(op="list_live"), self._secret)
try:
live_source_ips = self._broker.list_live(token)
except (BrokerUnavailableError, BrokerAuthError, BrokerClientError) as e:
log.info("reconcile skipped: live set unavailable",
context={"error": str(e)})
return []
reaped = self.registry.reap_absent(
live_source_ips, grace_seconds=grace_seconds)
for rec in reaped:
+8 -1
View File
@@ -36,6 +36,13 @@ ROLE_GATEWAY = "gateway"
ROLE_CLI = "cli"
ROLES: frozenset[str] = frozenset({ROLE_GATEWAY, ROLE_CLI})
# The host controller's own lifecycle role (#468). Deliberately OUTSIDE `ROLES`:
# it belongs to a separate trust domain (`HOST_CONTROLLER`) signed by a key the
# orchestrator never holds, so the orchestrator's control-plane key can neither
# mint nor accept it — the orchestrator must not be able to forge the credential
# used to start and stop it.
ROLE_HOST = "host"
_ALG = "HS256"
@@ -103,4 +110,4 @@ def verify(token: str, secret: str, *, roles: frozenset[str] = ROLES) -> str | N
return role if isinstance(role, str) and role in roles else None
__all__ = ["ROLE_GATEWAY", "ROLE_CLI", "ROLES", "mint", "verify"]
__all__ = ["ROLE_GATEWAY", "ROLE_CLI", "ROLE_HOST", "ROLES", "mint", "verify"]
+21
View File
@@ -47,6 +47,22 @@ ORCHESTRATOR_TOKEN_ENV = "BOT_BOTTLE_ORCHESTRATOR_TOKEN"
# cannot forge a higher-privilege `cli` token (issue #469 review).
ORCHESTRATOR_AUTH_JWT_ENV = "BOT_BOTTLE_ORCHESTRATOR_AUTH_JWT"
# The durable launch-broker signing key: the HS256 secret the orchestrator
# (signer) and the host control server (verifier) share to sign/verify launch
# requests (#468). A host-canonical key file (minted 0600 on first use) so it
# survives orchestrator restarts — re-adoption re-verifies against the same key —
# instead of the ephemeral per-process secret of the in-process broker.
LAUNCH_BROKER_KEY_FILENAME = "launch-broker-key"
LAUNCH_BROKER_KEY_ENV = "BOT_BOTTLE_LAUNCH_BROKER_KEY"
# The host controller's OWN key, for its lifecycle endpoints (the direct
# cli -> host controller path that starts/stops the orchestrator). Separate from
# the launch-broker key and never held by the orchestrator: the controller starts
# and stops the orchestrator, so the orchestrator must not be able to mint the
# credentials used to drive it (#468/#476).
HOST_CONTROLLER_KEY_FILENAME = "host-controller-key"
HOST_CONTROLLER_KEY_ENV = "BOT_BOTTLE_HOST_CONTROLLER_KEY"
HOST_CONTROLLER_AUTH_JWT_ENV = "BOT_BOTTLE_HOST_CONTROLLER_AUTH_JWT"
# The host directory holding the gateway's persistent mitmproxy CA. Bind-mounted
# into the infra/gateway container at mitmproxy's confdir so the self-generated
# CA survives container recreation — every agent installs this one CA to trust
@@ -142,6 +158,11 @@ __all__ = [
"ORCHESTRATOR_TOKEN_FILENAME",
"ORCHESTRATOR_TOKEN_ENV",
"ORCHESTRATOR_AUTH_JWT_ENV",
"LAUNCH_BROKER_KEY_FILENAME",
"LAUNCH_BROKER_KEY_ENV",
"HOST_CONTROLLER_KEY_FILENAME",
"HOST_CONTROLLER_KEY_ENV",
"HOST_CONTROLLER_AUTH_JWT_ENV",
"GATEWAY_CA_DIRNAME",
"bot_bottle_root",
"host_db_path",
+87 -1
View File
@@ -29,8 +29,13 @@ from collections.abc import Mapping
from dataclasses import dataclass
from . import orchestrator_auth
from .orchestrator_auth import ROLE_GATEWAY
from .orchestrator_auth import ROLE_GATEWAY, ROLE_HOST
from .paths import (
HOST_CONTROLLER_AUTH_JWT_ENV,
HOST_CONTROLLER_KEY_ENV,
HOST_CONTROLLER_KEY_FILENAME,
LAUNCH_BROKER_KEY_ENV,
LAUNCH_BROKER_KEY_FILENAME,
ORCHESTRATOR_AUTH_JWT_ENV,
ORCHESTRATOR_TOKEN_ENV,
ORCHESTRATOR_TOKEN_FILENAME,
@@ -99,6 +104,40 @@ CONTROL_PLANE = TrustDomain(
)
# The launch-broker domain (#468): durable key material for the broker's own
# signed launch requests (`broker.py`'s HS256 launch JWT), shared by the
# orchestrator (signer) and the host control server (verifier). Unlike
# `CONTROL_PLANE` it mints no role tokens — the broker's provenance is the launch
# JWT, not a role token — so its `roles` set is empty and it is used only as a
# provider of durable, host-canonical key material (`signing_key` / `key_from_env`).
# The durability is the point: the key survives orchestrator restarts, so a
# restarted orchestrator re-verifies against the same key instead of the
# ephemeral per-process secret the in-process broker used.
LAUNCH_BROKER = TrustDomain(
name="launch-broker",
key_filename=LAUNCH_BROKER_KEY_FILENAME,
roles=frozenset(),
key_env=LAUNCH_BROKER_KEY_ENV,
token_env="",
)
# The host controller's own domain (#468) — the SECOND domain #476 reserves. Its
# key, which the orchestrator never holds, signs the `host`-role tokens the CLI
# presents on the host controller's lifecycle endpoints (start / restart / status
# of the orchestrator itself). Keeping it separate from `CONTROL_PLANE` is the
# whole point: the host controller starts and stops the orchestrator, so the
# orchestrator must not be able to mint the credentials used to drive it. (The
# lifecycle endpoints themselves arrive in a later chunk; the domain is
# established here alongside the durable launch-broker key.)
HOST_CONTROLLER = TrustDomain(
name="host-controller",
key_filename=HOST_CONTROLLER_KEY_FILENAME,
roles=frozenset({ROLE_HOST}),
key_env=HOST_CONTROLLER_KEY_ENV,
token_env=HOST_CONTROLLER_AUTH_JWT_ENV,
)
@dataclass(frozen=True)
class ControlPlaneProvisioning:
"""The one seam every backend launcher uses to provision control-plane auth,
@@ -132,9 +171,56 @@ class ControlPlaneProvisioning:
return self.domain.mint(ROLE_GATEWAY)
@dataclass(frozen=True)
class LaunchBrokerProvisioning:
"""The seam that provisions the host-side launch broker's durable keys (#468),
the counterpart to `ControlPlaneProvisioning`. Both the orchestrator (signer)
and the host control server (verifier) receive the SAME launch-broker key
(carry it in `broker_domain.key_env`); the host controller ALSO receives its
own lifecycle key (`controller_domain.key_env`) the orchestrator never holds.
Fail-closed like the control-plane seam: minting returns "" only if the host
root is unwritable, and an empty launch-broker key would leave the verifier
unable to authenticate any launch so we raise rather than hand back a key
that would make the host controller reject (or, if a caller defaulted it,
accept) unsigned input."""
broker_domain: TrustDomain = LAUNCH_BROKER
controller_domain: TrustDomain = HOST_CONTROLLER
def broker_key(self) -> str:
"""The durable launch-broker key both the orchestrator and the host
control server must receive (in `broker_domain.key_env`). Raises rather
than return ""."""
key = self.broker_domain.signing_key()
if not key:
raise ProvisioningError(
f"refusing to provision the {self.broker_domain.name} broker "
"without a signing key: the host controller could then verify no "
"launch request's provenance"
)
return key
def controller_key(self) -> str:
"""The host controller's own lifecycle key — provisioned ONLY to the host
controller (in `controller_domain.key_env`), never to the orchestrator, so
the orchestrator cannot mint the `host`-role tokens that start and stop
it. Raises rather than return ""."""
key = self.controller_domain.signing_key()
if not key:
raise ProvisioningError(
f"refusing to provision the {self.controller_domain.name} without "
"a signing key: its lifecycle endpoints would authenticate no one"
)
return key
__all__ = [
"ProvisioningError",
"TrustDomain",
"CONTROL_PLANE",
"LAUNCH_BROKER",
"HOST_CONTROLLER",
"ControlPlaneProvisioning",
"LaunchBrokerProvisioning",
]
+4 -18
View File
@@ -190,8 +190,7 @@ class TestRegisterAgentReconciles(unittest.TestCase):
def _register(self, client: Mock) -> None:
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_UTIL}.provision_git_gate"), \
patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]):
patch(f"{_UTIL}.provision_git_gate"):
register_agent(
_egress_plan(), _git_plan(),
source_ip="10.0.0.7", endpoint=_endpoint(),
@@ -213,7 +212,9 @@ class TestRegisterAgentReconciles(unittest.TestCase):
client.register_bottle.side_effect = _register_bottle
self._register(client)
self.assertEqual(["reconcile", "register"], calls)
client.reconcile.assert_called_once_with(["10.0.0.7"])
# A bare trigger now — the orchestrator enumerates the live set itself
# (from the host controller), so the CLI passes no IPs.
client.reconcile.assert_called_once_with()
def test_a_reconcile_failure_does_not_block_the_launch(self) -> None:
from bot_bottle.orchestrator.client import OrchestratorClientError
@@ -221,18 +222,3 @@ class TestRegisterAgentReconciles(unittest.TestCase):
client.reconcile.side_effect = OrchestratorClientError("unreachable")
self._register(client)
client.register_bottle.assert_called_once()
def test_enumeration_error_does_not_block_the_launch(self) -> None:
"""A partial container listing must not abort the launch — skip
reconciliation and proceed, just as with an unreachable orchestrator."""
from bot_bottle.backend.macos_container.enumerate import EnumerationError
client = _client()
with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
patch(f"{_UTIL}.provision_git_gate"), \
patch(f"{_MOD}.live_source_ips",
side_effect=EnumerationError("container list failed")):
register_agent(
_egress_plan(), _git_plan(),
source_ip="10.0.0.7", endpoint=_endpoint(),
)
client.register_bottle.assert_called_once()
+85
View File
@@ -2,6 +2,10 @@
from __future__ import annotations
import base64
import hashlib
import hmac
import json
import secrets
import unittest
@@ -14,6 +18,21 @@ from bot_bottle.orchestrator.broker import (
)
def _sign_claims(claims: dict[str, object], secret: bytes) -> str:
"""Sign an arbitrary claims dict (bypassing `sign_request`'s fixed fields)
so a test can craft off-schema payloads e.g. an extra claim key, or a
query op carrying ids with a *valid* signature, isolating the schema check
from the provenance check."""
def b64(data: bytes) -> str:
return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
header = b64(json.dumps({"alg": "HS256", "typ": "JWT"}, sort_keys=True,
separators=(",", ":")).encode())
payload = b64(json.dumps(claims, sort_keys=True, separators=(",", ":")).encode())
signing_input = f"{header}.{payload}"
sig = b64(hmac.new(secret, signing_input.encode("ascii"), hashlib.sha256).digest())
return f"{signing_input}.{sig}"
class TestSignVerify(unittest.TestCase):
def setUp(self) -> None:
self.secret = secrets.token_bytes(16)
@@ -54,6 +73,41 @@ class TestSignVerify(unittest.TestCase):
with self.assertRaises(BrokerAuthError):
verify_request(token, self.secret)
def test_list_live_round_trip(self) -> None:
req = LaunchRequest(op="list_live")
got = verify_request(sign_request(req, self.secret), self.secret)
self.assertEqual("list_live", got.op)
self.assertEqual("", got.bottle_id)
def test_mutation_without_bottle_id_rejected(self) -> None:
# A launch/teardown must name its bottle even with a valid signature.
for op in ("launch", "teardown"):
token = _sign_claims(
{"op": op, "bottle_id": "", "jti": "x", "iat": 1}, self.secret)
with self.assertRaises(BrokerAuthError):
verify_request(token, self.secret)
def test_query_op_carrying_ids_or_flags_rejected(self) -> None:
# list_live is "no arguments" — a signed one that smuggles a bottle_id /
# source_ip / image_ref / slot is off-schema and refused.
for extra in (
{"bottle_id": "b1"}, {"source_ip": "10.0.0.1"},
{"image_ref": "img"}, {"slot": 2},
):
claims: dict[str, object] = {"op": "list_live", "jti": "x", "iat": 1, **extra}
with self.assertRaises(BrokerAuthError):
verify_request(_sign_claims(claims, self.secret), self.secret)
def test_unknown_claim_key_rejected(self) -> None:
# Strict schema (open question 1): a well-signed token with an extra claim
# key is refused, so the surface can't widen past the fixed fields.
token = _sign_claims(
{"op": "launch", "bottle_id": "b1", "cmd": "rm -rf /", "jti": "x", "iat": 1},
self.secret,
)
with self.assertRaises(BrokerAuthError):
verify_request(token, self.secret)
class TestStubBroker(unittest.TestCase):
def setUp(self) -> None:
@@ -81,6 +135,37 @@ class TestStubBroker(unittest.TestCase):
self.broker.submit(forged)
self.assertEqual([], self.broker.launched) # nothing acted on
def test_list_live_derives_from_launches(self) -> None:
# Default live set: launched, minus torn-down, by source IP.
self.broker.submit(
sign_request(LaunchRequest(op="launch", bottle_id="b1", source_ip="10.0.0.1"),
self.secret))
self.broker.submit(
sign_request(LaunchRequest(op="launch", bottle_id="b2", source_ip="10.0.0.2"),
self.secret))
self.broker.submit(
sign_request(LaunchRequest(op="teardown", bottle_id="b1", source_ip="10.0.0.1"),
self.secret))
live = self.broker.list_live(
sign_request(LaunchRequest(op="list_live"), self.secret))
self.assertEqual(["10.0.0.2"], live)
def test_list_live_override(self) -> None:
self.broker.live_source_ips = ["10.0.0.9"]
live = self.broker.list_live(
sign_request(LaunchRequest(op="list_live"), self.secret))
self.assertEqual(["10.0.0.9"], live)
def test_submit_rejects_a_list_live_token(self) -> None:
# The verbs don't cross: a query token routed to submit is fail-closed.
with self.assertRaises(BrokerAuthError):
self.broker.submit(sign_request(LaunchRequest(op="list_live"), self.secret))
def test_list_live_rejects_a_mutation_token(self) -> None:
with self.assertRaises(BrokerAuthError):
self.broker.list_live(
sign_request(LaunchRequest(op="launch", bottle_id="b1"), self.secret))
if __name__ == "__main__":
unittest.main()
@@ -113,5 +113,48 @@ class TestSubmit(unittest.TestCase):
self.c.submit("tok")
class TestListLive(unittest.TestCase):
def setUp(self) -> None:
self.c = BrokerClient("http://host:8091")
def test_returns_source_ips(self) -> None:
with patch(_URLOPEN, return_value=_resp({"source_ips": ["10.0.0.1", "10.0.0.2"]})):
self.assertEqual(["10.0.0.1", "10.0.0.2"], self.c.list_live("tok"))
def test_posts_token_to_live_endpoint(self) -> None:
with patch(_URLOPEN, return_value=_resp({"source_ips": []})) as m:
self.c.list_live("signed-token")
request = m.call_args.args[0]
self.assertEqual("POST", request.get_method())
self.assertTrue(request.full_url.endswith("/broker/live"))
self.assertEqual({"token": "signed-token"}, json.loads(request.data))
def test_non_string_entries_are_filtered(self) -> None:
with patch(_URLOPEN, return_value=_resp({"source_ips": ["10.0.0.1", 5, None, ""]})):
self.assertEqual(["10.0.0.1"], self.c.list_live("tok"))
def test_missing_source_ips_raises(self) -> None:
# A response without the field is malformed — better to fail (reconcile
# skips) than to reconcile against a silent empty set.
with patch(_URLOPEN, return_value=_resp({})):
with self.assertRaises(BrokerClientError):
self.c.list_live("tok")
def test_401_raises_broker_auth_error(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(401, {"error": "bad signature"})):
with self.assertRaises(BrokerAuthError):
self.c.list_live("forged")
def test_502_enumeration_failure_is_client_error(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(502, {"error": "docker down"})):
with self.assertRaises(BrokerClientError):
self.c.list_live("tok")
def test_unreachable_is_unavailable(self) -> None:
with patch(_URLOPEN, side_effect=urllib.error.URLError("refused")):
with self.assertRaises(BrokerUnavailableError):
self.c.list_live("tok")
if __name__ == "__main__":
unittest.main()
+8 -7
View File
@@ -159,26 +159,27 @@ class TestReconcile(unittest.TestCase):
def setUp(self) -> None:
self.c = OrchestratorClient("http://orch:8080")
def test_posts_live_ips_and_returns_reaped(self) -> None:
def test_triggers_sweep_and_returns_reaped(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["b1", "b2"]})) as m:
got = self.c.reconcile(["10.0.0.2", "10.0.0.3"])
got = self.c.reconcile()
self.assertEqual(["b1", "b2"], got)
sent = json.loads(m.call_args.args[0].data)
self.assertEqual(["10.0.0.2", "10.0.0.3"], sent["live_source_ips"])
# The live set is no longer sent — the orchestrator enumerates it.
self.assertNotIn("live_source_ips", sent)
self.assertNotIn("grace_seconds", sent) # omitted -> server default
def test_grace_seconds_is_forwarded_when_given(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": []})) as m:
self.c.reconcile([], grace_seconds=30)
self.c.reconcile(grace_seconds=30)
self.assertEqual(30, json.loads(m.call_args.args[0].data)["grace_seconds"])
def test_malformed_reaped_is_tolerated(self) -> None:
with patch(_URLOPEN, return_value=_resp(200, {"reaped": ["ok", 5, None]})):
self.assertEqual(["ok"], self.c.reconcile([]))
self.assertEqual(["ok"], self.c.reconcile())
with patch(_URLOPEN, return_value=_resp(200, {})):
self.assertEqual([], self.c.reconcile([]))
self.assertEqual([], self.c.reconcile())
def test_error_status_raises(self) -> None:
with patch(_URLOPEN, side_effect=_http_error(500)):
with self.assertRaises(OrchestratorClientError):
self.c.reconcile([])
self.c.reconcile()
@@ -8,6 +8,7 @@ from unittest.mock import Mock, patch
from bot_bottle.orchestrator.broker import (
BrokerAuthError,
BrokerUnavailableError,
LaunchRequest,
sign_request,
)
@@ -96,5 +97,50 @@ class TestDockerBroker(unittest.TestCase):
m.assert_not_called()
class TestDockerBrokerListLive(unittest.TestCase):
def setUp(self) -> None:
self.secret = secrets.token_bytes(16)
self.broker = DockerBroker(self.secret)
def test_enumerates_labeled_container_ips(self) -> None:
# `docker ps` lists two labeled containers; each inspect yields an IP.
ps = Mock(returncode=0, stdout="c1\nc2\n", stderr="")
i1 = Mock(returncode=0, stdout="10.0.0.1 \n", stderr="")
i2 = Mock(returncode=0, stdout="10.0.0.2 \n", stderr="")
with patch.object(self.broker, "_docker", side_effect=[ps, i1, i2]) as m:
got = self.broker._list_live()
self.assertEqual(["10.0.0.1", "10.0.0.2"], got)
# First call is the label-filtered `docker ps`.
ps_argv = m.call_args_list[0].args[0]
self.assertEqual(["docker", "ps", "--filter", f"label={BOTTLE_ID_LABEL}"], ps_argv[:4])
def test_no_containers_is_empty(self) -> None:
with patch.object(self.broker, "_docker",
return_value=Mock(returncode=0, stdout="", stderr="")):
self.assertEqual([], self.broker._list_live())
def test_ps_failure_raises(self) -> None:
with patch.object(self.broker, "_docker",
return_value=Mock(returncode=1, stdout="", stderr="daemon down")):
with self.assertRaises(DockerBrokerError):
self.broker._list_live()
def test_inspect_failure_raises(self) -> None:
ps = Mock(returncode=0, stdout="c1\n", stderr="")
bad = Mock(returncode=1, stdout="", stderr="no such container")
with patch.object(self.broker, "_docker", side_effect=[ps, bad]):
with self.assertRaises(DockerBrokerError):
self.broker._list_live()
def test_enumeration_failure_surfaces_as_unavailable_via_list_live(self) -> None:
# Through the verb, a backend failure is the single "live set unknown"
# signal (BrokerUnavailableError) reconcile catches to skip the sweep.
token = sign_request(LaunchRequest(op="list_live"), self.secret)
with patch.object(self.broker, "_docker",
return_value=Mock(returncode=1, stdout="", stderr="down")):
with self.assertRaises(BrokerUnavailableError):
self.broker.list_live(token)
if __name__ == "__main__":
unittest.main()
+82 -13
View File
@@ -10,7 +10,9 @@ from __future__ import annotations
import http.client
import io
import json
import os
import secrets
import tempfile
import threading
import typing
import unittest
@@ -28,11 +30,12 @@ from bot_bottle.orchestrator.host_server import (
MAX_BODY_BYTES,
Handler,
HostControlServer,
broker_secret_from_env,
broker_secret,
dispatch,
main,
make_host_server,
)
from bot_bottle.paths import LAUNCH_BROKER_KEY_ENV
def _body(obj: object) -> bytes:
@@ -40,8 +43,9 @@ def _body(obj: object) -> bytes:
class _RaisingBroker(LaunchBroker):
"""A broker whose backend launch always fails — exercises the 502 path (an
operational backend failure, distinct from a fail-closed provenance 401)."""
"""A broker whose backend launch/enumeration always fails — exercises the
502 path (an operational backend failure, distinct from a fail-closed
provenance 401)."""
def _launch(self, req: LaunchRequest) -> None:
raise RuntimeError("docker down")
@@ -49,6 +53,9 @@ class _RaisingBroker(LaunchBroker):
def _teardown(self, req: LaunchRequest) -> None:
raise RuntimeError("docker down")
def _list_live(self) -> list[str]:
raise RuntimeError("docker down")
class TestDispatch(unittest.TestCase):
def setUp(self) -> None:
@@ -97,6 +104,42 @@ class TestDispatch(unittest.TestCase):
self.assertEqual(502, status)
self.assertIn("backend launch failed", str(payload["error"]))
def test_broker_live_returns_source_ips(self) -> None:
# Two launched bottles -> the stub reports both as live.
self.broker.submit(self._token(op="launch", bottle_id="b1", source_ip="10.0.0.1"))
self.broker.submit(self._token(op="launch", bottle_id="b2", source_ip="10.0.0.2"))
token = self._token(op="list_live")
status, payload = dispatch(self.broker, "POST", "/broker/live", _body({"token": token}))
self.assertEqual(200, status)
self.assertEqual(
["10.0.0.1", "10.0.0.2"],
sorted(typing.cast("list[str]", payload["source_ips"])),
)
def test_broker_live_forged_token_is_401(self) -> None:
forged = sign_request(LaunchRequest(op="list_live"), secrets.token_bytes(16))
status, payload = dispatch(self.broker, "POST", "/broker/live", _body({"token": forged}))
self.assertEqual(401, status)
self.assertIn("broker auth failed", str(payload["error"]))
def test_broker_live_rejects_a_mutation_token(self) -> None:
# A launch token routed to the query endpoint is a fail-closed 401 — the
# endpoints don't share a schema even though they share a secret.
token = self._token(op="launch", bottle_id="b1", image_ref="img")
status, _ = dispatch(self.broker, "POST", "/broker/live", _body({"token": token}))
self.assertEqual(401, status)
def test_broker_live_backend_failure_is_502(self) -> None:
broker = _RaisingBroker(self.secret)
token = self._token(op="list_live")
status, payload = dispatch(broker, "POST", "/broker/live", _body({"token": token}))
self.assertEqual(502, status)
self.assertIn("backend enumeration failed", str(payload["error"]))
def test_broker_live_missing_token_is_400(self) -> None:
status, _ = dispatch(self.broker, "POST", "/broker/live", _body({}))
self.assertEqual(400, status)
def test_missing_token_is_400(self) -> None:
status, _ = dispatch(self.broker, "POST", "/broker", _body({}))
self.assertEqual(400, status)
@@ -124,16 +167,34 @@ class TestDispatch(unittest.TestCase):
self.assertEqual(200, status)
class TestBrokerSecretFromEnv(unittest.TestCase):
def test_reads_hex_secret(self) -> None:
s = secrets.token_bytes(16)
self.assertEqual(s, broker_secret_from_env({"BOT_BOTTLE_BROKER_SECRET": s.hex()}))
class TestBrokerSecret(unittest.TestCase):
"""The durable launch-broker key (#468/#476): prefer the env-injected key,
else the durable host key file, so signer and verifier resolve the same one."""
def test_unset_is_none(self) -> None:
self.assertIsNone(broker_secret_from_env({}))
def test_reads_injected_key_from_env(self) -> None:
# The injected key is honoured regardless of allow_host_file — both the
# host controller and the guest orchestrator take an injected key.
self.assertEqual(
b"injected-key", broker_secret({LAUNCH_BROKER_KEY_ENV: "injected-key"}))
self.assertEqual(
b"injected-key",
broker_secret({LAUNCH_BROKER_KEY_ENV: "injected-key"}, allow_host_file=True))
def test_invalid_hex_is_none(self) -> None:
self.assertIsNone(broker_secret_from_env({"BOT_BOTTLE_BROKER_SECRET": "not-hex"}))
def test_guest_without_injection_fails_closed(self) -> None:
# The default (guest orchestrator): no env key and NO host-file fallback,
# so it returns None rather than mint a divergent process-local key.
self.assertIsNone(broker_secret({}))
def test_host_side_falls_back_to_the_durable_key_file(self) -> None:
# allow_host_file=True (host controller / dev-harness): mint/read the
# durable host key file, the same key on every call (restart re-adoption).
with tempfile.TemporaryDirectory() as root:
with patch.dict("os.environ", {"BOT_BOTTLE_ROOT": root}, clear=False):
os.environ.pop(LAUNCH_BROKER_KEY_ENV, None)
first = broker_secret(allow_host_file=True)
second = broker_secret(allow_host_file=True)
self.assertTrue(first)
self.assertEqual(first, second)
class TestSeamRoundTrip(unittest.TestCase):
@@ -169,6 +230,14 @@ class TestSeamRoundTrip(unittest.TestCase):
client.submit(forged)
self.assertEqual([], broker.launched) # fail-closed across the wire
def test_list_live_enumerates_over_http(self) -> None:
secret = secrets.token_bytes(16)
broker = StubBroker(secret)
broker.live_source_ips = ["10.0.0.7", "10.0.0.8"]
client = self._serve(broker)
got = client.list_live(sign_request(LaunchRequest(op="list_live"), secret))
self.assertEqual(["10.0.0.7", "10.0.0.8"], got)
class TestRequestLimits(unittest.TestCase):
"""The privileged listener must not let a caller that can merely reach the
@@ -264,7 +333,7 @@ class TestServeUnit(unittest.TestCase):
class TestMain(unittest.TestCase):
def test_fail_closed_without_secret(self) -> None:
with patch("bot_bottle.orchestrator.host_server.broker_secret_from_env",
with patch("bot_bottle.orchestrator.host_server.broker_secret",
return_value=None):
self.assertEqual(2, main(["--port", "0"]))
@@ -272,7 +341,7 @@ class TestMain(unittest.TestCase):
fake = MagicMock()
fake.server_address = ("127.0.0.1", 0)
fake.serve_forever.side_effect = KeyboardInterrupt
with patch("bot_bottle.orchestrator.host_server.broker_secret_from_env",
with patch("bot_bottle.orchestrator.host_server.broker_secret",
return_value=b"k"), \
patch("bot_bottle.orchestrator.host_server.make_host_server",
return_value=fake):
+9 -6
View File
@@ -14,6 +14,7 @@ from pathlib import Path
from unittest.mock import MagicMock, patch
from bot_bottle.orchestrator.__main__ import main
from bot_bottle.paths import LAUNCH_BROKER_KEY_ENV
def _fake_server() -> MagicMock:
@@ -32,7 +33,7 @@ class TestMain(unittest.TestCase):
with patch("bot_bottle.orchestrator.__main__.make_server", return_value=fake), \
patch.dict("os.environ", env or {}, clear=False):
if env is None:
os.environ.pop("BOT_BOTTLE_BROKER_SECRET", None)
os.environ.pop(LAUNCH_BROKER_KEY_ENV, None)
rc = main(argv)
return rc, fake
@@ -46,16 +47,18 @@ class TestMain(unittest.TestCase):
rc, _ = self._run("docker")
self.assertEqual(0, rc)
def test_http_broker_with_secret_serves(self) -> None:
def test_http_broker_with_injected_key_serves(self) -> None:
# The guest orchestrator takes the launch-broker key by injection.
rc, _ = self._run(
"http", env={"BOT_BOTTLE_BROKER_SECRET": secrets.token_bytes(16).hex()})
"http", env={LAUNCH_BROKER_KEY_ENV: secrets.token_urlsafe(16)})
self.assertEqual(0, rc)
def test_http_broker_without_secret_exits(self) -> None:
# Fail-closed: --broker http with no shared secret is a usage error.
def test_http_broker_without_injected_key_exits(self) -> None:
# Fail-closed: no host-file fallback for the guest, so a missing injected
# key is a usage error rather than a silently-minted divergent key.
with tempfile.TemporaryDirectory() as d:
with patch.dict("os.environ", {}, clear=False):
os.environ.pop("BOT_BOTTLE_BROKER_SECRET", None)
os.environ.pop(LAUNCH_BROKER_KEY_ENV, None)
with self.assertRaises(SystemExit):
main(["--db", str(Path(d) / "r.db"), "--broker", "http"])
+21 -25
View File
@@ -624,13 +624,18 @@ if __name__ == "__main__":
class TestReconcileRoute(unittest.TestCase):
"""`POST /reconcile` — the host tells the orchestrator which bottles are
actually up, since the orchestrator can't see the backend from inside the
infra container."""
"""`POST /reconcile` — a bare self-heal trigger. The orchestrator pulls its
own live set from the broker (`list_live`); the caller no longer supplies it,
so the request body carries at most `grace_seconds`."""
def setUp(self) -> None:
self._tmp = tempfile.TemporaryDirectory()
self.orch = _orchestrator(Path(self._tmp.name) / "r.db")
store = RegistryStore(Path(self._tmp.name) / "r.db")
store.migrate()
secret = secrets.token_bytes(16)
# Hold a typed StubBroker so tests can set its live set directly.
self.broker = StubBroker(secret)
self.orch = OrchestratorCore(store, self.broker, secret)
def tearDown(self) -> None:
self._tmp.cleanup()
@@ -647,44 +652,35 @@ class TestReconcileRoute(unittest.TestCase):
def test_reaps_absent_and_reports_ids(self) -> None:
dead = self._old("10.0.0.1")
alive = self._old("10.0.0.2")
status, payload = dispatch(
self.orch, "POST", "/reconcile", _body({"live_source_ips": ["10.0.0.2"]}))
self.broker.live_source_ips = ["10.0.0.2"] # broker: only .2 is up
status, payload = dispatch(self.orch, "POST", "/reconcile", _body({}))
self.assertEqual(200, status)
self.assertEqual([dead], payload["reaped"])
self.assertIsNone(self.orch.registry.get(dead))
self.assertIsNotNone(self.orch.registry.get(alive))
def test_missing_live_source_ips_is_400(self) -> None:
status, _ = dispatch(self.orch, "POST", "/reconcile", _body({}))
self.assertEqual(400, status)
def test_empty_body_triggers_the_sweep(self) -> None:
"""No body is fine — the live set comes from the broker, not the body."""
dead = self._old("10.0.0.1")
self.broker.live_source_ips = [] # broker: nothing running
status, payload = dispatch(self.orch, "POST", "/reconcile", b"")
self.assertEqual(200, status)
self.assertEqual([dead], payload["reaped"])
def test_grace_seconds_is_honoured(self) -> None:
"""A grace window wide enough to cover the row protects it."""
self.orch.registry.register("10.0.0.3")
self.broker.live_source_ips = []
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [], "grace_seconds": 3600}))
self.orch, "POST", "/reconcile", _body({"grace_seconds": 3600}))
self.assertEqual(200, status)
self.assertEqual([], payload["reaped"])
def test_non_string_entries_are_rejected(self) -> None:
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [None, 7, "10.0.0.9"]}))
self.assertEqual(400, status)
self.assertIn("live_source_ips", str(payload["error"]))
def test_empty_live_source_ip_is_rejected(self) -> None:
status, payload = dispatch(
self.orch, "POST", "/reconcile", _body({"live_source_ips": [""]}))
self.assertEqual(400, status)
self.assertIn("live_source_ips", str(payload["error"]))
def test_invalid_grace_seconds_is_rejected(self) -> None:
for value in (True, "30", -1, float("inf"), float("nan")):
with self.subTest(value=value):
status, payload = dispatch(
self.orch, "POST", "/reconcile",
_body({"live_source_ips": [], "grace_seconds": value}))
_body({"grace_seconds": value}))
self.assertEqual(400, status)
self.assertIn("grace_seconds", str(payload["error"]))
+32 -6
View File
@@ -39,6 +39,9 @@ class _FailingBroker(LaunchBroker):
def _teardown(self, req: LaunchRequest) -> None:
pass
def _list_live(self) -> list[str]:
return []
class _UnavailableBroker(LaunchBroker):
"""Verifies the token, then raises the *ambiguous* BrokerUnavailableError —
@@ -51,6 +54,9 @@ class _UnavailableBroker(LaunchBroker):
def _teardown(self, req: LaunchRequest) -> None:
pass
def _list_live(self) -> list[str]:
return []
class TestOrchestrator(unittest.TestCase):
def setUp(self) -> None:
@@ -365,8 +371,9 @@ class TestOrchestratorSupervise(unittest.TestCase):
pid = self.orch.supervise_queue_proposal(
rec.bottle_id, tool=TOOL_EGRESS_ALLOW,
proposed_file="routes:\n - host: google.com\n", justification="j")
# No live source IPs -> the bottle is reaped (grace 0 so it's immediate).
self.assertEqual([rec.bottle_id], self.orch.reconcile([], grace_seconds=0))
# No live source IPs (nothing launched through the stub) -> the bottle
# is reaped (grace 0 so it's immediate).
self.assertEqual([rec.bottle_id], self.orch.reconcile(grace_seconds=0))
self.assertEqual(
{"status": "unknown"}, self.orch.supervise_poll_response(rec.bottle_id, pid))
@@ -411,7 +418,9 @@ class TestOrchestratorReconcile(unittest.TestCase):
live = self.orch.launch_bottle("10.243.0.2", tokens={"EGRESS_TOKEN_0": "keep"})
self._age_all(600)
self.assertEqual([dead.bottle_id], self.orch.reconcile(["10.243.0.2"]))
# The broker reports only .2 as still running -> .1 is reaped.
self.broker.live_source_ips = ["10.243.0.2"]
self.assertEqual([dead.bottle_id], self.orch.reconcile())
self.assertIsNone(self.store.get(dead.bottle_id))
self.assertIsNotNone(self.store.get(live.bottle_id))
# The in-memory egress credential goes with the row.
@@ -423,14 +432,31 @@ class TestOrchestratorReconcile(unittest.TestCase):
broker error must not stop the sweep clearing the row."""
self.orch.launch_bottle("10.243.0.1")
self._age_all(600)
self.broker.launched.clear()
self.orch.reconcile([])
self.broker.live_source_ips = [] # broker reports nothing running
self.orch.reconcile()
self.assertEqual([], self.broker.torn_down)
def test_reconcile_keeps_everything_when_all_are_live(self) -> None:
a = self.orch.launch_bottle("10.243.0.1")
b = self.orch.launch_bottle("10.243.0.2")
self._age_all(600)
self.assertEqual([], self.orch.reconcile(["10.243.0.1", "10.243.0.2"]))
self.broker.live_source_ips = ["10.243.0.1", "10.243.0.2"]
self.assertEqual([], self.orch.reconcile())
self.assertIsNotNone(self.store.get(a.bottle_id))
self.assertIsNotNone(self.store.get(b.bottle_id))
def test_reconcile_skipped_when_broker_cannot_enumerate(self) -> None:
"""A broker that can't return an authoritative live set must NOT be
treated as "nothing is live" that would reap every healthy bottle.
The sweep is skipped instead."""
a = self.orch.launch_bottle("10.243.0.1")
b = self.orch.launch_bottle("10.243.0.2")
self._age_all(600)
def _boom() -> list[str]:
raise RuntimeError("docker ps failed")
self.broker._list_live = _boom # type: ignore[method-assign]
self.assertEqual([], self.orch.reconcile())
self.assertIsNotNone(self.store.get(a.bottle_id))
self.assertIsNotNone(self.store.get(b.bottle_id))
+72 -1
View File
@@ -6,10 +6,13 @@ import unittest
from unittest.mock import patch
from bot_bottle import orchestrator_auth
from bot_bottle.orchestrator_auth import ROLE_CLI, ROLE_GATEWAY
from bot_bottle.orchestrator_auth import ROLE_CLI, ROLE_GATEWAY, ROLE_HOST
from bot_bottle.trust_domain import (
CONTROL_PLANE,
HOST_CONTROLLER,
LAUNCH_BROKER,
ControlPlaneProvisioning,
LaunchBrokerProvisioning,
ProvisioningError,
TrustDomain,
)
@@ -101,5 +104,73 @@ class TestControlPlaneProvisioning(unittest.TestCase):
self.assertNotEqual(ROLE_CLI, CONTROL_PLANE.verify(tok, "k"))
class TestLaunchBrokerAndHostControllerDomains(unittest.TestCase):
"""The real #468 domains: the launch-broker key (shared by orchestrator +
host controller) and the host controller's own lifecycle key."""
def test_launch_broker_mints_no_role_tokens(self) -> None:
# Empty role set — it provides durable key material for the broker's own
# launch JWT, not orchestrator_auth role tokens.
self.assertEqual(frozenset(), LAUNCH_BROKER.roles)
with patch("bot_bottle.trust_domain.host_signing_key", return_value="k"):
with self.assertRaises(ValueError):
LAUNCH_BROKER.mint(ROLE_CLI)
def test_host_controller_signs_host_role_only(self) -> None:
with patch("bot_bottle.trust_domain.host_signing_key", return_value="k"):
tok = HOST_CONTROLLER.mint(ROLE_HOST)
self.assertEqual(ROLE_HOST, HOST_CONTROLLER.verify(tok, "k"))
# A control-plane `cli` token (the orchestrator's key) never verifies as a
# host-controller role — the orchestrator can't forge lifecycle creds.
cli_tok = orchestrator_auth.mint(ROLE_CLI, "k")
self.assertIsNone(HOST_CONTROLLER.verify(cli_tok, "k"))
def test_control_plane_cannot_mint_the_host_role(self) -> None:
# `host` is outside the control-plane role set on purpose.
with patch("bot_bottle.trust_domain.host_signing_key", return_value="k"):
with self.assertRaises(ValueError):
CONTROL_PLANE.mint(ROLE_HOST)
def test_the_three_domains_use_distinct_keys_and_env_vars(self) -> None:
self.assertEqual(3, len({
CONTROL_PLANE.key_filename,
LAUNCH_BROKER.key_filename,
HOST_CONTROLLER.key_filename,
}))
self.assertEqual(3, len({
CONTROL_PLANE.key_env, LAUNCH_BROKER.key_env, HOST_CONTROLLER.key_env,
}))
class TestLaunchBrokerProvisioning(unittest.TestCase):
def test_broker_key_returns_the_durable_key(self) -> None:
prov = LaunchBrokerProvisioning()
with patch("bot_bottle.trust_domain.host_signing_key", return_value="bk"):
self.assertEqual("bk", prov.broker_key())
def test_broker_key_fail_closes_when_empty(self) -> None:
# An empty key would leave the host controller unable to verify any
# launch — fail-closed rather than hand back a useless/dangerous key.
prov = LaunchBrokerProvisioning()
with patch("bot_bottle.trust_domain.host_signing_key", return_value=""):
with self.assertRaises(ProvisioningError):
prov.broker_key()
def test_controller_key_is_distinct_from_the_broker_key(self) -> None:
# The orchestrator holds the broker key but NEVER the controller key.
prov = LaunchBrokerProvisioning()
keys = {"launch-broker-key": "bk", "host-controller-key": "ck"}
with patch("bot_bottle.trust_domain.host_signing_key",
side_effect=keys.__getitem__):
self.assertEqual("bk", prov.broker_key())
self.assertEqual("ck", prov.controller_key())
def test_controller_key_fail_closes_when_empty(self) -> None:
prov = LaunchBrokerProvisioning()
with patch("bot_bottle.trust_domain.host_signing_key", return_value=""):
with self.assertRaises(ProvisioningError):
prov.controller_key()
if __name__ == "__main__":
unittest.main()