#!/usr/bin/env python3 """Readiness check before a bounded-execution window opens (NK-WP-0058-T03). Read-only, value-free. python3 tools/bounded-execution-readiness/ready.py --admission admissions/BEA-....yaml --approval-id ID \ --receipt EVIDENCE/NN-readiness.json [--siblings DIR] [--now ISO] What the agent identity can check, from the owners' committed source at the revisions that matter: admission valid (the admission validator), executor named, window not closed approver_chain the admission's approvers resolve through the rapp config; informed-decision, both at the revision Argo pins and on main (live is one of the two), registers this executor and routes its review group to the first approver's subject (access-engine lists a memo only to that recipient, policy v14 recipient_ok) argo_pin the pinned informed-decision files equal its main branch (warning only: live may be newer) profile access-engine's present_existing profile for the executor matches informed-decision's registry ccrs CCR-2026-0058..0061 and -0063 are in railiance-platform's RUNBOOK_ALLOWED_STATUSES (approved, applied, verified, active), every bound_claims_confirmed true case the approval-client case is materialized or live; its type's bytes match the broker's hash pin What it cannot see is named as one read-only founder command each (`founder_checks`): the write role and fence (OpenBao), the preflight Job (reachability, lanes, source pins in the pinned image, trust, window, approval) and the custody deliveries (ExternalSecret Ready). Exit 0 when every agent check passes, 1 otherwise. """ import argparse import datetime as dt import hashlib import importlib.util import json import os import re import subprocess from pathlib import Path import yaml REPO = Path(__file__).resolve().parents[2] CASE = 'security-cases/nk-bind-executor-approval.yaml' CCRS = ('CCR-2026-0058', 'CCR-2026-0059', 'CCR-2026-0060', 'CCR-2026-0061', 'CCR-2026-0063') ARGO_APP = 'argocd/railiance01/applications/informed-decision.yaml' POLICY = 'examples/informed-decision-present-existing-review/policy.md' BROKER = 'scripts/keycape_client_secret_generate.py' RAPP_CONFIG = 'examples/config.pilot.json' def load_validator(): spec = importlib.util.spec_from_file_location('bea_validate_ready', REPO / 'tools/bounded-execution-admission/validate.py') module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def when(text): return dt.datetime.fromisoformat(text.replace('Z', '+00:00')) class Report: def __init__(self): self.checks = [] def add(self, check, ok, reason=None, warn=False): result = 'pass' if ok else ('warn' if warn else 'fail') self.checks.append({'check': check, 'result': result, **({'reason': reason} if reason and not ok else {})}) return ok def git_show(repo, rev, path): done = subprocess.run(['git', '-C', str(repo), 'show', f'{rev}:{path}'], capture_output=True) if done.returncode: raise FileNotFoundError(f'{repo.name}:{rev}:{path}') return done.stdout def check_admission(report, doc, now): bea = load_validator() result = bea.evaluate(doc, json.loads(bea.SCHEMA.read_text())) rules = ','.join(sorted({f['rule'] for f in result['findings']})) report.add('admission_valid', result['valid'], 'admission_invalid:' + rules) spec = doc['spec'] report.add('window_not_closed', now < when(spec['window']['not_after']), 'window_closed') def check_approver_chain(report, spec, cfg, infd, rev, label='pinned'): names = spec['approval']['approvers'] subjects = [cfg.get('approver_subjects', {}).get(n) for n in names] if label == 'pinned': if not report.add('approvers_resolve', all(subjects), 'approver_not_in_rapp_config'): return None report.add('single_approver', len(names) == 1, 'routing_has_one_recipient_per_group') elif not all(subjects): return None groups = json.loads(git_show(infd, rev, 'deploy/lifecycle/review-groups.json')) submitters = json.loads(git_show(infd, rev, 'deploy/lifecycle/intake-submitters.json')) row = submitters.get(cfg['runtime_principal']) if not report.add(f'executor_registered:{label}', row is not None, 'runtime_principal_not_in_intake_submitters'): return None actor = (row.get('present_existing') or {}).get('approval_actor') report.add(f'approval_actor_matches:{label}', actor == cfg['executor_identity'], 'approval_actor_differs') report.add(f'memo_class_admitted:{label}', 'bounded-execution' in row.get('memo_classes', []), 'memo_class_missing') recipients = {g: (groups.get(g) or {}).get('subject') for g in row.get('review_groups', [])} report.add(f'recipient_is_approver:{label}', bool(recipients) and set(recipients.values()) == {subjects[0]}, 'review_group_routes_elsewhere') return row def check_argo_pin(report, infd, rev): for name in ('review-groups.json', 'intake-submitters.json'): same = git_show(infd, rev, 'deploy/lifecycle/' + name) == git_show(infd, 'HEAD', 'deploy/lifecycle/' + name) report.add('argo_pin_matches_source:' + name, same, 'pinned_revision_differs_from_main', warn=True) def check_profile(report, row, cfg, access_engine): text = (access_engine / POLICY).read_text() block = re.search(r'present_existing_profiles := \{(.*?)\n\}', text, re.S) entry = block and re.search(r'"' + re.escape(cfg['executor_identity']) + r'": \{(.*?)\}', block.group(1), re.S) if not report.add('profile_present', bool(entry), 'executor_not_in_present_existing_profiles'): return fields = dict(re.findall(r'"(\w+)": "([^"]*)"', entry.group(1))) report.add('profile_submitter_matches', row is not None and fields.get('intake_submitter') == row['submitter_id'], 'intake_submitter_differs') report.add('profile_memo_class_matches', fields.get('memo_class') == 'bounded-execution', 'memo_class_differs') def confirmations(node): if isinstance(node, dict): for key, value in node.items(): if key == 'bound_claims_confirmed': yield value else: yield from confirmations(value) elif isinstance(node, list): for value in node: yield from confirmations(value) def ccr_statuses(platform): """The owner's own set of statuses under which a CCR may be acted on (credential-change.py), not a copy.""" spec = importlib.util.spec_from_file_location('credential_change_ready', platform / 'scripts/credential-change.py') module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return set(module.RUNBOOK_ALLOWED_STATUSES) def check_ccrs(report, platform): folder = platform / 'credential-change-requests' allowed = ccr_statuses(platform) for ccr in CCRS: files = sorted(folder.glob(ccr + '-*.yaml')) if not report.add(ccr + ':present', len(files) == 1, 'ccr_file_missing'): continue doc = yaml.safe_load(files[0].read_text()) report.add(ccr + ':approved', doc.get('status') in allowed, 'status_' + str(doc.get('status'))) report.add(ccr + ':bindings_confirmed', all(v is True for v in confirmations(doc)), 'binding_not_confirmed') def check_case(report, platform): case = yaml.safe_load((REPO / CASE).read_text())['spec'] report.add('case_materialized', case.get('status') in ('materialized', 'live'), 'case_' + str(case.get('status'))) pins = {ref: (path, digest) for ref, path, digest in re.findall( r"'([a-z0-9-]+@[0-9.]+)': \(\s*'(security-case-types/[^']+)',\s*'([0-9a-f]{64})'\)", (platform / BROKER).read_text())} pinned = pins.get(case.get('type')) if not report.add('case_type_acknowledged', pinned is not None, 'type_not_admitted_by_broker'): return path, digest = pinned report.add('case_type_bytes_match_broker', hashlib.sha256((REPO / path).read_bytes()).hexdigest() == digest, 'type_yaml_changed_since_acknowledgment') def founder_checks(admission, approval_id): """One read-only command each, for what the agent identity cannot read (NK-WP-0058-T05 status table).""" k = 'kubectl --kubeconfig ~/.kube/config-railiance01 -n sso' jsonpath = "-o jsonpath='{.status.conditions[?(@.type==\"Ready\")].status}'" return [ {'id': 'write_role_and_fence', 'needs': 'platform-admin OpenBao login', 'command': 'python3 tools/service-binds/executor-lanes.py status --receipt /NN-status.json', 'expect': 'ready_for_window: true'}, {'id': 'preflight_job', 'needs': 'admin kubeconfig', 'command': f'cd ~/rapp-bounded-executor && make preflight ADMISSION={admission} APPROVAL_ID={approval_id} ' 'CONFIG= KUBECTL="kubectl --kubeconfig ~/.kube/config-railiance01"', 'expect': '"ready": true'}, {'id': 'custody_delivery', 'needs': 'admin kubeconfig', 'command': f'for e in keycape-nk-bind-executor-approval-client bounded-executor-tokens; do ' f'{k} get externalsecret $e {jsonpath}; echo; done', 'expect': 'True, twice'}, ] def run(admission_path, approval_id, siblings, now): report = Report() doc = yaml.safe_load(Path(admission_path).read_text()) spec = doc['spec'] check_admission(report, doc, now) platform, infd = siblings / 'railiance-platform', siblings / 'informed-decision' access_engine = siblings / 'access-engine' cfg = json.loads((siblings / 'rapp-bounded-executor' / RAPP_CONFIG).read_text()) report.add('executor_named', spec['executor']['identity'] == cfg['executor_identity'], 'admission_names_another') rev = yaml.safe_load((platform / ARGO_APP).read_text())['spec']['source']['targetRevision'] row = check_approver_chain(report, spec, cfg, infd, rev) check_argo_pin(report, infd, rev) # live is the Argo-pinned revision or a newer founder-run release (2026-11-04: the pin lagged): hold both check_approver_chain(report, spec, cfg, infd, 'HEAD', label='main') check_profile(report, row, cfg, access_engine) check_ccrs(report, platform) check_case(report, platform) failures = [c['check'] for c in report.checks if c['result'] == 'fail'] return {'action': 'bounded-execution-readiness', 'admission': spec['id'], 'values_recorded': False, 'at': now.strftime('%Y-%m-%dT%H:%M:%SZ'), 'argo_pin': rev[:12], 'agent_checks_pass': not failures, 'failures': failures, 'warnings': [c['check'] for c in report.checks if c['result'] == 'warn'], 'checks': report.checks, 'founder_checks': founder_checks(admission_path, approval_id)} def main(argv=None): p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) p.add_argument('--admission', required=True) p.add_argument('--approval-id', required=True) p.add_argument('--receipt', required=True) p.add_argument('--siblings', type=Path, default=Path(os.environ.get('NK_SIBLINGS', REPO.parent))) p.add_argument('--now', type=when, default=None) a = p.parse_args(argv) now = a.now or dt.datetime.now(dt.timezone.utc) fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) try: r = run(a.admission, a.approval_id, a.siblings, now) except Exception as error: # value-free: a missing file or revision is named, nothing else is echoed r = {'action': 'bounded-execution-readiness', 'agent_checks_pass': False, 'values_recorded': False, 'failures': ['error:' + type(error).__name__], 'reason': str(error) if isinstance(error, FileNotFoundError) else type(error).__name__} with os.fdopen(fd, 'w') as out: json.dump(r, out, indent=2, sort_keys=True); out.write('\n') print(json.dumps({'agent_checks_pass': r['agent_checks_pass'], 'failures': r['failures'], 'warnings': r.get('warnings', []), 'founder_checks': [f['id'] for f in r.get('founder_checks', [])]})) return 0 if r['agent_checks_pass'] else 1 if __name__ == '__main__': raise SystemExit(main())