tegwick
a29d92ecda
tools/bounded-execution-readiness/ready.py (make bounded-execution-ready): agent-identity checks from owner source (admission, approver chain at the Argo pin and main, access-engine profile, CCRs by the owner's RUNBOOK_ALLOWED_STATUSES, case and broker type pin), and three named read-only founder checks. executor-lanes.py status: write role, fence, cas_required; no write. A replay test catches 2026-11-04. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 943911@bnt-lap001 Assistant-Session: fc322940-1c8e-4bed-ab39-86cec99f23a0
234 lines
12 KiB
Python
234 lines
12 KiB
Python
#!/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 <evidence>/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=<run 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())
|