"""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}$') AMBIENT_TOKEN_VARS = ('OPENBAO_TOKEN', 'BAO_TOKEN', 'VAULT_TOKEN') 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) parser.add_argument( '--allow-ambient-token', action='store_true', help='Proceed despite a standing token in the environment. The receipt ' 'then records attended_identity false.') parser.add_argument( '--questions', default='Q1,Q2,Q3,Q4,Q5', help='Comma-separated subset to run. Q1,Q2 settle the npm field alone; ' 'Q5 is the only step that reads the backup lane.') args = parser.parse_args() ambient = sorted(v for v in AMBIENT_TOKEN_VARS if os.environ.get(v)) if ambient and not args.allow_ambient_token: # A standing token in the environment would let these reads succeed # without the attended login, producing a receipt that looks attended # and is not. Refuse rather than silently record the wrong provenance. raise SystemExit( 'ambient OpenBao token present in ' + ', '.join(ambient) + '; unset it so the attended envelope supplies the identity, or pass ' '--allow-ambient-token to record the read as unattended') selected = {q.strip().upper() for q in args.questions.split(',') if q.strip()} unknown = selected - {'Q1', 'Q2', 'Q3', 'Q4', 'Q5'} if unknown: raise SystemExit('unknown question(s): ' + ', '.join(sorted(unknown))) if not selected: raise SystemExit('at least one question is required') 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': sorted(selected), 'attended_identity': not ambient, 'ambient_token_vars_present': ambient, } if 'Q1' in selected: # 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 if 'Q2' in selected: # 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, } if 'Q3' in selected: # 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 if 'Q4' in selected: # 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', } if 'Q5' in selected: # 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()