"""Read-only, silent owner session for the four open custody questions. Runs inside Warden's attended login envelope. Reads no secret value into the receipt: where a data read is unavoidable (field-name resolution), only sorted key names leave this process. Applies nothing, writes nothing to OpenBao, and records no approval. Warden owns the temporary token helper and self-revokes after this command returns. Questions settled: Q1 Does legacy mount/path secret/coulomb/whynot-design/npm/publish exist? Q2 Which field name backs the authoritative npm lane, NPM_AUTH_TOKEN or npm_token? (secrets-engine message 546403e4) Q3 Do the two KeyCape approval policies live in OpenBao and match the repo source? (activation receipt recorded policy_applied false) Q4 Which netkingdom OIDC roles and bound group claims already exist, as non-binding input to CCR-2026-0019? Q5 Governed backup lane field presence, as RPF-WP-0029-T02 / RISK-F-0010 context. Never the value, fingerprint, length or shape. """ import argparse import hashlib import json import os import re import subprocess from datetime import datetime, timezone from pathlib import Path REPO = Path(__file__).resolve().parent.parent LEGACY_MOUNT = 'secret/' LEGACY_PATH = 'secret/coulomb/whynot-design/npm/publish' NPM_PATH = 'platform/workloads/coulomb/whynot-design/npm-publish' BACKUP_PATH = 'platform/workloads/railiance/backup/offsite-lane' POLICIES = { 'workload-kv-read-keycape-secrets-engine-approval': 'openbao/policies/workload-kv-read-keycape-secrets-engine-approval.hcl', 'workload-kv-read-keycape-approval-engine-operator': 'openbao/policies/workload-kv-read-keycape-approval-engine-operator.hcl', } FIELD_NAME_RE = re.compile(r'^[A-Za-z0-9_.-]{1,64}$') def bao(*args, timeout=20): """Run one bao command. Returns (returncode, parsed-json-or-None).""" result = subprocess.run(['bao', *args], capture_output=True, text=True, timeout=timeout) if result.returncode: return result.returncode, None try: return 0, json.loads(result.stdout) except json.JSONDecodeError: return 0, None def field_names(payload): """Sorted key names only. A value never leaves this function.""" data = ((payload or {}).get('data') or {}).get('data') or {} names = sorted(str(k) for k in data.keys()) if any(not FIELD_NAME_RE.match(n) for n in names): raise ValueError('unexpected field-name shape; refusing to record') return names def kv_version(payload): meta = ((payload or {}).get('data') or {}).get('metadata') or {} return meta.get('version') def normalized_policy(text): lines = [line.strip() for line in text.splitlines()] return '\n'.join(line for line in lines if line) def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--receipt', type=Path, required=True) args = parser.parse_args() receipt = { 'schema': 'platform.openbao-open-questions-session.v1', 'observed_at': datetime.now(timezone.utc).isoformat(), 'operation': 'read-only observation', 'credential_values_emitted': False, 'openbao_mutations': 0, 'questions': ['Q1', 'Q2', 'Q3', 'Q4', 'Q5'], } # Q1 - legacy mount and path existence. rc, mounts = bao('secrets', 'list', '-format=json') legacy_mount_present = bool(mounts and LEGACY_MOUNT in mounts) q1 = {'legacy_mount_present': legacy_mount_present, 'legacy_path': LEGACY_PATH} if legacy_mount_present: rc, meta = bao('kv', 'metadata', 'get', '-format=json', LEGACY_PATH) q1['legacy_path_present'] = rc == 0 if rc == 0 and meta: data = meta.get('data') or {} q1['current_version'] = data.get('current_version') q1['created_time'] = data.get('created_time') q1['updated_time'] = data.get('updated_time') else: q1['legacy_path_present'] = False receipt['q1_legacy_npm_path'] = q1 # Q2 - authoritative npm lane field names. rc, payload = bao('kv', 'get', '-format=json', NPM_PATH) receipt['q2_npm_lane'] = { 'path': NPM_PATH, 'readable': rc == 0, 'field_names': field_names(payload) if rc == 0 else None, 'kv_version': kv_version(payload) if rc == 0 else None, 'values_recorded': False, } # Q3 - KeyCape approval policy presence and drift. q3 = {} for name, rel in POLICIES.items(): rc, payload = bao('policy', 'read', '-format=json', name) entry = {'present': rc == 0} if rc == 0: live = normalized_policy(((payload or {}).get('data') or {}).get('policy', '')) source = normalized_policy((REPO / rel).read_text()) entry['matches_repo_source'] = live == source entry['live_sha256'] = hashlib.sha256(live.encode()).hexdigest() entry['source_sha256'] = hashlib.sha256(source.encode()).hexdigest() q3[name] = entry receipt['q3_keycape_policies'] = q3 # Q4 - existing netkingdom OIDC roles and bound claims (non-secret config). rc, listing = bao('list', '-format=json', 'auth/netkingdom/role') roles = {} if rc == 0 and isinstance(listing, list): for role in listing: rc, payload = bao('read', '-format=json', f'auth/netkingdom/role/{role}') if rc: continue data = (payload or {}).get('data') or {} roles[str(role)] = { 'bound_claims': data.get('bound_claims'), 'groups_claim': data.get('groups_claim'), 'user_claim': data.get('user_claim'), 'token_policies': data.get('token_policies'), 'token_ttl': data.get('token_ttl'), } receipt['q4_netkingdom_roles'] = { 'listed': rc == 0, 'roles': roles, 'note': 'input only; the authorized operator group claim is confirmed by ' 'NetKingdom/KeyCape, not inferred from this listing', } # Q5 - governed backup lane field presence. rc, payload = bao('kv', 'get', '-format=json', BACKUP_PATH) receipt['q5_backup_lane'] = { 'path': BACKUP_PATH, 'readable': rc == 0, 'field_names': field_names(payload) if rc == 0 else None, 'kv_version': kv_version(payload) if rc == 0 else None, 'predecessor_material_recorded': False, } fd = os.open(args.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, 'w') as stream: json.dump(receipt, stream, indent=2) stream.write('\n') if __name__ == '__main__': main()