railiance-platform/scripts/kv_duplicate_retire.py

#!/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 or metadata path (exact, ``+`` and trailing ``*`` globs).

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')

    paths = ['platform/data/' + DUPLICATE, 'platform/metadata/' + DUPLICATE]
    granting = {}
    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')
        hits = granting_stanzas(result.stdout.decode(), paths)
        if hits:
            granting[name] = hits
    receipt['granting_policies'] = granting
    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())