feat(orchestrator): durable launch-broker secret via TrustDomain (#468)

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>

tests/unit/test_orchestrator_host_server.py

@@ -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):

tests/unit/test_orchestrator_main.py

@@ -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"])
 

tests/unit/test_trust_domain.py

@@ -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()