#!/usr/bin/env python3 """Attended, silent retirement of a duplicate KV entry nothing may still read. Founder ruling 2026-10-09 (ops-mason message b72ff77d): ``platform/workloads/activity-core/default/llm-connect-provider-secrets`` (v1, 2026-06-18) duplicates the live ``platform/workloads/activity-core/llm-connect/llm-connect-provider-secrets`` and no source references it. Before the metadata delete this process proves: * the canonical entry exists with a live current version; * the duplicate exists with exactly the expected version count; * no live ACL policy other than platform-admin grants a non-deny capability on the duplicate's data path (exact, ``+`` and trailing ``*`` globs). Grants on the metadata path only (e.g. mount-wide ``platform/metadata/*`` survey and read-only inventory policies) cannot read the value, so they are recorded in the receipt as ``metadata_only_policies`` but do not block. 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/kv_duplicate_retire.py \\ --confirm 'RETIRE activity-core/default/llm-connect-provider-secrets' \\ --receipt "$HOME/kv-duplicate-retire-receipt.json" ``--dry-run`` runs every check and deletes nothing. The process prints nothing; the receipt (mode 0600, refused if it exists) holds versions, booleans, the names of granting policies and an outcome code. Exit codes: 0 retired / dry-run clean / already absent; 1 contained internal error; 2 precondition refused; 3 a live policy still grants the duplicate; 4 delete failed; 5 duplicate still present after delete. """ from __future__ import annotations import argparse import json import os from pathlib import Path import re import sys sys.path.insert(0, str(Path(__file__).resolve().parent)) from tenant_engine_audit_core_sender_seed import Exit, bao, live_version, read_json # noqa: E402 DUPLICATE = 'workloads/activity-core/default/llm-connect-provider-secrets' CANONICAL = 'workloads/activity-core/llm-connect/llm-connect-provider-secrets' EXPECTED_VERSIONS = 1 CONFIRM = 'RETIRE activity-core/default/llm-connect-provider-secrets' SCHEMA = 'platform.kv-duplicate-retire.v1' EXEMPT_POLICIES = {'platform-admin', 'root'} STANZA_RE = re.compile(r'path\s+"([^"]+)"\s*\{(.*?)\}', re.S) CAPS_RE = re.compile(r'capabilities\s*=\s*\[([^\]]*)\]') def glob_matches(pattern, path): regex = ''.join('[^/]+' if part == '+' else re.escape(part) for part in re.split(r'(\+)', pattern.rstrip('*'))) return re.fullmatch(regex + ('.*' if pattern.endswith('*') else ''), path) is not None def granting_stanzas(policy_text, paths): """Patterns in ``policy_text`` that grant anything but deny on one of ``paths``.""" hits = [] for pattern, body in STANZA_RE.findall(policy_text): caps = CAPS_RE.search(body) names = {c.strip().strip('"') for c in caps.group(1).split(',')} if caps else set() if names - {'deny', ''} and 'deny' not in names and any(glob_matches(pattern, p) for p in paths): hits.append(pattern) return hits def policy_names(): result = bao(['policy', 'list', '-format=json']) if result.returncode != 0: raise Exit(2, 'policy_list_failed') try: names = json.loads(result.stdout) except ValueError: raise Exit(2, 'policy_list_unparseable') from None if not isinstance(names, list) or not names: raise Exit(2, 'policy_list_unparseable') return names def run(args, receipt): 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 = read_json('auth/token/lookup-self', absent_ok=False) policies = ident.get('policies') or [] if 'platform-admin' not in policies or 'root' in policies: raise Exit(2, 'attended_platform_admin_required') canonical = read_json('platform/metadata/' + CANONICAL, absent_ok=True) receipt['canonical_live'] = canonical is not None and live_version(canonical) is not None if not receipt['canonical_live']: raise Exit(2, 'canonical_not_live') duplicate = read_json('platform/metadata/' + DUPLICATE, absent_ok=True) receipt['duplicate_present_before'] = duplicate is not None if duplicate is None: return 'already_absent' receipt['duplicate_versions'] = len(duplicate.get('versions') or {}) if duplicate.get('current_version') != EXPECTED_VERSIONS or receipt['duplicate_versions'] != EXPECTED_VERSIONS: raise Exit(2, 'duplicate_version_unexpected') data_path, metadata_path = 'platform/data/' + DUPLICATE, 'platform/metadata/' + DUPLICATE granting, metadata_only = {}, {} for name in policy_names(): if name in EXEMPT_POLICIES: continue result = bao(['policy', 'read', name]) if result.returncode != 0: raise Exit(2, 'policy_read_failed') text = result.stdout.decode() hits = granting_stanzas(text, [data_path]) if hits: granting[name] = hits else: meta_hits = granting_stanzas(text, [metadata_path]) if meta_hits: metadata_only[name] = meta_hits receipt['granting_policies'] = granting receipt['metadata_only_policies'] = metadata_only if granting: raise Exit(3, 'live_policy_grants_duplicate') if args.dry_run: return 'dry_run_would_retire' if bao(['delete', 'platform/metadata/' + DUPLICATE]).returncode != 0: raise Exit(4, 'delete_failed') if read_json('platform/metadata/' + DUPLICATE, absent_ok=True) is not None: raise Exit(5, 'duplicate_still_present') receipt['duplicate_deleted'] = True return 'retired' 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', 'duplicate': 'platform/' + DUPLICATE, 'canonical': 'platform/' + CANONICAL, '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 try: receipt['outcome'] = run(args, receipt) code = 0 except Exit as exc: receipt['outcome'] = exc.reason code = exc.code except Exception: receipt['outcome'] = 'contained_internal_error' code = 1 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())