Prepare bounded State Hub preflight signing lane

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a06ecb-456a-71c2-b41e-0755d336e883

scripts/credential-change.py

@@ -236,6 +236,8 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
         policy_file = require_string(
             openbao.get("policy_file"), "openbao.policy_file", errors
         )
+    if "metadata_read" in openbao and not isinstance(openbao["metadata_read"], bool):
+        errors.append("openbao.metadata_read must be boolean")
     fields = [str(field) for field in require_list(openbao.get("fields"), "openbao.fields", errors)]
     if not fields:
         errors.append("openbao.fields must contain at least one field")
@@ -264,6 +266,14 @@ def validate_workload_kv_read(ccr: dict[str, Any], errors: list[str], warnings:
     if method in {"oidc", "kubernetes"}:
         require_string(auth.get("mount"), "openbao.auth.mount", errors)
         require_string(auth.get("role"), "openbao.auth.role", errors)
+    if method == "kubernetes":
+        if "audience" in auth:
+            require_string(auth["audience"], "openbao.auth.audience", errors)
+        for key in ("token_max_ttl", "token_explicit_max_ttl"):
+            if key in auth and (not isinstance(auth[key], str) or not TTL_RE.match(auth[key])):
+                errors.append(f"openbao.auth.{key} must be a TTL string")
+        if "token_no_default_policy" in auth and not isinstance(auth["token_no_default_policy"], bool):
+            errors.append("openbao.auth.token_no_default_policy must be boolean")
     if method == "oidc":
         redirect_uris = require_list(
             auth.get("allowed_redirect_uris"),
@@ -529,14 +539,12 @@ def generated_policy_hcl(ccr: dict[str, Any]) -> str:
     openbao = ccr["openbao"]
     mount = openbao["mount"]
     suffix = openbao["kv_path"][len(mount) + 1 :]
-    return (
-        f'path "{mount}/data/{suffix}" {{\n'
-        '  capabilities = ["read"]\n'
-        "}\n\n"
-        f'path "{mount}/metadata/{suffix}" {{\n'
-        '  capabilities = ["read"]\n'
-        "}\n"
-    )
+    body = (f'path "{mount}/data/{suffix}" {{\n'
+            '  capabilities = ["read"]\n' "}\n")
+    if openbao.get("metadata_read", True):
+        body += (f'\npath "{mount}/metadata/{suffix}" {{\n'
+                 '  capabilities = ["read"]\n' "}\n")
+    return body
 
 
 
@@ -598,14 +606,17 @@ def auth_payload(ccr: dict[str, Any]) -> dict[str, Any]:
     auth = ccr["openbao"]["auth"]
     if auth["method"] == "kubernetes":
         claims = auth["bound_claims"]
-        return {
+        payload = {
             "bound_service_account_names": claims.get("service_account_names", []),
-            "bound_service_account_namespaces": claims.get(
-                "service_account_namespaces", []
-            ),
+            "bound_service_account_namespaces": claims.get("service_account_namespaces", []),
             "policies": ",".join(auth["policies"]),
             "ttl": auth.get("ttl", "15m"),
         }
+        for key in ("audience", "token_max_ttl", "token_explicit_max_ttl", "token_no_default_policy"):
+            if key in auth:
+                payload[key] = auth[key]
+        return payload
+
 
     payload: dict[str, Any] = {
         "role_type": "oidc",

scripts/state_hub_preflight_lane.py

@@ -0,0 +1,214 @@
+#!/usr/bin/env python3
+"""Attended, silent signing-key writer. Never a workload credential front door."""
+from __future__ import annotations
+
+import argparse
+import hashlib
+import importlib.util
+import json
+import os
+from pathlib import Path
+import secrets
+import subprocess
+import sys
+
+ROOT = Path(__file__).resolve().parents[1]
+CCR = ROOT / 'credential-change-requests/CCR-2026-0015-state-hub-preflight-signing.yaml'
+KV = 'platform/workloads/state-hub/repository-rename-preflight'
+FIELD = 'REPOSITORY_RENAME_PREFLIGHT_SECRET'
+POLICY = 'workload-kv-read-state-hub-rename-preflight'
+ROLE = 'state-hub-rename-preflight-eso'
+SA = 'state-hub-preflight-eso'
+
+
+class LaneError(Exception):
+    """Only fixed, non-secret diagnostics may leave the envelope."""
+
+
+def command(argv, *, payload=None, env=None, allow_failure=False):
+    result = subprocess.run(argv, input=None if payload is None else json.dumps(payload).encode(),
+                            capture_output=True, env=env, timeout=60)
+    if result.returncode and not allow_failure:
+        raise LaneError('command_failed')
+    return result
+
+
+def bao(args, *, payload=None, token=None, allow_failure=False):
+    env = os.environ.copy()
+    if token is not None:
+        env['BAO_TOKEN'] = token
+        env['VAULT_TOKEN'] = token
+    result = command(['bao', *args], payload=payload, env=env, allow_failure=allow_failure)
+    return result
+
+
+def data(result):
+    return json.loads(result.stdout)
+
+
+def approved_contract():
+    spec = importlib.util.spec_from_file_location('credential_change', ROOT / 'scripts/credential-change.py')
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    ccr, errors, _ = module.validate_ccr(CCR)
+    if errors or ccr['status'] not in {'approved', 'applied', 'verified', 'active'}:
+        raise LaneError('approved_ccr_required')
+    auth = ccr['openbao']['auth']
+    expected = {'bound_service_account_names': [SA], 'bound_service_account_namespaces': ['state-hub'],
+                'policies': POLICY, 'ttl': '15m', 'audience': 'openbao',
+                'token_max_ttl': '15m', 'token_explicit_max_ttl': '15m', 'token_no_default_policy': True}
+    if (not auth['bound_claims_confirmed'] or ccr['openbao']['kv_path'] != KV
+            or ccr['openbao']['fields'] != [FIELD] or auth['role'] != ROLE
+            or auth['mount'] != 'kubernetes' or module.auth_payload(ccr) != expected
+            or ccr['openbao'].get('metadata_read') is not False):
+        raise LaneError('exact_contract_required')
+    policy = module.generated_policy_hcl(ccr)
+    if (ROOT / ccr['openbao']['policy_file']).read_text() != policy:
+        raise LaneError('policy_source_mismatch')
+    return expected, policy
+
+
+def assert_fenced(kube):
+    deployment = data(command(kube + ['-n', 'state-hub', 'get', 'deployment', 'state-hub', '-o', 'json']))
+    pods = data(command(kube + ['-n', 'state-hub', 'get', 'pods', '-l', 'app=state-hub', '-o', 'json']))
+    hpas = data(command(kube + ['-n', 'state-hub', 'get', 'hpa', '-o', 'json']))
+    if deployment['spec'].get('replicas', 1) != 0 or pods['items'] or hpas['items']:
+        raise LaneError('all_api_replicas_must_be_stopped_without_autoscaler')
+
+
+def revoke(token):
+    bao(['write', 'auth/token/revoke', '-'], payload={'token': token})
+
+
+def capabilities(token, paths):
+    return data(bao(['write', '-format=json', 'sys/capabilities', '-'],
+                    payload={'paths': paths, 'token': token}))['data']
+
+
+def verify_access(kube, receipt):
+    jwt = command(kube + ['-n', 'state-hub', 'create', 'token', SA,
+                          '--audience=openbao', '--duration=10m']).stdout.decode().strip()
+    auth = data(bao(['write', '-format=json', 'auth/kubernetes/login', '-'],
+                    payload={'role': ROLE, 'jwt': jwt}))['auth']
+    token = auth['client_token']
+    try:
+        if auth['token_policies'] != [POLICY] or auth['lease_duration'] > 900:
+            raise LaneError('effective_policy_or_ttl_mismatch')
+        paths = [KV.replace('platform/', 'platform/data/', 1),
+                 KV.replace('platform/', 'platform/metadata/', 1),
+                 'platform/data/workloads/state-hub/forge-derivation',
+                 'platform/metadata/workloads/state-hub']
+        caps = capabilities(token, paths)
+        if caps[paths[0]] != ['read'] or any(caps[p] != ['deny'] for p in paths[1:]):
+            raise LaneError('scope_negative_check_failed')
+        # Native GET, no secret value emitted or retained in receipt.
+        value = data(bao(['read', '-format=json', paths[0]], token=token))['data']['data'][FIELD]
+        if len(value) != 64 or any(c not in '0123456789abcdef' for c in value):
+            raise LaneError('invalid_key_shape')
+        receipt['exact_read_and_scope_denials'] = True
+    finally:
+        revoke(token)
+    for label, namespace, service_account, audience in [
+        ('wrong_sa', 'state-hub', 'default', 'openbao'),
+        ('wrong_namespace', 'default', SA, 'openbao'),
+        ('wrong_audience', 'state-hub', SA, 'not-openbao'),
+    ]:
+        temporary = label == 'wrong_namespace'
+        if temporary:
+            # Exclusive create fails if an unrelated identity already exists.
+            command(kube + ['-n', namespace, 'create', 'serviceaccount', service_account])
+        try:
+            jwt = command(kube + ['-n', namespace, 'create', 'token', service_account,
+                                  '--audience=' + audience, '--duration=10m']).stdout.decode().strip()
+            result = bao(['write', '-format=json', 'auth/kubernetes/login', '-'],
+                         payload={'role': ROLE, 'jwt': jwt}, allow_failure=True)
+            if result.returncode == 0:
+                revoke(data(result)['auth']['client_token'])
+                raise LaneError('negative_login_unexpectedly_succeeded')
+            if b'403' not in result.stderr and b'400' not in result.stderr:
+                raise LaneError('negative_login_inconclusive')
+            receipt[label] = True
+        finally:
+            if temporary:
+                command(kube + ['-n', namespace, 'delete', 'serviceaccount', service_account])
+    agent = data(bao(['read', '-format=json', 'auth/approle/role/coding-agent-railiance-platform']))['data']
+    if 'agent-high-risk-boundary' not in agent['token_policies']:
+        raise LaneError('coding_agent_boundary_missing')
+    child = data(bao(['token', 'create', '-format=json', '-policy=' + POLICY,
+                     '-policy=agent-high-risk-boundary', '-no-default-policy', '-ttl=60s']))['auth']['client_token']
+    try:
+        caps = capabilities(child, paths[:2])
+        if any(caps[p] != ['deny'] for p in paths[:2]):
+            raise LaneError('coding_agent_deny_failed')
+        receipt['coding_agent_deny_wins'] = True
+    finally:
+        revoke(child)
+
+
+def run(args, receipt):
+    role, policy = approved_contract()
+    identity = data(bao(['token', 'lookup', '-format=json']))['data']
+    if 'platform-admin' not in identity['policies'] or 'root' in identity['policies']:
+        raise LaneError('attended_platform_admin_required')
+    kube = ['kubectl', '--kubeconfig', args.kubeconfig]
+    if args.action == 'verify':
+        verify_access(kube, receipt)
+        receipt['status'] = 'custody_verified_pending_eso_and_api_acceptance'
+        return
+    if args.action == 'provision':
+        if args.expected_version != 0:
+            raise LaneError('bootstrap_requires_cas_zero')
+        # Refuse drift rather than overwriting another operator's policy.
+        boundary = ROOT / 'openbao/policies/agent-high-risk-boundary.hcl'
+        current = data(bao(['read', '-format=json', 'sys/policies/acl/agent-high-risk-boundary']))['data']['policy']
+        baseline = (ROOT / 'openbao/policies/inputs/state-hub-preflight-boundary-baseline.sha256').read_text().strip()
+        if current != boundary.read_text() and hashlib.sha256(current.encode()).hexdigest() != baseline:
+            raise LaneError('boundary_policy_drift')
+        bao(['write', 'sys/policies/acl/agent-high-risk-boundary', '-'], payload={'policy': boundary.read_text()})
+        existing = bao(['read', '-format=json', 'auth/kubernetes/role/' + ROLE], allow_failure=True)
+        if existing.returncode == 0:
+            raise LaneError('role_already_exists_review_partial_apply')
+        if b'No value found' not in existing.stderr and b'404' not in existing.stderr:
+            raise LaneError('role_absence_not_proven')
+        bao(['write', 'sys/policies/acl/' + POLICY, '-'], payload={'policy': policy})
+        bao(['write', 'auth/kubernetes/role/' + ROLE, '-'], payload=role)
+        command(kube + ['apply', '-f', str(ROOT / 'openbao/state-hub-preflight/delivery.yaml')])
+    else:
+        if args.expected_version < 1:
+            raise LaneError('rotation_requires_current_version')
+        assert_fenced(kube)
+    # Never read or import an old key; protected CSPRNG generation and CAS only.
+    result = data(bao(['write', '-format=json', KV.replace('platform/', 'platform/data/', 1), '-'],
+                      payload={'options': {'cas': args.expected_version}, 'data': {FIELD: secrets.token_hex(32)}}))
+    receipt['kv_version'] = result['data']['version']
+    receipt['key_generation'] = 'CSPRNG-32-bytes-CAS'
+    verify_access(kube, receipt)
+    receipt['status'] = 'custody_verified_pending_eso_and_api_acceptance'
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument('action', choices=['provision', 'rotate', 'verify'])
+    parser.add_argument('--expected-version', required=True, type=int)
+    parser.add_argument('--kubeconfig', required=True)
+    parser.add_argument('--receipt', required=True)
+    parser.add_argument('--confirm', required=True)
+    args = parser.parse_args()
+    receipt = {'schema': 'platform.statehub-preflight-custody.v1', 'status': 'failed', 'action': args.action}
+    # Exclusive creation before mutations; no symlinks or overwriting old evidence.
+    fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
+    try:
+        if args.confirm != 'APPLY CCR-2026-0015':
+            raise LaneError('confirmation_mismatch')
+        run(args, receipt)
+    except Exception as error:
+        receipt['error'] = str(error) if isinstance(error, LaneError) else 'internal_error'
+    finally:
+        with os.fdopen(fd, 'w') as out:
+            json.dump(receipt, out, indent=2)
+            out.write('\n')
+    return 0 if receipt['status'] != 'failed' else 1
+
+
+if __name__ == '__main__':
+    sys.exit(main())