diff --git a/scripts/operators_entry_retire.py b/scripts/operators_entry_retire.py new file mode 100644 index 0000000..032971e --- /dev/null +++ b/scripts/operators_entry_retire.py @@ -0,0 +1,196 @@ +#!/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()) diff --git a/tests/test_operators_entry_retire.py b/tests/test_operators_entry_retire.py new file mode 100644 index 0000000..34aa45a --- /dev/null +++ b/tests/test_operators_entry_retire.py @@ -0,0 +1,130 @@ +"""Offline tests for the operators/nextcloud retirement helper against a fake ``bao``.""" +import json +import os +from pathlib import Path +import subprocess +import sys +import tempfile +import textwrap +import unittest + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT = ROOT / 'scripts/operators_entry_retire.py' +CONFIRM = 'RETIRE operators/nextcloud' +ONE_SHOT = 'hvs.oneshot-' + 'K3l4' * 6 + +# KV calls are refused unless they carry the one-shot token and its policy exists. +FAKE_BAO = textwrap.dedent(f'''\ + #!/usr/bin/env python3 + import json, os, sys + sf = os.environ['FAKE_BAO_STATE'] + st = json.load(open(sf)) + a = sys.argv[1:] + tok = os.environ.get('BAO_TOKEN', 'caller') + st['calls'].append({{'args': a, 'token': tok}}) + def save(): json.dump(st, open(sf, 'w')) + def out(obj): save(); print(json.dumps(obj)); sys.exit(0) + def deny(): save(); sys.stderr.write('permission denied\\n'); sys.exit(2) + if a[:2] == ['token', 'lookup']: out({{'data': {{'policies': st['caller_policies']}}}}) + if a[:2] == ['policy', 'write']: st['policy'] = open(a[3]).read(); save(); sys.exit(0) + if a[:2] == ['policy', 'delete']: st.pop('policy', None); save(); sys.exit(0) + if a[:2] == ['token', 'create']: + if st.get('fail_create'): save(); sys.exit(2) + out({{'auth': {{'client_token': '{ONE_SHOT}'}}}}) + if a[:2] == ['token', 'revoke']: st['revoked'] = tok; save(); sys.exit(0) + if tok != '{ONE_SHOT}' or 'policy' not in st: deny() + if a[0] == 'read': + rest = a[-1].split('/', 2)[2] + if rest not in st['kv']: save(); sys.stderr.write('No value found\\n'); sys.exit(2) + n = st['kv'][rest] + out({{'data': {{'current_version': n, 'versions': {{str(i + 1): {{}} for i in range(n)}}}}}}) + if a[0] == 'delete': + if st.get('fail_delete'): deny() + st['kv'].pop(a[1].split('/', 2)[2], None); save(); sys.exit(0) + save(); sys.exit(9) +''') + + +class Retire(unittest.TestCase): + def setUp(self): + self.tmp = Path(tempfile.mkdtemp()) + (self.tmp / 'bin').mkdir() + (self.tmp / 'bin/bao').write_text(FAKE_BAO) + (self.tmp / 'bin/bao').chmod(0o755) + self.home = self.tmp / '.warden-attended-login' / 'session' + self.home.mkdir(parents=True) + self.state = {'calls': [], 'caller_policies': ['default', 'platform-admin'], + 'kv': {'nextcloud': 2, 'nextcloud/backup': 3}} + + def run_helper(self, *extra, confirm=CONFIRM): + (self.tmp / 'state.json').write_text(json.dumps(self.state)) + receipt = self.tmp / f'r{len(list(self.tmp.glob("r*.json")))}.json' + env = {k: v for k, v in os.environ.items() if k not in ('BAO_TOKEN', 'VAULT_TOKEN')} + env.update(PATH=f'{self.tmp / "bin"}:{env["PATH"]}', HOME=str(self.home), + FAKE_BAO_STATE=str(self.tmp / 'state.json')) + proc = subprocess.run([sys.executable, str(SCRIPT), '--confirm', confirm, '--receipt', str(receipt), *extra], + env=env, capture_output=True, text=True) + self.assertEqual((proc.stdout, proc.stderr), ('', '')) + self.state = json.loads((self.tmp / 'state.json').read_text()) + text = receipt.read_text() + self.assertNotIn(ONE_SHOT, text) + return proc.returncode, json.loads(text) + + def assert_cleaned_up(self, receipt): + self.assertNotIn('policy', self.state) + self.assertTrue(receipt['policy_deleted']) + self.assertEqual(self.state.get('revoked'), ONE_SHOT) + + def test_retires_entry_and_leaves_backup_untouched(self): + code, receipt = self.run_helper() + self.assertEqual((code, receipt['outcome']), (0, 'retired')) + self.assertNotIn('nextcloud', self.state['kv']) + self.assertEqual(self.state['kv']['nextcloud/backup'], 3) + self.assertEqual(receipt['entry_before'], {'current_version': 2, 'versions': 2}) + self.assert_cleaned_up(receipt) + self.assertFalse(any(c['args'][0] == 'read' and '/data/' in c['args'][-1] for c in self.state['calls'])) + + def test_one_shot_policy_is_exact(self): + self.run_helper('--dry-run') + calls = [c for c in self.state['calls'] if c['args'][:2] == ['token', 'create']] + self.assertEqual(len(calls), 1) + args = calls[0]['args'] + for flag in ('-orphan', '-policy=operators-retire-nextcloud', '-no-default-policy', '-ttl=60s'): + self.assertIn(flag, args) + + def test_dry_run_deletes_nothing_and_cleans_up(self): + code, receipt = self.run_helper('--dry-run') + self.assertEqual((code, receipt['outcome']), (0, 'dry_run_would_retire')) + self.assertIn('nextcloud', self.state['kv']) + self.assert_cleaned_up(receipt) + + def test_rerun_reports_already_absent(self): + self.state['kv'].pop('nextcloud') + code, receipt = self.run_helper() + self.assertEqual((code, receipt['outcome']), (0, 'already_absent')) + self.assert_cleaned_up(receipt) + + def test_missing_sibling_refuses(self): + self.state['kv'].pop('nextcloud/backup') + code, receipt = self.run_helper() + self.assertEqual((code, receipt['outcome']), (2, 'sibling_absent_refusing_to_guess_layout')) + self.assertIn('nextcloud', self.state['kv']) + self.assert_cleaned_up(receipt) + + def test_delete_failure_still_cleans_up(self): + self.state['fail_delete'] = True + code, receipt = self.run_helper() + self.assertEqual((code, receipt['outcome']), (4, 'delete_failed')) + self.assert_cleaned_up(receipt) + + def test_refuses_without_confirmation_or_platform_admin(self): + code, receipt = self.run_helper(confirm='RETIRE') + self.assertEqual((code, receipt['outcome']), (2, 'exact_confirmation_required')) + self.state['caller_policies'] = ['operator-custody'] + code, receipt = self.run_helper() + self.assertEqual((code, receipt['outcome']), (2, 'attended_platform_admin_required')) + self.assertFalse(any(c['args'][:2] == ['policy', 'write'] for c in self.state['calls'])) + + +if __name__ == '__main__': + unittest.main() diff --git a/workplans/ADHOC-2026-10-10.md b/workplans/ADHOC-2026-10-10.md new file mode 100644 index 0000000..a4dbb61 --- /dev/null +++ b/workplans/ADHOC-2026-10-10.md @@ -0,0 +1,53 @@ +--- +id: RPF-WP-ADHOC-2026-10-10 +type: workplan +title: "Retire the deprecated operators/nextcloud credential" +domain: financials +repo: railiance-platform +status: active +owner: claude-code +topic_slug: railiance +created: "2026-10-10" +updated: "2026-10-10" +--- + +# RPF-WP-ADHOC-2026-10-10 — retire operators/nextcloud + +## Delete the deprecated base entry, keep operators/nextcloud/backup + +```task +id: RPF-WP-ADHOC-2026-10-10-T01 +status: wait +needs_human: true +priority: low +``` + +The founder confirmed on 2026-10-10 that `operators/nextcloud` (the base entry) +is a deprecated credential to remove. The live backup account stays at +`operators/nextcloud/backup`. The delete cannot be done from the UI by design: +`operator-custody` is delete-free and `platform-admin` has no grant on the +`operators` mount (openbao/policies/operator-custody.hcl). + +`scripts/operators_entry_retire.py` does it without granting anything standing. +As platform-admin it writes a one-shot policy (`read`/`delete` on +`operators/metadata/nextcloud`, `read` on the backup sibling's metadata, +`revoke-self`) and mints a 60s, 8-use orphan token holding only that policy. It +proves the entry and the sibling exist, then metadata-deletes the entry with that +token. It then proves the entry is absent and the sibling unchanged, revokes the +token and deletes the policy on every path. It prints nothing and reads no value. +7 offline tests. + +```bash +cd ~/railiance-platform && git pull --ff-only +export BAO_ADDR=http://127.0.0.1:18200 VAULT_ADDR=http://127.0.0.1:18200 +unset BAO_TOKEN VAULT_TOKEN +W='warden access openbao-platform-admin-login --exec --' +$W python3 scripts/operators_entry_retire.py --dry-run --confirm 'RETIRE operators/nextcloud' --receipt ~/operators-nextcloud-dry.json +cat ~/operators-nextcloud-dry.json +$W python3 scripts/operators_entry_retire.py --confirm 'RETIRE operators/nextcloud' --receipt ~/operators-nextcloud-retire.json +cat ~/operators-nextcloud-retire.json +``` + +**Done when:** the receipt says `retired`, with `sibling_after` equal to +`sibling_before`, `token_revoked` and `policy_deleted` true; and ops-mason drops +`operators/nextcloud` from its custody descriptions.