#!/usr/bin/env python3 """Attended custody for the bounded executor's credentials (NK-WP-0056-T06, RPF-WP-0069-T05/T09). Covers the platform side of three requests. The delegated applier (credential-change.py applier-apply) writes their policies and roles; the credential broker (credential.py generate) mints the KeyCape client secret. audit-core's founder-run scripts/seed-nk-bounded-executor-sender.py seeds the audit bearer (CAS=0 custody plus registry merge and receiver reload). This script adds what none of those does: plan offline: check the three CCRs against the fixed contract and print names and paths. fence attended platform-admin: add the coding-agent deny for both custody paths to the live agent-high-risk-boundary (append only, live content preserved) and read it back. verify attended platform-admin plus admin kubeconfig: metadata, capabilities, login results and in-memory hashes only. Never reads the KeyCape client secret's data path. CCR-2026-0060 KeyCape ESO reader of platform/workloads/keycape/nk-bind-executor-approval CCR-2026-0061 executor (sso/nk-bind-executor) reader of the same path CCR-2026-0063 audit sender custody platform/workloads/net-kingdom/bounded-executor/audit-sender Every action except plan requires the three CCRs approved by their named owners with confirmed bindings, committed unmodified in Git, and the Warden attended-login envelope (no BAO_TOKEN/VAULT_TOKEN, a platform-admin token that is not root). fence and verify print nothing (the Warden envelope fails closed on output); the exit code is 0 or 1 and the receipt (mode 0600, refused if it exists) holds names, versions, booleans and digests only. """ from __future__ import annotations import argparse import base64 import hashlib import importlib.util import json import os from pathlib import Path import re import subprocess from datetime import datetime, timezone from state_hub_preflight_lane import (ROOT, LaneError, assert_cluster, bao, capabilities, command, data, revoke) BOUNDARY = 'sys/policies/acl/agent-high-risk-boundary' REGISTRY = 'platform/data/workloads/audit-core/senders' CLIENT_ID = 'nk-bind-executor-approval' KEYCAPE_KV = 'platform/workloads/keycape/' + CLIENT_ID AUDIT_KV = 'platform/workloads/net-kingdom/bounded-executor/audit-sender' SA, NS = 'nk-bind-executor', 'sso' ESO = ('external-secrets', 'external-secrets') SENDER = { 'name': 'nk-bounded-executor', 'sources': ['net-kingdom.bounded-executor'], 'tenants': ['platform'], 'may_write': True, 'may_read': False, 'evidence_kind': 'load-bearing', 'secret_policy': 'reject', 'closing_classes': {'execution.run.started': {'closed_by': 'execution.attribution.receipt', 'within_seconds': 7200}}, } def data_path(kv): return kv.replace('platform/', 'platform/data/', 1) def metadata_path(kv): return kv.replace('platform/', 'platform/metadata/', 1) # The fixed contract. Any drift in a CCR refuses before OpenBao is contacted. LANES = { 'eso': { 'ccr': 'CCR-2026-0060', 'file': 'CCR-2026-0060-nk-bind-executor-approval-keycape-eso-reader.yaml', 'approvers': ('platform-owner',), 'kv': KEYCAPE_KV, 'fields': ['CLIENT_SECRET'], 'policy': 'workload-kv-read-keycape-nk-bind-executor-approval-eso', 'role': 'external-secrets-keycape-nk-bind-executor-approval', 'bound': ESO, 'external_secret': ('sso', 'keycape-nk-bind-executor-approval-client'), 'store': 'openbao-keycape-nk-bind-executor-approval-issuer', 'secret_key': 'client-secret', }, 'consumer': { 'ccr': 'CCR-2026-0061', 'file': 'CCR-2026-0061-nk-bind-executor-approval-consumer-reader.yaml', 'approvers': ('platform-owner', 'net-kingdom-owner'), 'kv': KEYCAPE_KV, 'fields': ['CLIENT_SECRET'], 'policy': 'workload-kv-read-keycape-nk-bind-executor-approval-bounded-executor', 'role': 'keycape-nk-bind-executor-approval-bounded-executor-workload-kv-read', 'bound': (SA, NS), }, 'audit': { 'ccr': 'CCR-2026-0063', 'file': 'CCR-2026-0063-nk-bind-executor-audit-sender.yaml', 'approvers': ('platform-owner', 'audit-core-owner', 'net-kingdom-owner'), 'kv': AUDIT_KV, 'fields': ['AUDIT_TOKEN', 'CUSTODY_REQUEST'], 'policy': 'workload-kv-read-nk-bind-executor-audit', 'role': 'external-secrets-nk-bind-executor-audit', 'bound': ESO, 'external_secret': ('sso', 'bounded-executor-tokens'), 'store': 'openbao-nk-bind-executor-audit', 'secret_key': 'audit-core', }, } FENCED = (data_path(KEYCAPE_KV), metadata_path(KEYCAPE_KV), data_path(AUDIT_KV), metadata_path(AUDIT_KV)) def require(condition, code): if not condition: raise LaneError(code) def credential_module(): spec = importlib.util.spec_from_file_location('nk_ccr', ROOT / 'scripts/credential-change.py') module = importlib.util.module_from_spec(spec) spec.loader.exec_module(module) return module def committed(path): """The CCR must be the reviewed, committed bytes: tracked and unmodified.""" result = subprocess.run(['git', '-C', str(ROOT), 'status', '--porcelain', '--', str(path)], capture_output=True, text=True, timeout=20) tracked = subprocess.run(['git', '-C', str(ROOT), 'ls-files', '--error-unmatch', str(path)], capture_output=True, timeout=20) return result.returncode == 0 and not result.stdout.strip() and tracked.returncode == 0 def contracts(*, approved, module=None, git_check=committed): module = module or credential_module() out = {} for name, lane in LANES.items(): path = ROOT / 'credential-change-requests' / lane['file'] ccr, errors, _ = module.validate_ccr(path) require(not errors and ccr['id'] == lane['ccr'], name + ':invalid_credential_request') o = ccr['openbao'] auth = o['auth'] require(ccr['request_type'] == 'workload-kv-read' and o['kv_path'] == lane['kv'] and o['fields'] == lane['fields'] and o['policy_name'] == lane['policy'] and o.get('metadata_read') is False and o.get('token_self_lifecycle') is True and auth.get('method') == 'kubernetes' and auth.get('mount') == 'kubernetes' and auth.get('role') == lane['role'] and auth.get('policies') == [lane['policy']] and auth.get('audience') == 'openbao' and auth.get('token_no_default_policy') is True and all(auth.get(k) == '15m' for k in ('ttl', 'token_max_ttl', 'token_explicit_max_ttl')) and not auth.get('token_num_uses') and auth.get('bound_claims') == {'service_account_names': [lane['bound'][0]], 'service_account_namespaces': [lane['bound'][1]]}, name + ':exact_contract_required') hcl = module.generated_policy_hcl(ccr) require(module.normalized_policy_body((ROOT / o['policy_file']).read_text()) == module.normalized_policy_body(hcl), name + ':policy_source_drift') if approved: require(ccr['status'] in {'approved', 'applied', 'verified', 'active'}, name + ':approved_ccr_required') given = {c.get('reviewer') for c in ccr['review'].get('comments', []) if c.get('decision') == 'approved'} require(set(lane['approvers']) <= given, name + ':named_owner_approvals_required') require(auth.get('bound_claims_confirmed') is True, name + ':binding_unconfirmed') require(git_check(path), name + ':ccr_not_committed') out[name] = {'ccr': ccr, 'hcl': hcl, 'sha256': hashlib.sha256(path.read_bytes()).hexdigest()} return out def attended_platform_admin(): require(Path.home().parent.name == '.warden-attended-login' and not os.environ.get('BAO_TOKEN') and not os.environ.get('VAULT_TOKEN'), 'attended_warden_envelope_required') identity = data(bao(['token', 'lookup', '-format=json']))['data'] require('platform-admin' in identity.get('policies', []) and 'root' not in identity.get('policies', []), 'attended_platform_admin_required') def read_optional(path): result = bao(['read', '-format=json', path], allow_failure=True) if result.returncode == 0: return data(result)['data'] require(b'404' in result.stderr or b'No value found' in result.stderr or b'No value found' in result.stdout, 'absence_not_proven') return None def deny_line(path): return f'path "{path}" {{ capabilities = ["deny"] }}' def fence(receipt): current = read_optional(BOUNDARY) require(current is not None, 'coding_agent_boundary_required') current = current['policy'] additions = [] for path in FENCED: if f'"{path}"' in current: require(re.search(r'path\s+"' + re.escape(path) + r'"\s*\{\s*capabilities\s*=\s*\["deny"\]\s*\}', current), 'boundary_path_drift') else: additions.append(deny_line(path)) if additions: block = ('\n# Bounded executor custody (CCR-2026-0060/0061/0063), ' 'scripts/nk_bind_executor_custody.py fence.\n' + '\n'.join(additions) + '\n') require(read_optional(BOUNDARY)['policy'] == current, 'boundary_revision_changed') bao(['write', BOUNDARY, '-'], payload={'policy': current + block}) require(read_optional(BOUNDARY)['policy'] == current + block, 'boundary_readback_failed') receipt.update(fence_lines_added=len(additions), fenced_paths=list(FENCED), status='fenced') def registry_snapshot(): raw = data(bao(['read', '-format=json', REGISTRY]))['data'] version = raw['metadata']['version'] require(type(version) is int and version > 0, 'registry_version_missing') body = raw['data'] require(isinstance(body, dict) and isinstance(body.get('senders.json'), str), 'registry_shape_invalid') try: rows = json.loads(body['senders.json']) except (ValueError, TypeError): raise LaneError('registry_shape_invalid') from None require(isinstance(rows, list) and bool(rows), 'registry_must_preserve_existing_senders') names, tokens = set(), set() for row in rows: require(isinstance(row, dict) and isinstance(row.get('name'), str) and row['name'] and row['name'] not in names, 'registry_identity_invalid_or_duplicate') names.add(row['name']) values = row.get('tokens') or ([row['token']] if row.get('token') else []) require(isinstance(values, list) and bool(values), 'registry_tokens_invalid') for value in values: require(isinstance(value, str) and value and value not in tokens, 'registry_token_collision') tokens.add(value) return version, body, rows, tokens def role_ok(actual, lane): return (actual.get('bound_service_account_names') == [lane['bound'][0]] and actual.get('bound_service_account_namespaces') == [lane['bound'][1]] and not actual.get('bound_service_account_namespace_selector') and actual.get('token_policies') == [lane['policy']] and actual.get('audience') == 'openbao' and actual.get('token_ttl') == 900 and actual.get('token_max_ttl') == 900 and actual.get('token_explicit_max_ttl') == 900 and actual.get('token_no_default_policy') is True and not actual.get('token_period') and not actual.get('token_num_uses')) def kube_json(kube, args): return data(command(kube + [*args, '-o', 'json'])) def ready(obj): return any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in (obj.get('status') or {}).get('conditions', [])) def sha(value): return hashlib.sha256(value.encode() if isinstance(value, str) else value).hexdigest() def login(kube, sa, namespace, audience, role, negative=False): jwt = command(kube + ['-n', namespace, 'create', 'token', sa, '--audience=' + audience, '--duration=10m']).stdout.decode().strip() result = bao(['write', '-format=json', 'auth/kubernetes/login', '-'], payload={'role': role, 'jwt': jwt}, allow_failure=True) del jwt if negative: if result.returncode == 0: revoke(data(result)['auth']['client_token']) raise LaneError('negative_login_accepted') require(b'403' in result.stderr or b'400' in result.stderr, 'negative_login_unclear') return None require(result.returncode == 0, 'login_refused') return data(result)['auth'] def verify(kube, contract, receipt): module = credential_module() receipt['phase'] = 'read_back' for name, lane in LANES.items(): live = read_optional('sys/policies/acl/' + lane['policy']) require(live is not None and module.normalized_policy_body(live['policy']) == module.normalized_policy_body(contract[name]['hcl']), name + ':policy_drift') role = read_optional('auth/kubernetes/role/' + lane['role']) require(role is not None and role_ok(role, lane), name + ':role_drift') boundary = read_optional(BOUNDARY)['policy'] require(all(deny_line(p) in boundary or re.search(r'path\s+"' + re.escape(p) + r'"\s*\{\s*capabilities\s*=\s*\["deny"\]\s*\}', boundary) for p in FENCED), 'agent_fence_missing') receipt['phase'] = 'custody_metadata' for kv, key in ((KEYCAPE_KV, 'keycape_client_secret'), (AUDIT_KV, 'audit_sender')): meta = read_optional(metadata_path(kv)) require(meta is not None, key + ':custody_missing') current = meta.get('current_version') version = (meta.get('versions') or {}).get(str(current)) or {} require(type(current) is int and current >= 1 and not version.get('deletion_time') and not version.get('destroyed'), key + ':custody_version_not_live') receipt[key + '_kv_version'] = current receipt['phase'] = 'fence_effective' for name, lane in LANES.items(): 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, [data_path(lane['kv']), metadata_path(lane['kv'])]) require(all(v == ['deny'] for v in caps.values()), name + ':fence_ineffective') finally: revoke(child) default_child = data(bao(['token', 'create', '-format=json', '-policy=default', '-ttl=60s']))['auth']['client_token'] try: caps = capabilities(default_child, [data_path(KEYCAPE_KV), data_path(AUDIT_KV)]) require(all(v == ['deny'] for v in caps.values()), 'default_identity_can_read') finally: revoke(default_child) receipt['phase'] = 'consumer_login' consumer = LANES['consumer'] auth = login(kube, SA, NS, 'openbao', consumer['role']) token = auth['client_token'] try: require(auth['token_policies'] == [consumer['policy']] and not auth.get('identity_policies') and 0 < auth['lease_duration'] <= 900, 'consumer_policy_or_ttl') siblings = ['platform/data/workloads/keycape/secrets-engine-openbao', 'platform/data/workloads/keycape/forgejo', data_path(AUDIT_KV), 'platform/data/workloads/net-kingdom/directory-binds/svc-authelia', metadata_path(KEYCAPE_KV), 'platform/metadata/workloads/keycape', 'auth/token/create'] caps = capabilities(token, [data_path(KEYCAPE_KV), *siblings]) require(caps[data_path(KEYCAPE_KV)] == ['read'] and all(caps[p] == ['deny'] for p in siblings), 'consumer_capabilities') finally: revoke(token) receipt['phase'] = 'negative_logins' login(kube, 'default', NS, 'openbao', consumer['role'], True) login(kube, SA, NS, 'wrong-audience', consumer['role'], True) login(kube, SA, NS, 'openbao', LANES['eso']['role'], True) login(kube, SA, NS, 'openbao', LANES['audit']['role'], True) receipt['phase'] = 'rbac' can = command(kube + ['auth', 'can-i', 'get', 'secrets', '-n', NS, '--as=system:serviceaccount:%s:%s' % (NS, SA)], allow_failure=True) require(can.stdout.strip() == b'no', 'consumer_can_read_secrets_in_sso') receipt['phase'] = 'eso_delivery' for name in ('eso', 'audit'): lane = LANES[name] store = kube_json(kube, ['get', 'clustersecretstore', lane['store']]) vault = store['spec']['provider']['vault'] k8s = vault['auth']['kubernetes'] require(ready(store) and vault['path'] == 'platform' and k8s['role'] == lane['role'] and k8s['serviceAccountRef'] == {'name': 'external-secrets', 'namespace': 'external-secrets', 'audiences': ['openbao']} and store['spec'].get('conditions') == [{'namespaces': [NS]}], name + ':store_drift_or_not_ready') namespace, es_name = lane['external_secret'] es = kube_json(kube, ['-n', namespace, 'get', 'externalsecret', es_name]) entries = es['spec']['data'] require(ready(es) and es['spec']['secretStoreRef'] == {'kind': 'ClusterSecretStore', 'name': lane['store']} and len(entries) == 1 and entries[0]['secretKey'] == lane['secret_key'] and entries[0]['remoteRef']['key'] == lane['kv'].removeprefix('platform/') and entries[0]['remoteRef']['property'] == lane['fields'][0], name + ':external_secret_drift_or_not_ready') target = es['spec']['target'].get('name', es_name) secret = kube_json(kube, ['-n', namespace, 'get', 'secret', target]) require(sorted(secret.get('data') or {}) == [lane['secret_key']], name + ':delivered_keys') receipt[name + '_delivery'] = {'external_secret': f'{namespace}/{es_name}', 'ready': True, 'secret_uid': secret['metadata']['uid'], 'keys': [lane['secret_key']]} if name == 'audit': # Hash equality in memory: custody, delivered Secret and registry row hold the same bearer. custody = data(bao(['read', '-format=json', data_path(AUDIT_KV)]))['data']['data'] require(sorted(custody) == ['AUDIT_TOKEN', 'CUSTODY_REQUEST'] and custody['CUSTODY_REQUEST'] == 'CCR-2026-0063', 'audit_custody_shape') digest = sha(custody['AUDIT_TOKEN']) delivered = sha(base64.b64decode(secret['data'][lane['secret_key']])) _, _, rows, _ = registry_snapshot() row = [r for r in rows if r['name'] == SENDER['name']] require(len(row) == 1 and {k: v for k, v in row[0].items() if k != 'tokens'} == SENDER and [sha(t) for t in row[0]['tokens']] == [digest], 'audit_registry_row_mismatch') require(delivered == digest, 'audit_delivery_disagrees_with_custody') del custody, rows, row receipt['audit_copies_agree_by_hash'] = True receipt.update(phase='done', status='custody_verified_pending_keycape_and_executor', exact_policies_and_roles=True, coding_agents_denied=True, default_identity_denied=True, consumer_reads_only_its_path=True, consumer_lease_at_most_15m=True, wrong_service_account_denied=True, wrong_audience_denied=True, consumer_cannot_use_eso_roles=True, consumer_cannot_read_sso_secrets=True, keycape_client_secret_value_read=False) def main(argv=None): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument('action', choices=['plan', 'fence', 'verify']) parser.add_argument('--receipt', type=Path) parser.add_argument('--kubeconfig') args = parser.parse_args(argv) receipt = {'schema': 'platform.nk-bind-executor-custody.v1', 'action': args.action, 'status': 'refused', 'credential_values_emitted': False, 'started_at': datetime.now(timezone.utc).isoformat()} fd = None def save(): if fd is not None: os.lseek(fd, 0, os.SEEK_SET) os.ftruncate(fd, 0) os.write(fd, (json.dumps(receipt, indent=2, sort_keys=True) + '\n').encode()) os.fsync(fd) try: if args.action == 'plan': contract = contracts(approved=False) receipt.update(status='proposed', fenced_paths=list(FENCED), sender=SENDER, lanes={ n: {'ccr': l['ccr'], 'status': contract[n]['ccr']['status'], 'kv': l['kv'], 'policy': l['policy'], 'role': l['role'], 'bound': '/'.join(reversed(l['bound'])), 'store': l.get('store'), 'external_secret': '/'.join(l.get('external_secret', ())) or None, 'ccr_sha256': contract[n]['sha256']} for n, l in LANES.items()}) print(json.dumps(receipt, indent=2)) return 0 require(args.receipt is not None, 'receipt_required') fd = os.open(args.receipt, os.O_RDWR | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600) receipt['phase'] = 'contract' contract = contracts(approved=True) receipt['ccr_sha256'] = {LANES[n]['ccr']: c['sha256'] for n, c in contract.items()} receipt['phase'] = 'identity' attended_platform_admin() save() if args.action == 'fence': fence(receipt) else: require(args.kubeconfig, 'kubeconfig_required') kube = ['kubectl', '--kubeconfig', args.kubeconfig, '--request-timeout=20s'] assert_cluster(kube) verify(kube, contract, receipt) code = 0 except LaneError as exc: receipt.update(status='refused', error=str(exc)) code = 1 except Exception: receipt.update(status='refused', error='contained_operation_failed') code = 1 receipt['finished_at'] = datetime.now(timezone.utc).isoformat() save() if fd is not None: os.close(fd) # Silent outside plan: the Warden envelope fails closed on any output. Read the receipt. return code if __name__ == '__main__': raise SystemExit(main())