From 634b3a41dfa0efee09134dde498b545a9df29a23 Mon Sep 17 00:00:00 2001 From: tegwick Date: Sat, 31 Oct 2026 19:15:46 +0100 Subject: [PATCH] NK-WP-0056-T10: attended OpenBao setup and proof for the executor lanes tools/service-binds/executor-lanes.py (bind-lane.py shape): under both sha256-bound, approved, binding-confirmed CCRs (0058 write, 0059 journey) and an attended platform-admin, apply both policies, set cas_required on the svc-authelia key, create the standing journey-read role, refuse if the window-scoped write role exists, prove the agent fence; verify proves the journey token reads only its path (<=15m) and denies wrong SA, wrong audience and write-role login. Value-free receipts. 12 tests on a fake OpenBao (real helpers), in make bounded-executor-test. Founder runbook: sso-mfa/k8s/bounded-executor/T10-LANES-RUNBOOK.md. Co-Authored-By: Claude Opus 5.5 Assistant: claude-code Assistant-Model: opus Assistant-Process: 5502@bnt-lap001 Assistant-Session: 68b3b02f-5d0f-4e71-8421-796f62e50196 --- Makefile | 2 +- .../k8s/bounded-executor/T10-LANES-RUNBOOK.md | 73 ++++++ tests/test_executor_lanes.py | 163 ++++++++++++ tools/service-binds/executor-lanes.py | 242 ++++++++++++++++++ 4 files changed, 479 insertions(+), 1 deletion(-) create mode 100644 sso-mfa/k8s/bounded-executor/T10-LANES-RUNBOOK.md create mode 100644 tests/test_executor_lanes.py create mode 100644 tools/service-binds/executor-lanes.py diff --git a/Makefile b/Makefile index 87f4ed4..7fdeab5 100644 --- a/Makefile +++ b/Makefile @@ -210,7 +210,7 @@ bounded-execution-admission-check: ## Validate the example admissions (read-only python3 tools/bounded-execution-admission/validate.py validate examples/bounded-execution-admission/*.yaml bounded-executor-test: ## Bounded executor RBAC and declaration tests; the pilot admission inside its declared grants (NK-WP-0056-T03) - python3 -m pytest sso-mfa/k8s/bounded-executor bounded-executor/tests + python3 -m pytest sso-mfa/k8s/bounded-executor bounded-executor/tests tests/test_executor_lanes.py python3 tools/service-identity/validate.py tools/service-identity/declarations/bounded-executor.yaml \ --admissions examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml diff --git a/sso-mfa/k8s/bounded-executor/T10-LANES-RUNBOOK.md b/sso-mfa/k8s/bounded-executor/T10-LANES-RUNBOOK.md new file mode 100644 index 0000000..2c18a47 --- /dev/null +++ b/sso-mfa/k8s/bounded-executor/T10-LANES-RUNBOOK.md @@ -0,0 +1,73 @@ +# T10 runbook: executor standing objects and OpenBao lanes (founder-run) + +NK-WP-0056-T10. Three steps, about 15 minutes. Nothing here starts a Job or touches the svc-authelia value. The +bind write role (CCR-2026-0058) is **not** created: it exists only inside an admitted window. + +| Step | Where | What changes | +| --- | --- | --- | +| 1 | `~/rapp-bounded-executor` | creates ServiceAccount `nk-bind-executor`, its Role and RoleBinding, and NetworkPolicy `bounded-executor` in `sso` | +| 2 | `~/railiance-platform` | records `binding_confirmed` in CCR-2026-0058 and CCR-2026-0059 | +| 3 | `~/net-kingdom` | OpenBao: both policies, `cas_required` on the svc-authelia key, the standing journey-read role, and proof | + +## 1. Standing objects (rapp) + +```bash +cd ~/rapp-bounded-executor && git pull --ff-only +K="kubectl --kubeconfig $HOME/.kube/config-railiance01" +make server-dry-run KUBECTL="$K" # 4 objects "serverside-applied (server dry run)" +make deploy KUBECTL="$K" # the same 4 objects, without "(server dry run)" +make verify-live KUBECTL="$K" # {"role_matches_ceiling": true, "standing_objects_present": true, ...} +``` + +The pod runs nothing yet: no Job exists until `make run` under an admission. + +## 2. Confirm the binding (railiance-platform) + +The CCRs bind `service_account_names: [nk-bind-executor]` in `sso`, audience `openbao`. Confirm that this +ServiceAccount now exists, as declared: + +```bash +cd ~/railiance-platform && git pull --ff-only +UID_SA=$($K -n sso get serviceaccount nk-bind-executor -o jsonpath='{.metadata.uid}'); echo "$UID_SA" +for C in CCR-2026-0058 CCR-2026-0059; do + scripts/credential-change.py confirm-binding "$C" --reviewer platform-owner --comment \ + "Binding confirmed: ServiceAccount nk-bind-executor in namespace sso only (uid $UID_SA), audience openbao, from rapp-bounded-executor." +done +git status --short # exactly the two CCR files modified +``` + +## 3. OpenBao lanes (attended platform-admin) + +```bash +cd ~/net-kingdom && git pull --ff-only +bao login -method=oidc -path=netkingdom role=platform-admin # MFA; the tool fails closed without it +W=$(sha256sum ~/railiance-platform/credential-change-requests/CCR-2026-0058-*.yaml | cut -d' ' -f1) +J=$(sha256sum ~/railiance-platform/credential-change-requests/CCR-2026-0059-*.yaml | cut -d' ' -f1) +E=docs/evidence/$(date +%F)-nk56-t10-executor-lanes; mkdir -p "$E" + +python3 tools/service-binds/executor-lanes.py setup --write-ccr-sha256 "$W" --journey-ccr-sha256 "$J" \ + --receipt "$E/01-setup.json" # {"status": "lanes_prepared", "phase": "done"} +python3 tools/service-binds/executor-lanes.py verify --write-ccr-sha256 "$W" --journey-ccr-sha256 "$J" \ + --kubeconfig "$HOME/.kube/config-railiance01" --receipt "$E/02-verify.json" # {"status": "lanes_verified", ...} +``` + +`setup` is idempotent: it creates what is missing and refuses on any drift. It never changes the coding-agent fence, +because CCR-2026-0053 and CCR-2026-0057 established it; it only proves the fence. Receipts hold no values. `verify` +reads the journey entry into memory only to check its field names. + +If a step stops, the receipt's `phase` and `reason` say where: + +| `reason` | Meaning, next step | +| --- | --- | +| `write:ccr_digest` / `journey:ccr_digest` | the CCR changed after `W`/`J` were computed; compute them again | +| `*:binding_unconfirmed` | step 2 not done (or not saved) | +| `attended_platform_admin_required` | no `bao login` as platform-admin, or a root token | +| `agent_fence_missing` | the boundary policy lacks a deny line for one of the two paths; stop and hand to claude-code | +| `command_failed` in phase `cas_required` | platform-admin may not patch KV metadata; hand to claude-code | +| `write_role_exists_outside_a_window` | someone created the window-scoped write role; delete it, or stop | +| `*:policy_drift` | a policy of that name exists with other text; stop | +| `negative_login_accepted` / `journey_capabilities` | the role is wider than approved; delete the journey role and stop | + +## 4. Afterwards + +Tell claude-code. It commits the two CCRs and the receipts, moves CCR-2026-0059 on, and records T10 progress. diff --git a/tests/test_executor_lanes.py b/tests/test_executor_lanes.py new file mode 100644 index 0000000..681b31a --- /dev/null +++ b/tests/test_executor_lanes.py @@ -0,0 +1,163 @@ +"""executor-lanes.py against an in-memory OpenBao and Kubernetes (NK-WP-0056-T10); no values leave the fakes.""" +import hashlib +import importlib.util +import json +import shutil +import subprocess +import sys +from pathlib import Path +from types import SimpleNamespace + +import pytest + +ROOT = Path(__file__).resolve().parents[1] +PLATFORM = ROOT.parent / 'railiance-platform' +CCRS = {'write': 'CCR-2026-0058-net-kingdom-bind-executor-svc-authelia-write.yaml', + 'journey': 'CCR-2026-0059-net-kingdom-bind-executor-syn-journey-read.yaml'} +JOURNEY = 'platform/data/workloads/net-kingdom/synthetic-journey/syn-journey' +BIND = 'platform/data/workloads/net-kingdom/directory-binds/svc-authelia' +FENCE = ''.join(f'path "{p}" {{ capabilities = ["deny"] }}\n' for p in ( + JOURNEY, JOURNEY.replace('/data/', '/metadata/'), BIND, BIND.replace('/data/', '/metadata/'))) + +pytestmark = pytest.mark.skipif(not all((PLATFORM / 'credential-change-requests' / c).exists() for c in CCRS.values()), + reason='railiance-platform checkout with CCR-2026-0058/0059 required') + + +def platform_copy(tmp_path, confirm=True): + root = tmp_path / 'railiance-platform' + shutil.copytree(PLATFORM / 'scripts', root / 'scripts', ignore=shutil.ignore_patterns('__pycache__')) + (root / 'openbao').mkdir(); shutil.copytree(PLATFORM / 'openbao' / 'policies', root / 'openbao' / 'policies') + (root / 'credential-change-requests').mkdir() + for name in CCRS.values(): + shutil.copy(PLATFORM / 'credential-change-requests' / name, root / 'credential-change-requests' / name) + if confirm: + subprocess.run([sys.executable, str(root / 'scripts/credential-change.py'), 'confirm-binding', + '--reviewer', 'platform-owner', '--comment', 'synthetic test', + str(root / 'credential-change-requests' / name)], check=True, capture_output=True) + return root + + +def digests(root): + return {k: hashlib.sha256((root / 'credential-change-requests' / v).read_bytes()).hexdigest() for k, v in CCRS.items()} + + +class FakeBao: + def __init__(self, m): + self.m = m + self.store = {m.BOUNDARY: {'policy': 'path "x" { capabilities = ["deny"] }\n' + FENCE}} + self.cas = False + self.identity = {'policies': ['platform-admin'], 'path': 'auth/netkingdom/oidc/callback'} + self.login_ok = {('nk-bind-executor', 'openbao', m.LANES['journey']['role'])} + self.wide = False + self.revoked = [] + + def __call__(self, args, *, payload=None, token=None, allow_failure=False): + out = lambda d, code=0, err=b'': SimpleNamespace(returncode=code, stdout=json.dumps(d).encode(), stderr=err) + missing = out({}, 2, b'No value found') + if args[:2] == ['token', 'lookup']: return out({'data': self.identity}) + if args[:2] == ['token', 'create']: return out({'auth': {'client_token': 'child'}}) + if args[:3] == ['kv', 'metadata', 'get']: return out({'data': {'cas_required': self.cas}}) + if args[:3] == ['kv', 'metadata', 'patch']: + assert '-cas-required=true' in args; self.cas = True; return out({}) + if args[0] == 'read': + path = args[-1] + if path == JOURNEY and token == 'reader': + return out({'data': {'data': {'PASSWORD': 'SYNTHETIC-PW-VALUE', 'TOTP_SEED': 'SYNTHETIC-SEED-VALUE'}, 'metadata': {'version': 1}}}) + return out({'data': self.store[path]}) if path in self.store else missing + if args[0] == 'write': + path = args[-2] if args[-1] == '-' else args[-1] + if path == 'auth/token/revoke': self.revoked.append(payload['token']); return out({}) + if path == 'sys/capabilities': + def cap(p): + if payload['token'] == 'reader' and (p == JOURNEY or self.wide): return ['read'] + return ['deny'] + return out({'data': {p: cap(p) for p in payload['paths']}}) + if path == 'auth/kubernetes/login': + sa, aud = payload['jwt'].split('|') + if (sa, aud, payload['role']) in self.login_ok: + return out({'auth': {'client_token': 'reader', 'lease_duration': 900, + 'token_policies': [self.m.LANES['journey']['policy']]}}) + return out({}, 2, b'Code: 403') + if path.startswith('auth/kubernetes/role/'): + self.store[path] = {k: (900 if v == '15m' else v) for k, v in payload.items()}; return out({}) + self.store[path] = payload; return out({}) + raise AssertionError(args) + + +def build(tmp_path, monkeypatch, confirm=True): + root = platform_copy(tmp_path, confirm) + monkeypatch.setenv('NK_PLATFORM_ROOT', str(root)) + spec = importlib.util.spec_from_file_location('executor_lanes', ROOT / 'tools/service-binds/executor-lanes.py') + m = importlib.util.module_from_spec(spec); spec.loader.exec_module(m) + fake = FakeBao(m) + for module in (m, sys.modules['state_hub_preflight_lane'], sys.modules['telemetry_smtp_delivery']): + if hasattr(module, 'bao'): + monkeypatch.setattr(module, 'bao', fake) # the real helpers (ensure, capabilities, revoke) run on the fake + monkeypatch.setattr(m, 'kube', lambda kc, args: f'{args[4]}|{args[5].split("=", 1)[1]}'.encode()) + return m, fake, digests(root) + + +def run(m, action, d): + r = {} + try: + (m.setup(d, r) if action == 'setup' else m.verify(d, 'kc', r)) + except m.Refused as error: + r.update(status='failed', reason=str(error)) + return r + + +def test_setup_applies_policies_cas_and_journey_role_but_never_the_write_role(tmp_path, monkeypatch): + m, fake, d = build(tmp_path, monkeypatch) + r = run(m, 'setup', d) + assert r['status'] == 'lanes_prepared', r + for lane in m.LANES.values(): + assert fake.store['sys/policies/acl/' + lane['policy']]['policy'] == lane['policy_text'] + assert fake.cas is True + assert 'auth/kubernetes/role/' + m.LANES['journey']['role'] in fake.store + assert 'auth/kubernetes/role/' + m.LANES['write']['role'] not in fake.store + assert run(m, 'setup', d)['status'] == 'lanes_prepared' # idempotent + + +def test_verify_after_setup(tmp_path, monkeypatch): + m, fake, d = build(tmp_path, monkeypatch) + run(m, 'setup', d) + r = run(m, 'verify', d) + assert r['status'] == 'lanes_verified', r + assert r['write_login_denied'] and r['wrong_audience_denied'] and 'reader' in fake.revoked + assert 'SYNTHETIC-' not in json.dumps(r) # no values in the receipt + + +@pytest.mark.parametrize('mutate, reason', [ + (lambda m, f, d: d.update(write='0' * 64), 'write:ccr_digest'), + (lambda m, f, d: f.store.update({m.BOUNDARY: {'policy': 'path "x" { capabilities = ["deny"] }\n'}}), + 'agent_fence_missing'), + (lambda m, f, d: f.identity.update(policies=['root']), 'attended_platform_admin_required'), + (lambda m, f, d: f.store.update({'auth/kubernetes/role/' + m.LANES['write']['role']: {}}), + 'write_role_exists_outside_a_window'), + (lambda m, f, d: f.store.update({'sys/policies/acl/' + m.LANES['journey']['policy']: {'policy': 'path "*" {}'}}), + 'net-kingdom-bind-executor-syn-journey-read:policy_drift'), +]) +def test_setup_refusals(tmp_path, monkeypatch, mutate, reason): + m, fake, d = build(tmp_path, monkeypatch) + mutate(m, fake, d) + r = run(m, 'setup', d) + assert (r['status'], r['reason']) == ('failed', reason) + + +def test_unconfirmed_binding_is_refused(tmp_path, monkeypatch): + m, fake, d = build(tmp_path, monkeypatch, confirm=False) + assert run(m, 'setup', d)['reason'] == 'write:binding_unconfirmed' + + +@pytest.mark.parametrize('mutate, reason', [ + (lambda m, f: f.login_ok.add(('nk-bind-executor', 'openbao', m.LANES['write']['role'])), 'negative_login_accepted'), + (lambda m, f: f.login_ok.add(('default', 'openbao', m.LANES['journey']['role'])), 'negative_login_accepted'), + (lambda m, f: setattr(f, 'wide', True), 'journey_capabilities'), + (lambda m, f: setattr(f, 'cas', False), 'cas_required_not_set'), +]) +def test_verify_refusals(tmp_path, monkeypatch, mutate, reason): + m, fake, d = build(tmp_path, monkeypatch) + run(m, 'setup', d) + mutate(m, fake) + r = run(m, 'verify', d) + assert (r['status'], r['reason']) == ('failed', reason) diff --git a/tools/service-binds/executor-lanes.py b/tools/service-binds/executor-lanes.py new file mode 100644 index 0000000..1b73640 --- /dev/null +++ b/tools/service-binds/executor-lanes.py @@ -0,0 +1,242 @@ +#!/usr/bin/env python3 +"""Attended OpenBao setup and proof for the bounded executor's two lanes (NK-WP-0056-T10, T06). + + write CCR-2026-0058 policy only. Its Kubernetes role is window-scoped: created per admitted run, + never here. Also sets cas_required on the svc-authelia key. + journey CCR-2026-0059 policy and the standing read role for sso/nk-bind-executor. + +setup: under both approved, sha256-bound CCRs (binding confirmed) and an attended platform-admin session: + check that the coding-agent fence denies both paths, apply both policies, set cas_required, create the + journey role, prove that the write role does not exist, and prove that the fence beats both policies. +verify: read back policies, cas_required and the role; log in as sso/nk-bind-executor (audience openbao) and prove + that the token reads only the journey path, with exactly its policy and at most 15m. Prove that a wrong + ServiceAccount, a wrong audience and a login to the write role are all denied. +Receipts hold closed booleans only. The journey values are read into memory to check their field names, and never +printed or written. +""" +import argparse +import hashlib +import importlib.util +import json +import os +import subprocess +import sys +from pathlib import Path + +HERE = Path(__file__).resolve().parent +REPO = HERE.parents[1] +PLATFORM = Path(os.environ.get('NK_PLATFORM_ROOT', REPO.parent / 'railiance-platform')) +sys.path.insert(0, str(PLATFORM / 'scripts')) +from state_hub_preflight_lane import bao, data, capabilities, revoke # noqa: E402 +from repair_eso_kubernetes_auth import role_payload, check_role # noqa: E402 +from telemetry_smtp_delivery import ensure # noqa: E402 + +SA, NS = 'nk-bind-executor', 'sso' +APPROVERS = {'platform-owner', 'net-kingdom-owner'} +BOUNDARY = 'sys/policies/acl/agent-high-risk-boundary' +BINDS = 'platform/data/workloads/net-kingdom/directory-binds/' +LANES = { + 'write': { + 'ccr': 'CCR-2026-0058-net-kingdom-bind-executor-svc-authelia-write.yaml', + 'request_type': 'workload-kv-write-next-version', + 'kv': 'platform/workloads/net-kingdom/directory-binds/svc-authelia', 'fields': ['LDAP_BIND_PASSWORD'], + 'policy': 'net-kingdom-bind-executor-svc-authelia-write', + 'policy_text': ('path "platform/data/workloads/net-kingdom/directory-binds/svc-authelia" {\n' + ' capabilities = ["read", "update"]\n allowed_parameters = {\n "data" = []\n' + ' "options" = []\n }\n}\n\npath "auth/token/lookup-self" {\n capabilities = ["read"]\n}\n\n' + 'path "auth/token/revoke-self" {\n capabilities = ["update"]\n}\n'), + }, + 'journey': { + 'ccr': 'CCR-2026-0059-net-kingdom-bind-executor-syn-journey-read.yaml', + 'request_type': 'workload-kv-read', + 'kv': 'platform/workloads/net-kingdom/synthetic-journey/syn-journey', 'fields': ['PASSWORD', 'TOTP_SEED'], + 'policy': 'net-kingdom-bind-executor-syn-journey-read', + 'policy_text': ('path "platform/data/workloads/net-kingdom/synthetic-journey/syn-journey" ' + '{ capabilities = ["read"] }\n' + 'path "auth/token/lookup-self" { capabilities = ["read"] }\n' + 'path "auth/token/revoke-self" { capabilities = ["update"] }\n'), + }, +} +for _lane in LANES.values(): + _lane.update(role=_lane['policy'], service_account=SA, namespace=NS, + data_path=_lane['kv'].replace('platform/', 'platform/data/', 1), + metadata_path=_lane['kv'].replace('platform/', 'platform/metadata/', 1)) +CAS_KEY = ['-mount=platform', 'workloads/net-kingdom/directory-binds/svc-authelia'] + + +class Refused(Exception): + pass + + +def load(name, path): + spec = importlib.util.spec_from_file_location(name, path) + module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module) + return module + + +def contract(name, expected): + lane = LANES[name] + path = PLATFORM / 'credential-change-requests' / lane['ccr'] + if hashlib.sha256(path.read_bytes()).hexdigest() != expected: raise Refused(name + ':ccr_digest') + ccr, errors, _ = load('credential_change', PLATFORM / 'scripts/credential-change.py').validate_ccr(path) + if errors or ccr['status'] not in {'approved', 'applied', 'verified', 'active'}: raise Refused(name + ':ccr_status') + approved = {r['reviewer'] for r in ccr['review'].get('comments', []) if r.get('decision') == 'approved'} + if not APPROVERS <= approved: raise Refused(name + ':ccr_approval') + a = ccr['openbao']; auth = a['auth'] + if (ccr['request_type'] != lane['request_type'] or a['kv_path'] != lane['kv'] or a['fields'] != lane['fields'] + or a['policy_name'] != lane['policy'] or auth['role'] != lane['role'] or auth['policies'] != [lane['policy']] + or auth.get('method') != 'kubernetes' or auth.get('audience') != 'openbao' + or auth['bound_claims'] != {'service_account_names': [SA], 'service_account_namespaces': [NS]} + or auth.get('token_no_default_policy') is not True + or any(auth.get(k) != '15m' for k in ('ttl', 'token_max_ttl', 'token_explicit_max_ttl'))): + raise Refused(name + ':ccr_shape') + if auth.get('bound_claims_confirmed') is not True: raise Refused(name + ':binding_unconfirmed') + if (PLATFORM / a['policy_file']).read_text() != lane['policy_text']: raise Refused(name + ':policy_file') + return ccr + + +def attended_platform_admin(): + ident = data(bao(['token', 'lookup', '-format=json']))['data'] + if ('platform-admin' not in ident.get('policies', []) or 'root' in ident.get('policies', []) + or not ident.get('path', '').startswith('auth/netkingdom/')): + raise Refused('attended_platform_admin_required') + + +def fence_present(): + current = data(bao(['read', '-format=json', BOUNDARY]))['data']['policy'] + return all('path "' + p + '" { capabilities = ["deny"] }' in current + for lane in LANES.values() for p in (lane['data_path'], lane['metadata_path'])) + + +def policy_check(lane): + def check(actual): + if actual.get('policy') != lane['policy_text']: raise Refused(lane['policy'] + ':policy_drift') + return check + + +def cas_required(): + return data(bao(['kv', 'metadata', 'get', '-format=json', *CAS_KEY]))['data'].get('cas_required') is True + + +def write_role_absent(): + result = bao(['read', '-format=json', 'auth/kubernetes/role/' + LANES['write']['role']], allow_failure=True) + return result.returncode != 0 and b'No value found' in result.stdout + result.stderr + + +def contracts(digests): + return {name: contract(name, digests[name]) for name in LANES} + + +def setup(digests, receipt): + receipt['phase'] = 'contract'; ccrs = contracts(digests) + receipt['phase'] = 'identity'; attended_platform_admin() + receipt['phase'] = 'coding_agent_fence' + if not fence_present(): raise Refused('agent_fence_missing') # established by CCR-2026-0053 and -0057 + receipt['phase'] = 'policies' + for lane in LANES.values(): + ensure('sys/policies/acl/' + lane['policy'], {'policy': lane['policy_text']}, policy_check(lane)) + receipt['phase'] = 'cas_required' + if not cas_required(): + bao(['kv', 'metadata', 'patch', '-cas-required=true', *CAS_KEY]) + if not cas_required(): raise Refused('cas_required_not_set') + receipt['phase'] = 'journey_role' + journey = LANES['journey'] + ensure('auth/kubernetes/role/' + journey['role'], role_payload(journey), lambda actual: check_role(actual, journey)) + receipt['phase'] = 'write_role_absent' + if not write_role_absent(): raise Refused('write_role_exists_outside_a_window') + receipt['phase'] = 'fence_effective' + for lane in LANES.values(): + child = data(bao(['token', 'create', '-format=json', '-policy=' + lane['policy'], + '-policy=agent-high-risk-boundary', '-no-default-policy', '-ttl=60s']))['auth']['client_token'] + try: + caps = capabilities(child, [lane['data_path'], lane['metadata_path']]) + if any(v != ['deny'] for v in caps.values()): raise Refused(lane['policy'] + ':fence_ineffective') + finally: + revoke(child) + receipt.update(phase='done', status='lanes_prepared', policies_applied=True, cas_required=True, + journey_role_created=True, write_role_absent=True, coding_agents_denied=True, + source_ccrs={n: c['id'] for n, c in ccrs.items()}, ccr_sha256=digests) + + +def kube(kubeconfig, args): + return subprocess.run(['kubectl', '--kubeconfig', kubeconfig, *args], check=True, capture_output=True).stdout + + +def verify(digests, kubeconfig, receipt): + receipt['phase'] = 'contract'; ccrs = contracts(digests) + receipt['phase'] = 'identity'; attended_platform_admin() + receipt['phase'] = 'read_back' + for lane in LANES.values(): + policy_check(lane)(data(bao(['read', '-format=json', 'sys/policies/acl/' + lane['policy']]))['data']) + journey, write = LANES['journey'], LANES['write'] + check_role(data(bao(['read', '-format=json', 'auth/kubernetes/role/' + journey['role']]))['data'], journey) + if not cas_required(): raise Refused('cas_required_not_set') + if not write_role_absent(): raise Refused('write_role_exists_outside_a_window') + if not fence_present(): raise Refused('agent_fence_missing') + + def login(sa, audience, role, negative=False): + jwt = kube(kubeconfig, ['-n', NS, 'create', 'token', sa, '--audience=' + audience, '--duration=10m']).decode().strip() + result = bao(['write', '-format=json', 'auth/kubernetes/login', '-'], + payload={'role': role, 'jwt': jwt}, allow_failure=negative) + if negative: + if result.returncode == 0: + revoke(data(result)['auth']['client_token']); raise Refused('negative_login_accepted') + if b'403' not in result.stderr and b'400' not in result.stderr: raise Refused('negative_login_unclear') + return None + return data(result)['auth'] + + receipt['phase'] = 'journey_login' + auth = login(SA, 'openbao', journey['role']); token = auth['client_token'] + try: + if (auth['token_policies'] != [journey['policy']] or auth.get('identity_policies') + or not 0 < auth['lease_duration'] <= 900): + raise Refused('journey_policy_or_ttl') + receipt['phase'] = 'journey_capabilities' + others = [BINDS + n for n in ('svc-authelia', 'svc-keycape', 'svc-privacyidea', 'svc-provisioner')] + paths = [journey['data_path'], journey['metadata_path'], write['metadata_path'], *others, 'auth/token/create'] + caps = capabilities(token, paths) + if caps[paths[0]] != ['read'] or any(caps[p] != ['deny'] for p in paths[1:]): + raise Refused('journey_capabilities') + receipt['phase'] = 'journey_read' + native = data(bao(['read', '-format=json', journey['data_path']], token=token))['data'] + if sorted(native['data']) != sorted(journey['fields']): raise Refused('journey_fields') + receipt['journey_kv_version'] = native['metadata']['version'] + finally: + revoke(token) + receipt['phase'] = 'negative_logins' + login('default', 'openbao', journey['role'], True) + login(SA, 'wrong-audience', journey['role'], True) + login(SA, 'openbao', write['role'], True) + receipt.update(phase='done', status='lanes_verified', exact_policies_and_role=True, cas_required=True, + journey_reads_only_its_path=True, journey_lease_at_most_15m=True, write_role_absent=True, + wrong_service_account_denied=True, wrong_audience_denied=True, write_login_denied=True, + coding_agents_denied=True, temporary_token_revoked=True, + source_ccrs={n: c['id'] for n, c in ccrs.items()}, ccr_sha256=digests) + + +def main(): + p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) + p.add_argument('action', choices=['setup', 'verify']) + p.add_argument('--write-ccr-sha256', required=True, help='sha256 of the CCR-2026-0058 file') + p.add_argument('--journey-ccr-sha256', required=True, help='sha256 of the CCR-2026-0059 file') + p.add_argument('--kubeconfig', help='admin kubeconfig; required for verify') + p.add_argument('--receipt', required=True) + a = p.parse_args() + if a.action == 'verify' and not a.kubeconfig: p.error('--kubeconfig is required for verify') + digests = {'write': a.write_ccr_sha256, 'journey': a.journey_ccr_sha256} + fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) + r = {'status': 'failed', 'action': a.action, 'values_recorded': False, + 'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest()} + try: + if a.action == 'setup': setup(digests, r) + else: verify(digests, a.kubeconfig, r) + except Exception as error: + r.update(status='failed', reason=str(error) if isinstance(error, Refused) else type(error).__name__) + finally: + with os.fdopen(fd, 'w') as out: json.dump(r, out, indent=2, sort_keys=True); out.write('\n') + print(json.dumps({k: r[k] for k in ('status', 'phase', 'reason') if k in r})) + return int(r['status'] == 'failed') + + +if __name__ == '__main__': + raise SystemExit(main())