railiance-platform/scripts/keycape_secrets_engine_openbao_custody.py

#!/usr/bin/env python3
"""Attended custody child for CCR-2026-0033: KeyCape `secrets-engine-openbao` client secret.

SOURCE ONLY until the founder approves this exact child (A3). One run generates
one value in process memory, writes it as the authoritative KV-v2 version with
compare-and-set (cas=0 to create, cas=<current> to rotate with --rotate), then
compare-before-writes the declared KV metadata and writes a metadata-only receipt.

There is no consumer copy: secrets-engine reads the value in memory per command
through the attended operator reader CCR-2026-0034 (founder direction 2026-10-06,
ops-mason plan keycape-secrets-engine-openbao-client-custody). The child never
reads the KV value, never prints it, never puts it in argv or the environment,
and does not touch policies, roles, ESO, KeyCape or the cluster.
"""
from __future__ import annotations

import argparse
import hashlib
import importlib.util
import json
import os
from pathlib import Path
import re
import secrets
import sys
from datetime import datetime, timezone

sys.path.insert(0, str(Path(__file__).resolve().parent))
from state_hub_preflight_lane import ROOT, LaneError, bao, command, data  # noqa: E402

CCR_ID = 'CCR-2026-0033'
CCR_PATH = ROOT / 'credential-change-requests/CCR-2026-0033.yaml'
SIBLING_ID = 'CCR-2026-0034'
SIBLING_PATH = ROOT / 'credential-change-requests/CCR-2026-0034.yaml'
BOUNDARY_DECISION = '632a8893-0d6a-4591-a50b-40bbdb798f32'
KV_PATH = 'platform/workloads/keycape/secrets-engine-openbao'
KV_DATA = 'platform/data/workloads/keycape/secrets-engine-openbao'
KV_METADATA = 'platform/metadata/workloads/keycape/secrets-engine-openbao'
FIELD = 'CLIENT_SECRET'
APPROVERS = ('platform-owner', 'key-cape-owner', 'net-kingdom-owner')
UUID_RE = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$')


def require(condition, reason):
    if not condition:
        raise LaneError(reason)


def attended_envelope():
    return (Path.home().parent.name == '.warden-attended-login'
            and not os.environ.get('BAO_TOKEN') and not os.environ.get('VAULT_TOKEN'))


def _load_validator():
    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)
    return module


def _approved(module, path, ccr_id):
    ccr, errors, _ = module.validate_ccr(path)
    require(not errors and ccr['id'] == ccr_id, 'valid_ccr_required')
    require(ccr['status'] == 'approved', 'approved_ccr_required')
    for role in APPROVERS:
        require(any(c.get('decision') == 'approved' and role in c.get('reviewer', '')
                    for c in ccr['review']['comments']), 'all_named_reviews_required')
    require(ccr['openbao']['auth']['bound_claims_confirmed'] is True, 'bound_claims_confirmation_required')
    require(ccr['openbao']['kv_path'] == KV_PATH and ccr['openbao']['fields'] == [FIELD],
            'exact_approved_contract_required')
    return ccr


def contract(decision_id):
    """Returns (ccr digest, desired metadata). Refuses before any OpenBao call."""
    require(isinstance(decision_id, str) and UUID_RE.match(decision_id), 'approval_decision_id_required')
    require(decision_id != BOUNDARY_DECISION, 'boundary_decision_does_not_approve_custody_child')
    module = _load_validator()
    ccr = _approved(module, CCR_PATH, CCR_ID)
    require(any(c.get('decision') == 'approved' and decision_id in str(c.get('comment', ''))
                for c in ccr['review']['comments']), 'custody_child_decision_not_recorded_in_ccr')
    require(ccr['custody_child']['script'] == 'scripts/keycape_secrets_engine_openbao_custody.py'
            and 'consumer_copy' not in ccr, 'exact_approved_contract_required')
    _approved(module, SIBLING_PATH, SIBLING_ID)
    meta = ccr['openbao'].get('kv_metadata') or {}
    require(meta.get('cas_required') is True and meta.get('max_versions') == 2
            and meta.get('delete_version_after') == '0s' and meta.get('custom_metadata'),
            'declared_kv_metadata_required')
    return hashlib.sha256(CCR_PATH.read_bytes()).hexdigest(), meta


def read_metadata():
    """Metadata only (never the data path). None when the path has never been written."""
    result = bao(['read', '-format=json', KV_METADATA], allow_failure=True)
    if result.returncode != 0:
        require(b'404' in result.stderr or b'No value found' in result.stderr, 'absence_not_proven')
        return None
    return data(result)['data']


def live_version(meta):
    current = meta['current_version']
    version = meta['versions'][str(current)]
    require(not version.get('destroyed') and not version.get('deletion_time'), 'current_version_not_live')
    return current


def write_kv(value, cas):
    result = data(bao(['write', '-format=json', KV_DATA, '-'],
                      payload={'options': {'cas': cas}, 'data': {FIELD: value}}))
    require(result['data']['version'] == cas + 1, 'unexpected_kv_version')
    return result['data']['version'], result.get('request_id')


def metadata_matches(observed, desired):
    observed = observed or {}
    return all((observed.get(key) or {}) == value if key == 'custom_metadata' else observed.get(key) == value
               for key, value in desired.items())


def ensure_metadata(desired):
    """Compare before write; idempotent. Returns 'unchanged' or 'applied'."""
    if metadata_matches(read_metadata(), desired):
        return 'unchanged'
    bao(['write', KV_METADATA, '-'], payload=dict(desired))
    require(metadata_matches(read_metadata(), desired), 'kv_metadata_readback_mismatch')
    return 'applied'


def run(args, receipt):
    receipt['approval_decision_id'] = args.approval_decision
    receipt['ccr_sha256'], desired = contract(args.approval_decision)
    require(attended_envelope(), 'attended_warden_envelope_required')
    os.umask(0o077)
    identity = data(bao(['token', 'lookup', '-format=json']))['data']
    require('platform-admin' in identity['policies'] and 'root' not in identity['policies'],
            'attended_platform_admin_required')
    meta = read_metadata()
    current = None if meta is None else live_version(meta)
    if args.rotate:
        require(current is not None, 'rotation_requires_existing_custody')
        cas = current
    else:
        require(current is None, 'existing_custody_requires_rotate')
        cas = 0
    receipt['phase'] = 'kv_write'
    value = secrets.token_urlsafe(48)
    try:
        version, request_id = write_kv(value, cas)
    finally:
        value = None
    receipt.update(kv_written=True, kv_version=version, kv_request_id=request_id, kv_cas=cas)
    receipt['phase'] = 'kv_metadata'
    receipt['kv_metadata'] = ensure_metadata(desired)
    receipt['status'] = 'custody_written_pending_issuer_delivery'
    receipt['phase'] = 'done'


def main(argv=None):
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument('action', choices=['provision'])
    parser.add_argument('--approval-decision', required=True,
                        help='founder decision id approving this exact custody child (A3)')
    parser.add_argument('--receipt', required=True)
    parser.add_argument('--rotate', action='store_true',
                        help='roll forward an existing entry with cas=<observed current version>')
    args = parser.parse_args(argv)
    receipt = {'schema': 'platform.keycape-secrets-engine-openbao-custody.v1', 'ccr': CCR_ID,
               'status': 'failed', 'phase': 'preflight',
               'started_at': datetime.now(timezone.utc).isoformat(),
               'credential_values_emitted': False, 'client_side_read_admitted': False,
               'kv_written': False}
    fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600)
    try:
        run(args, receipt)
    except Exception as exc:
        receipt['status'] = 'failed'
        receipt['error'] = str(exc) if isinstance(exc, LaneError) else 'contained_operation_failed'
    finally:
        receipt['finished_at'] = datetime.now(timezone.utc).isoformat()
        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__':
    raise SystemExit(main())