railiance-platform/scripts/operators_entry_retire.py

#!/usr/bin/env python3
"""Attended, silent retirement of the deprecated operators/nextcloud entry.

Founder decision 2026-10-10: ``operators/nextcloud`` (the base entry, not
``operators/nextcloud/backup``) is a deprecated credential and is removed. The
operators mount is deliberately delete-free for ``operator-custody`` and absent
from ``platform-admin`` (openbao/policies/operator-custody.hcl: "destroying a
version is a separate, deliberate act under a broader identity"). This helper is
that act, and it grants nothing standing:

1. As platform-admin (warden attended envelope), write one-shot policy
   ``operators-retire-nextcloud``: ``delete`` and ``read`` on
   ``operators/metadata/nextcloud``, ``read`` on
   ``operators/metadata/nextcloud/backup``, ``revoke-self``, nothing else.
2. Mint an orphan token with only that policy (60s, 8 uses, no default policy).
3. With that token: prove the entry exists (record its version count) and that
   ``nextcloud/backup`` exists; metadata-delete ``operators/nextcloud``; prove it
   is absent and that ``nextcloud/backup`` is unchanged.
4. Revoke the token and delete the one-shot policy, on every path.

No value is read. Run from a railiance-platform checkout:

  BAO_ADDR=http://127.0.0.1:18200 VAULT_ADDR=http://127.0.0.1:18200 \\
  warden access openbao-platform-admin-login --exec -- \\
    python3 scripts/operators_entry_retire.py \\
      --confirm 'RETIRE operators/nextcloud' --receipt "$HOME/operators-nextcloud-retire.json"

``--dry-run`` runs steps 1, 2 and the existence checks, deletes nothing, and
cleans up. The process prints nothing; the receipt (0600, refused if it exists)
holds versions, booleans and an outcome code.

Exit codes: 0 retired / dry-run clean / already absent; 1 contained internal
error; 2 precondition refused; 4 policy, token or delete failure; 5 verification
failure (entry still present, or the sibling changed).
"""
from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
import subprocess
import sys
import tempfile

ENTRY = 'nextcloud'
SIBLING = 'nextcloud/backup'
POLICY = 'operators-retire-nextcloud'
CONFIRM = 'RETIRE operators/nextcloud'
SCHEMA = 'platform.operators-entry-retire.v1'
POLICY_HCL = f'''# One-shot: {CONFIRM} (founder decision 2026-10-10). Deleted after use.
path "operators/metadata/{ENTRY}" {{
  capabilities = ["read", "delete"]
}}
path "operators/metadata/{SIBLING}" {{
  capabilities = ["read"]
}}
path "auth/token/revoke-self" {{
  capabilities = ["update"]
}}
'''


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


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


def meta(path, token):
    """Version count of a KV metadata entry, None if provably absent."""
    result = bao(['read', '-format=json', f'operators/metadata/{path}'], token=token)
    if result.returncode != 0:
        if b'404' in result.stderr or b'No value found' in result.stderr:
            return None
        raise Exit(2, 'metadata_read_failed')
    try:
        data = json.loads(result.stdout)['data']
        return {'current_version': data['current_version'], 'versions': len(data.get('versions') or {})}
    except (ValueError, KeyError, TypeError):
        raise Exit(2, 'metadata_unparseable') from None


def run(args, receipt, state):
    if args.confirm != CONFIRM:
        raise Exit(2, 'exact_confirmation_required')
    if (Path.home().parent.name != '.warden-attended-login'
            or os.environ.get('BAO_TOKEN') or os.environ.get('VAULT_TOKEN')):
        raise Exit(2, 'attended_warden_envelope_required')
    ident = bao(['token', 'lookup', '-format=json'])
    if ident.returncode != 0:
        raise Exit(2, 'identity_lookup_failed')
    policies = (json.loads(ident.stdout).get('data') or {}).get('policies') or []
    if 'platform-admin' not in policies or 'root' in policies:
        raise Exit(2, 'attended_platform_admin_required')

    with tempfile.TemporaryDirectory() as tmp:
        hcl = Path(tmp) / 'policy.hcl'
        hcl.write_text(POLICY_HCL)
        if bao(['policy', 'write', POLICY, str(hcl)]).returncode != 0:
            raise Exit(4, 'policy_write_failed')
    state['policy'] = True
    created = bao(['token', 'create', '-orphan', f'-policy={POLICY}', '-no-default-policy',
                   '-ttl=60s', '-explicit-max-ttl=60s', '-use-limit=8', '-format=json'])
    if created.returncode != 0:
        raise Exit(4, 'token_create_failed')
    try:
        state['token'] = json.loads(created.stdout)['auth']['client_token']
    except (ValueError, KeyError, TypeError):
        raise Exit(4, 'token_create_failed') from None
    token = state['token']

    before = meta(ENTRY, token)
    sibling_before = meta(SIBLING, token)
    receipt['entry_before'] = before
    receipt['sibling_before'] = sibling_before
    if sibling_before is None:
        raise Exit(2, 'sibling_absent_refusing_to_guess_layout')
    if before is None:
        return 'already_absent'
    if args.dry_run:
        return 'dry_run_would_retire'

    if bao(['delete', f'operators/metadata/{ENTRY}'], token=token).returncode != 0:
        raise Exit(4, 'delete_failed')
    if meta(ENTRY, token) is not None:
        raise Exit(5, 'entry_still_present')
    sibling_after = meta(SIBLING, token)
    receipt['sibling_after'] = sibling_after
    if sibling_after != sibling_before:
        raise Exit(5, 'sibling_changed')
    receipt['entry_deleted'] = True
    return 'retired'


def cleanup(receipt, state):
    if state.get('token'):
        receipt['token_revoked'] = bao(['token', 'revoke', '-self'], token=state['token']).returncode == 0
    if state.get('policy'):
        receipt['policy_deleted'] = bao(['policy', 'delete', POLICY]).returncode == 0


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('--confirm', default='')
    parser.add_argument('--receipt', type=Path)
    parser.add_argument('--dry-run', action='store_true')
    try:
        args = parser.parse_args(argv)
    except SystemExit:
        return 2
    receipt = {'schema': SCHEMA, 'mode': 'dry-run' if args.dry_run else 'retire',
               'entry': f'operators/{ENTRY}', 'sibling': f'operators/{SIBLING}', 'value_reads': 0}
    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
    try:
        cleanup(receipt, state)
    except Exception:
        receipt['cleanup_error'] = True
    if code == 0 and state.get('policy') and not receipt.get('policy_deleted'):
        receipt['outcome'] += '_policy_cleanup_failed'
        code = 4
    receipt['exit_code'] = code
    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())