From 07b6b63fe655cad6a020741014f94becaa71f3a1 Mon Sep 17 00:00:00 2001 From: codex Date: Thu, 10 Sep 2026 08:06:30 +0200 Subject: [PATCH] Let the session run the npm questions alone, and refuse an ambient token --questions selects a subset, so the npm field can be settled with Q1,Q2 without the two data reads the other questions imply; Q5 is the only step touching the backup lane and is now excludable by name. The runner also refuses to start when OPENBAO_TOKEN, BAO_TOKEN or VAULT_TOKEN is set. A standing token would let every read succeed without the attended login and produce a receipt that looks attended and is not. --allow-ambient-token overrides and records attended_identity false rather than claiming provenance it does not have. Co-Authored-By: Claude Opus 5 Claude-Session: https://claude.ai/code/session_01WLUjpv3ssxNRAEPPgLFnEB Assistant: claude-code Assistant-Model: opus Assistant-Process: 1275505@bnt-lap001 Assistant-Session: 97265baa-f08f-4032-b290-a1e2965a69c5 --- docs/openbao-open-questions-session.md | 19 +++ scripts/openbao_open_questions_session.py | 170 +++++++++++++--------- 2 files changed, 120 insertions(+), 69 deletions(-) diff --git a/docs/openbao-open-questions-session.md b/docs/openbao-open-questions-session.md index 23a2153..caf0f7f 100644 --- a/docs/openbao-open-questions-session.md +++ b/docs/openbao-open-questions-session.md @@ -39,6 +39,25 @@ the writer opens it `O_EXCL` at mode 0600 and refuses to overwrite. The runner prints nothing. Its entire output is the receipt. +**To settle the npm field alone**, run only the two questions that bear on it: + +```sh +scripts/openbao-attended-exec.py -- \ + /usr/bin/python3 /home/worsch/railiance-platform/scripts/openbao_open_questions_session.py \ + --questions Q1,Q2 --receipt /tmp/.json +``` + +`--questions` defaults to all five. Q5 is the only step that reads the backup +lane, so naming a subset is also how to exclude it. + +**Ambient-token guard.** The runner refuses to start if `OPENBAO_TOKEN`, +`BAO_TOKEN` or `VAULT_TOKEN` is set in the environment. A standing token would +let these reads succeed without the attended login, producing a receipt that +looks attended and is not. Unset it first; the envelope supplies the identity. +`--allow-ambient-token` overrides, and then the receipt records +`attended_identity: false` — which is the honest label for that read, not a +formality. + ## What the runner reads, and the one honest caveat | Step | Call | Emitted | diff --git a/scripts/openbao_open_questions_session.py b/scripts/openbao_open_questions_session.py index 5b3f06d..b8ea8df 100755 --- a/scripts/openbao_open_questions_session.py +++ b/scripts/openbao_open_questions_session.py @@ -38,6 +38,7 @@ POLICIES = { '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): @@ -73,7 +74,31 @@ def normalized_policy(text): 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', @@ -81,81 +106,88 @@ def main(): 'operation': 'read-only observation', 'credential_values_emitted': False, 'openbao_mutations': 0, - 'questions': ['Q1', 'Q2', 'Q3', 'Q4', 'Q5'], + 'questions': sorted(selected), + 'attended_identity': not ambient, + 'ambient_token_vars_present': ambient, } - # 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 '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 - # 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 '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, + } - # 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 '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 - # 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 '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', + } - # 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, - } + 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: