Compare commits

...

1 Commits

Author SHA1 Message Date
didericis-claude 0e2af7de62 feat(orchestrator): durable launch-broker secret via TrustDomain (#468)
tracker-policy-pr / check-pr (pull_request) Successful in 10s
lint / lint (push) Successful in 1m0s
test / integration-docker (pull_request) Successful in 1m7s
test / unit (pull_request) Successful in 2m10s
test / coverage (pull_request) Successful in 14s
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 20:36:46 +00:00
8 changed files with 288 additions and 59 deletions
+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:
+48 -31
View File
@@ -23,12 +23,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 +46,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 +79,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
@@ -219,20 +235,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 +276,9 @@ __all__ = [
"Handler",
"HostControlServer",
"make_host_server",
"broker_secret_from_env",
"broker_secret",
"main",
"Json",
"BROKER_SECRET_ENV",
"DEFAULT_PORT",
]
+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",
]
+32 -11
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:
@@ -124,16 +127,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):
@@ -264,7 +285,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 +293,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"])
+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()