#!/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())