railiance-platform/scripts/openbao_revoke_initial_root_token.py

#!/usr/bin/env python3
"""Revoke the held initial OpenBao root token at production entry (NK-WP-0055-T16).

Founder-run in their own terminal, through `kubectl port-forward` with the admin
kubeconfig. The token is read at a hidden prompt and kept in memory only: never
argv, environment, file or output. One 0600 receipt (O_EXCL) records only
non-secret facts.

  check  --addr A --breakglass-addr B --receipt R
      lookup-self only: the held token is valid, `root`, orphan, no TTL; and the
      loopback break-glass listener answers `generate-root` status without a token.
  revoke --addr A --breakglass-addr B --decision ID --receipt R
      the same gates, then `revoke-self` (expects 204) and a fresh lookup-self
      (expects 403). Refuses unless the break-glass `generate-root` path answers,
      so a replacement path exists before the standing token is gone.

A and B must be literal loopback URLs (the port-forwarded main listener 8200 and
break-glass listener 8210). Revoking is irreversible: run it only at production
entry, recorded under a founder decision id.
"""
import argparse
from datetime import datetime, timezone
import getpass
import json
import os
from pathlib import Path
import re
import sys
import urllib.error
import urllib.parse
import urllib.request
import warnings

DECISION_RE = re.compile(r'^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$')
TOKEN_RE = re.compile(r'^[A-Za-z0-9._-]{16,256}$')


class Refused(Exception):
    pass


class NoRedirect(urllib.request.HTTPRedirectHandler):
    def redirect_request(self, *args, **kwargs):
        raise Refused('redirect')


def opener():
    return urllib.request.build_opener(urllib.request.ProxyHandler({}), NoRedirect())


def loopback(value):
    parsed = urllib.parse.urlsplit(value or '')
    if (parsed.scheme != 'http' or parsed.hostname not in ('127.0.0.1', '::1') or parsed.port is None
            or parsed.path not in ('', '/') or parsed.query or parsed.fragment or parsed.username):
        raise Refused('address')
    return value.rstrip('/')


def call(client, method, url, token=None):
    headers = {'X-Vault-Token': token} if token else {}
    request = urllib.request.Request(url, method=method, headers=headers, data=b'' if method == 'POST' else None)
    try:
        with client.open(request, timeout=15) as response:
            body = response.read(65536)
            return response.status, json.loads(body) if body else {}
    except urllib.error.HTTPError as error:
        return error.code, {}


def lookup(client, addr, token):
    status, body = call(client, 'GET', addr + '/v1/auth/token/lookup-self', token)
    if status != 200:
        return {'valid': False, 'lookup_status': status}
    data = body.get('data') or {}
    return {'valid': True, 'policies': sorted(data.get('policies') or []), 'orphan': data.get('orphan') is True,
            'ttl_zero': data.get('ttl') == 0, 'expire_time_null': data.get('expire_time') is None}


def is_initial_root(facts):
    return (facts.get('valid') and facts.get('policies') == ['root'] and facts.get('orphan')
            and facts.get('ttl_zero') and facts.get('expire_time_null'))


def breakglass_available(client, addr):
    status, body = call(client, 'GET', addr + '/v1/sys/generate-root/attempt')
    return status == 200 and isinstance(body.get('started'), bool)


def run(mode, addr, breakglass, token, receipt, decision=None, client=None):
    if receipt.exists() or receipt.is_symlink():
        raise Refused('receipt_exists')
    if not TOKEN_RE.fullmatch(token or ''):
        raise Refused('token_shape')
    if mode == 'revoke' and not DECISION_RE.fullmatch(decision or ''):
        raise Refused('decision')
    addr, breakglass = loopback(addr), loopback(breakglass)
    client = client or opener()
    record = {'schema': 'railiance-platform.openbao-initial-root-token.v1', 'mode': mode,
              'observed_at': datetime.now(timezone.utc).isoformat(), 'decision_id': decision,
              'credential_values_emitted': False}
    record['breakglass_generate_root_available'] = breakglass_available(client, breakglass)
    record['before'] = lookup(client, addr, token)
    if not record['breakglass_generate_root_available']:
        record['status'] = 'refused_no_breakglass_path'
    elif not is_initial_root(record['before']):
        record['status'] = 'refused_not_initial_root_token' if record['before'].get('valid') else 'token_already_invalid'
    elif mode == 'check':
        record['status'] = 'checked'
    else:
        record['revoke_status'] = call(client, 'POST', addr + '/v1/auth/token/revoke-self', token)[0]
        record['after'] = lookup(client, addr, token)
        ok = record['revoke_status'] == 204 and record['after'] == {'valid': False, 'lookup_status': 403}
        record['status'] = 'revoked' if ok else 'revoke_unconfirmed'
    fd = os.open(receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
    with os.fdopen(fd, 'w') as out:
        json.dump(record, out, indent=2, sort_keys=True)
        out.write('\n')
    return record


def main():
    parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
    parser.add_argument('mode', choices=('check', 'revoke'))
    parser.add_argument('--addr', required=True)
    parser.add_argument('--breakglass-addr', required=True)
    parser.add_argument('--decision')
    parser.add_argument('--receipt', type=Path, required=True)
    args = parser.parse_args()
    if not sys.stdin.isatty():
        print('refused: run in your own terminal', file=sys.stderr)
        return 2
    try:
        loopback(args.addr), loopback(args.breakglass_addr)
        with warnings.catch_warnings():
            warnings.simplefilter('error', getpass.GetPassWarning)
            token = getpass.getpass('Held initial root token (hidden): ')
        record = run(args.mode, args.addr, args.breakglass_addr, token, args.receipt.expanduser(), args.decision)
    except (Refused, getpass.GetPassWarning) as error:
        print(f'refused: {error}', file=sys.stderr)
        return 1
    except (urllib.error.URLError, OSError):
        print('refused: OpenBao unreachable or local I/O failure', file=sys.stderr)
        return 1
    print(f"{record['status']}; receipt {args.receipt}")
    return 0 if record['status'] in ('checked', 'revoked') else 1


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