railiance-platform/scripts/keycape_client_secret_generate.py

#!/usr/bin/env python3
"""RPF-WP-0054-T03: generate a KeyCape confidential-client secret inside custody.

Entry point for the oidc-browser-relying-party ``mint-secret`` step (lane
``keycape-client-secret-generate``). Normally reached as
``credential.py generate ...``; ops-warden runs it as a child in the caller's
own environment and holds no token.

  credential.py generate --grant keycape/client-secret-generate \\
    --client-id forgejo --case-file <case.yaml> \\
    --case-ref forgejo-netkingdom-oidc@sha256:<hex64> --receipt <file>

Steps (every one fails closed):

1. Refuse agents (``WARDEN_AGENT_ID``) and root. The caller's session must hold
   the broker issuer policy for this grant.
2. Re-derive the case reference from ``--case-file`` (sha256 over its exact
   bytes) and require: it equals ``--case-ref``; type
   an exactly source-pinned acknowledged browser revision; status ``materialized``; client_type
   ``confidential``; ``parameters.client_id`` equals ``--client-id``, which
   matches the type's pattern and is not a reserved entry.
3. Mint a 5-minute grant token through token role
   ``keycape-client-secret-generate``. All KV calls below use only that token.
4. Prove absence by metadata (404). An existing entry is never touched.
5. Generate the value in this process and write it with CAS 0 through the
   ``bao`` child's stdin.
6. Verify by metadata only: current version 1, live. The data path is never read.
7. Revoke the grant token (revoke-self).

The process prints nothing. The receipt (mode 0600, refused if it exists) and
the optional State Hub event hold the path, version, case reference, booleans and
an outcome code only.

Exit codes: 0 generated (dry-run: would generate); 1 contained internal error;
2 precondition refused; 3 entry already exists (never overwritten, including a
lost CAS race); 4 token or write failure; 5 verification failure.
"""
from __future__ import annotations

import argparse
import hashlib
import json
import os
from pathlib import Path
import re
import secrets
import subprocess
import sys
import tempfile
import urllib.request

import yaml

GRANT = 'keycape/client-secret-generate'
TOKEN_ROLE = 'keycape-client-secret-generate'
GRANT_POLICY = 'keycape-client-secret-generate'
ISSUER_POLICY = 'credential-broker-keycape-client-secret-generate-issuer'
PREFIX = 'workloads/keycape'
RESERVED = {'secrets-engine-openbao'}
FIELD = 'CLIENT_SECRET'
# Owner configuration, never a caller-supplied root or a version search/fallback.
# The sibling source checkout must contain these exact reviewed bytes.
NETKINGDOM_ROOT = Path(__file__).resolve().parents[2] / 'net-kingdom'
ACKNOWLEDGED_TYPES = {
    'oidc-browser-relying-party@1.1': (
        'security-case-types/oidc-browser-relying-party/type.yaml',
        'f705c64e4875e9b8e18a912fb51c135e77898926d6a451f5cb74ab3c0cbe4d1c'),
}
# Same authenticated producer and parameter projection as ops-warden case_inputs
# at b593c879, restricted to this broker's currently admitted 1.1. Source-only
# acknowledgment (including 1.2) never expands broker admission. A future 1.3
# entry requires real acknowledgment and a separately reviewed admission binding.
# Neither source status nor caller-supplied YAML expands this set.
SOURCE_PINS = {
    'tools/security-case-types/security_case_validator.py':
        '1bcc62e9bc374c711363295549f3cd5bb0b32952671ed4b08fddc7f74742959b',
    'tools/security-scenario-composer/security_scenario_composer.py':
        '9f0cc69930aa8ee2cc3fb92236e6a99728df17b8c0fe36e1250241c38fc01c4f',
    'tools/playbook-capability-contract/playbook_contract_validator.py':
        '2e1e0c809d4bc9d64c2d974cc23fecd1c5636785154c2554edd9191852fd926d',
}
PARAMETER_PINS = {
    '1.1': '9ec2d6113bf70dbd35cfc247412363ff187e792bc82e24159a6f78e090aa3e4c',
}
_VALIDATE = """
import json, sys
from pathlib import Path
sys.path.insert(0, str(Path(sys.argv[1]) / 'tools/security-case-types'))
import security_case_validator as v
case, typ = json.load(sys.stdin)
issues = v.validate_type(typ) + v.validate_case(case, typ, require_implemented=False)
print('valid' if not v.has_errors(issues) else 'invalid')
"""
# oidc-browser-relying-party@1.1 parameter client_id.
CLIENT_ID_RE = re.compile(r'^[a-z0-9][a-z0-9-]{1,62}$')
CASE_REF_RE = re.compile(r'^[a-z0-9][a-z0-9-]*@sha256:[0-9a-f]{64}$')
# 48 random bytes -> 64 URL-safe characters, confirmed by key-cape 2026-10-10
# (message 4e0213b3; opaque constant-time comparison, no alphabet limit).
SECRET_BYTES = 48
TOKEN_TTL = '5m'
SCHEMA = 'platform.keycape-client-secret-generate.v1'


class Exit(Exception):
    def __init__(self, code: int, reason: str):
        super().__init__(reason)
        self.code = code
        self.reason = reason


def bao(args, token=None, payload=None):
    env = dict(os.environ)
    if token is not None:
        env['BAO_TOKEN'] = env['VAULT_TOKEN'] = token
    return subprocess.run(['bao', *args], env=env,
                          input=None if payload is None else json.dumps(payload).encode(),
                          capture_output=True, timeout=60)


def absent(result):
    return result.returncode != 0 and (b'404' in result.stderr or b'No value found' in result.stderr)


def read_json(result, code):
    try:
        return json.loads(result.stdout)
    except ValueError:
        raise Exit(code, 'unparseable_bao_output') from None


def authenticated_source(relative, expected):
    path = NETKINGDOM_ROOT / relative
    # Refuse symlinks at every selected component, including the configured root.
    if any(part.is_symlink() for part in (path, *path.parents)):
        raise ValueError
    raw = path.read_bytes()
    if hashlib.sha256(raw).hexdigest() != expected:
        raise ValueError
    return raw


def check_acknowledged_case(doc):
    # Authenticates the selected TYPE/producer bytes and validates case schema.
    # The caller's self-hashed case reference is not authenticated case admission
    # or owner acceptance. Those require independent Warden/native evidence;
    # this source-only hardening does not complete the real admission binding.
    binding = ACKNOWLEDGED_TYPES.get(doc['spec'].get('type'))
    if binding is None:
        raise Exit(2, 'case_type_unacknowledged')
    try:
        typ = yaml.safe_load(authenticated_source(*binding))
        spec = typ['spec']
        version = str(spec['version'])
        fingerprint = hashlib.sha256(json.dumps(
            spec['parameters'], sort_keys=True, separators=(',', ':')).encode()).hexdigest()
        if (spec['status'] not in ('acknowledged', 'implemented')
                or f"{spec['id']}@{version}" != doc['spec']['type']
                or PARAMETER_PINS.get(version) != fingerprint):
            raise ValueError
        with tempfile.TemporaryDirectory(prefix='platform-case-validator-') as directory:
            for relative, expected in SOURCE_PINS.items():
                raw = authenticated_source(relative, expected)
                dest = Path(directory) / relative
                dest.parent.mkdir(parents=True, exist_ok=True)
                dest.write_bytes(raw)
            result = subprocess.run(
                [sys.executable, '-I', '-c', _VALIDATE, directory],
                input=json.dumps([doc, typ]), text=True,
                stdout=subprocess.PIPE, stderr=subprocess.DEVNULL,
                env={}, timeout=10, check=True)
            if result.stdout != 'valid\n':
                raise ValueError
    except (OSError, ValueError, TypeError, KeyError, yaml.YAMLError,
            subprocess.SubprocessError):
        raise Exit(2, 'case_source_validation_failed') from None


def check_case(args, receipt):
    if not CLIENT_ID_RE.match(args.client_id) or args.client_id in RESERVED:
        raise Exit(2, 'client_id_invalid_or_reserved')
    if not CASE_REF_RE.match(args.case_ref):
        raise Exit(2, 'case_ref_malformed')
    try:
        raw = Path(args.case_file).read_bytes()
        doc = yaml.safe_load(raw)
    except (OSError, yaml.YAMLError):
        raise Exit(2, 'case_file_unreadable') from None
    spec = (doc or {}).get('spec') if isinstance(doc, dict) else None
    if not isinstance(spec, dict):
        raise Exit(2, 'case_file_shape')
    params = spec.get('parameters') or {}
    derived = f"{spec.get('id')}@sha256:{hashlib.sha256(raw).hexdigest()}"
    if derived != args.case_ref:
        raise Exit(2, 'case_ref_mismatch')
    if not isinstance(spec.get('type'), str):
        raise Exit(2, 'case_type_unacknowledged')
    if spec.get('status') != 'materialized':
        raise Exit(2, 'case_not_materialized')
    if params.get('client_type') != 'confidential':
        raise Exit(2, 'case_client_not_confidential')
    if params.get('client_id') != args.client_id:
        raise Exit(2, 'case_client_id_mismatch')
    check_acknowledged_case(doc)
    receipt['case_verified'] = True


def run(args, receipt, state):
    if args.grant != GRANT:
        raise Exit(2, 'unknown_grant')
    if os.environ.get('WARDEN_AGENT_ID'):
        raise Exit(2, 'agent_callers_refused')
    check_case(args, receipt)

    ident = bao(['token', 'lookup', '-format=json'])
    if ident.returncode != 0:
        raise Exit(2, 'caller_session_missing')
    policies = (read_json(ident, 2).get('data') or {}).get('policies') or []
    if 'root' in policies:
        raise Exit(2, 'root_refused')
    if ISSUER_POLICY not in policies:
        raise Exit(2, 'caller_lacks_issuer_policy')

    created = bao(['token', 'create', f'-role={TOKEN_ROLE}', f'-policy={GRANT_POLICY}',
                   f'-ttl={TOKEN_TTL}', '-format=json'])
    if created.returncode != 0:
        raise Exit(4, 'grant_token_create_failed')
    try:
        state['token'] = read_json(created, 4)['auth']['client_token']
    except (KeyError, TypeError):
        raise Exit(4, 'grant_token_create_failed') from None
    token = state['token']

    meta_path = f'platform/metadata/{PREFIX}/{args.client_id}'
    meta = bao(['read', '-format=json', meta_path], token=token)
    if meta.returncode == 0:
        raise Exit(3, 'already_present')
    if not absent(meta):
        raise Exit(2, 'absence_not_proven')
    if args.dry_run:
        return 'dry_run_would_generate'

    value = secrets.token_urlsafe(SECRET_BYTES)
    written = bao(['write', '-format=json', f'platform/data/{PREFIX}/{args.client_id}', '-'],
                  token=token, payload={'options': {'cas': 0}, 'data': {FIELD: value}})
    del value
    if written.returncode != 0:
        if bao(['read', '-format=json', meta_path], token=token).returncode == 0:
            raise Exit(3, 'already_present_lost_race')
        raise Exit(4, 'write_failed')

    meta = bao(['read', '-format=json', meta_path], token=token)
    if meta.returncode != 0:
        raise Exit(5, 'verify_metadata_unreadable')
    data = read_json(meta, 5).get('data') or {}
    version = (data.get('versions') or {}).get('1') or {}
    if data.get('current_version') != 1 or version.get('deletion_time') or version.get('destroyed'):
        raise Exit(5, 'verify_version_mismatch')
    receipt['kv_version'] = 1
    return 'generated'


def record_state_hub(url, receipt):
    body = {
        'summary': (f"keycape-client-secret-generate {receipt['outcome']}: {receipt['path']} "
                    f"v{receipt.get('kv_version', '-')} case {receipt['case_ref']}"),
        'event_type': 'credential_change_evidence',
        'author': 'railiance-platform-credential-broker',
    }
    request = urllib.request.Request(url.rstrip('/') + '/progress/', data=json.dumps(body).encode(),
                                     headers={'Content-Type': 'application/json'}, method='POST')
    with urllib.request.urlopen(request, timeout=10) as response:
        return 200 <= response.status < 300


def main(argv=None):
    devnull = os.open(os.devnull, os.O_WRONLY)
    os.dup2(devnull, 1)
    os.dup2(devnull, 2)
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument('--grant', required=True)
    parser.add_argument('--client-id', required=True)
    parser.add_argument('--case-ref', required=True)
    parser.add_argument('--case-file', required=True)
    parser.add_argument('--receipt', type=Path)
    parser.add_argument('--dry-run', action='store_true')
    parser.add_argument('--record-state-hub', action='store_true')
    parser.add_argument('--state-hub-url', default=os.environ.get('STATE_HUB_URL', 'http://127.0.0.1:8000'))
    try:
        args = parser.parse_args(argv)
    except SystemExit:
        return 2
    receipt = {'schema': SCHEMA, 'mode': 'dry-run' if args.dry_run else 'generate',
               'grant': args.grant, 'client_id': args.client_id, 'case_ref': args.case_ref,
               'path': f'platform/{PREFIX}/{args.client_id}', 'value_reads': 0,
               'credential_values_emitted': False}
    fd = None
    if args.receipt is not None:
        try:
            fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
        except OSError:
            return 2
    state = {}
    try:
        receipt['outcome'] = run(args, receipt, state)
        code = 0
    except Exit as exc:
        receipt['outcome'] = exc.reason
        code = exc.code
    except Exception:
        receipt['outcome'] = 'contained_internal_error'
        code = 1
    if state.get('token'):
        receipt['grant_token_revoked'] = bao(['token', 'revoke', '-self'], token=state['token']).returncode == 0
    receipt['exit_code'] = code
    if args.record_state_hub:
        try:
            receipt['state_hub_recorded'] = record_state_hub(args.state_hub_url, receipt)
        except Exception:
            receipt['state_hub_recorded'] = False
    if fd is not None:
        os.write(fd, (json.dumps(receipt, indent=2, sort_keys=True) + '\n').encode())
        os.close(fd)
    return code


if __name__ == '__main__':
    sys.exit(main())