ADHOC-2026-10-10: retire the deprecated operators/nextcloud entry
The founder confirmed operators/nextcloud (not /backup) is deprecated. The operators mount is deliberately delete-free for operator-custody and absent from platform-admin, so a silent helper performs the deletion with a one-shot policy and a 60s orphan token scoped to that entry. It verifies the backup sibling is unchanged and removes the token and policy afterwards. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 469992@bnt-lap001 Assistant-Session: e4b8e420-df0a-441d-83ae-7f19130594d9
This commit is contained in:
parent
64d2543151
commit
7b75823095
3 changed files with 379 additions and 0 deletions
196
scripts/operators_entry_retire.py
Normal file
196
scripts/operators_entry_retire.py
Normal file
|
|
@ -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())
|
||||
130
tests/test_operators_entry_retire.py
Normal file
130
tests/test_operators_entry_retire.py
Normal file
|
|
@ -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()
|
||||
53
workplans/ADHOC-2026-10-10.md
Normal file
53
workplans/ADHOC-2026-10-10.md
Normal file
|
|
@ -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.
|
||||
Loading…
Add table
Add a link
Reference in a new issue