refactor: unify component naming — control_plane/control_auth -> orchestrator
tracker-policy-pr / check-pr (pull_request) Successful in 14s
test / integration-docker (pull_request) Successful in 18s
test / unit (pull_request) Successful in 46s
lint / lint (push) Failing after 54s
test / integration-firecracker (pull_request) Successful in 3m21s
test / coverage (pull_request) Successful in 18s
test / publish-infra (pull_request) Has been skipped

The codebase used "control plane" both as an architectural role term AND
as an identifier alias for the orchestrator component, producing
duplicate names for one thing (control_plane_url vs orchestrator_url,
CONTROL_PLANE_PORT, host_control_plane_token, …). Going forward the
concrete component is always named for what it is — Gateway or
Orchestrator — and the plane vocabulary is reserved for prose (module
descriptions, the security argument).

Renamed (identifiers + the in-repo env/wire/file string values, all
setters/getters are in this repo so the change is atomic):

  ControlPlaneServer            -> OrchestratorServer
  control_plane_url             -> orchestrator_url
  probe_control_plane_url       -> probe_orchestrator_url
  host_control_plane_token      -> host_orchestrator_token
  CONTROL_PLANE_PORT            -> ORCHESTRATOR_PORT
  CONTROL_PLANE_TOKEN_ENV/FILE  -> ORCHESTRATOR_TOKEN_ENV/FILENAME
  BOT_BOTTLE_CONTROL_PLANE_TOKEN-> BOT_BOTTLE_ORCHESTRATOR_TOKEN
  control-plane-token (file)    -> orchestrator-token

  control_auth (module)         -> orchestrator_auth  (stays top-level;
                                   the gateway imports it and must not
                                   import the orchestrator/ package)
  CONTROL_AUTH_HEADER           -> ORCHESTRATOR_AUTH_HEADER
  x-bot-bottle-control-auth     -> x-bot-bottle-orchestrator-auth
  CONTROL_AUTH_JWT_ENV          -> ORCHESTRATOR_AUTH_JWT_ENV
  BOT_BOTTLE_CONTROL_AUTH_JWT   -> BOT_BOTTLE_ORCHESTRATOR_AUTH_JWT
  _control_auth_headers         -> _orchestrator_auth_headers

Prose plane-terms ("control plane", "data plane") are preserved,
including the test name test_data_plane_daemons_get_jwt_not_key (it
names the security invariant). Gateway and orchestrator verified to
agree on the renamed wire header; full unit suite green (2243).

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
2026-07-24 18:17:48 -04:00
parent 4166057abc
commit ca1d341d4f
25 changed files with 146 additions and 147 deletions
+100
View File
@@ -0,0 +1,100 @@
"""Role-scoped control-plane credentials (issue #469 review follow-up).
The control plane no longer trusts a single shared bearer secret for every
route. Instead the orchestrator holds a *signing key* and issues short,
HMAC-signed tokens (compact HS256 JWTs) that embed a **role** naming the kind
of caller:
* ``gateway`` — the data plane (egress / git-gate / supervise). Restricted to
the agent-facing routes it actually needs (``/resolve``,
``/supervise/propose``, ``/supervise/poll``).
* ``cli`` — the host operator / launcher. Full access to the mutating and
operator routes (launch/teardown, policy, ``/supervise/respond``, …).
Only the orchestrator (and the host CLI, which shares the host trust domain)
holds the signing key; the gateway is handed a pre-minted ``gateway`` token it
cannot rewrite into a ``cli`` token. So a compromised data-plane process can no
longer approve its own supervise proposals or drive operator routes — it can
only present the ``gateway`` role it was issued (the control plane rejects it
on operator routes with 403).
Stdlib-only (HMAC-SHA256 over a JSON payload); no JWT dependency — the project
carries no runtime pip deps. Tokens are **signed, not encrypted** (the role is
not a secret) and **long-lived** (parity with the static token they replace;
the security win is the unforgeable role claim, not rotation).
"""
from __future__ import annotations
import base64
import binascii
import hashlib
import hmac
import json
ROLE_GATEWAY = "gateway"
ROLE_CLI = "cli"
ROLES: frozenset[str] = frozenset({ROLE_GATEWAY, ROLE_CLI})
_ALG = "HS256"
def _b64url_encode(raw: bytes) -> str:
return base64.urlsafe_b64encode(raw).rstrip(b"=").decode("ascii")
def _b64url_decode(text: str) -> bytes:
padded = text + "=" * (-len(text) % 4)
return base64.urlsafe_b64decode(padded.encode("ascii"))
def _sign(secret: str, signing_input: str) -> str:
mac = hmac.new(secret.encode("utf-8"), signing_input.encode("ascii"), hashlib.sha256)
return _b64url_encode(mac.digest())
# The fixed, canonical JOSE header — the same for every token we mint.
_HEADER_SEGMENT = _b64url_encode(
json.dumps({"alg": _ALG, "typ": "JWT"}, separators=(",", ":")).encode("utf-8")
)
def mint(role: str, secret: str) -> str:
"""A compact HS256 token asserting `role`, signed with `secret`.
Raises ValueError for an unknown role (mint only what the control plane will
accept) or an empty signing key (an unsigned credential is never valid)."""
if role not in ROLES:
raise ValueError(f"unknown control-plane role {role!r}")
if not secret:
raise ValueError("cannot mint a control-plane token without a signing key")
payload = _b64url_encode(json.dumps({"role": role}, separators=(",", ":")).encode("utf-8"))
signing_input = f"{_HEADER_SEGMENT}.{payload}"
return f"{signing_input}.{_sign(secret, signing_input)}"
def verify(token: str, secret: str) -> str | None:
"""The role a valid `token` carries, or None if it is malformed, wrongly
signed, or names an unknown role. Constant-time signature check; rejects any
header whose alg isn't HS256 (no alg-confusion / `none`)."""
if not token or not secret:
return None
parts = token.split(".")
if len(parts) != 3:
return None
header_b64, payload_b64, sig = parts
expected = _sign(secret, f"{header_b64}.{payload_b64}")
if not hmac.compare_digest(sig, expected):
return None
try:
header = json.loads(_b64url_decode(header_b64))
payload = json.loads(_b64url_decode(payload_b64))
except (ValueError, binascii.Error):
return None
if not isinstance(header, dict) or header.get("alg") != _ALG:
return None
role = payload.get("role") if isinstance(payload, dict) else None
return role if isinstance(role, str) and role in ROLES else None
__all__ = ["ROLE_GATEWAY", "ROLE_CLI", "ROLES", "mint", "verify"]