f3b0c4f4b2
- CCR-2026-0063: NetKingdom's answer recorded (ESO as drafted); delivery aligned to the Secret the rapp Job mounts (sso/bounded-executor-tokens, key audit-core); reader role in the CCR-2026-0033/0041/0060 shape (audience openbao, explicit max TTL, self-lifecycle); custody_procedure names audit-core's seeder (one seeder only). - CCR-2026-0060: delivery block (store rendered from the issuer template); fence step named. - manifests/nk-bind-executor-eso.yaml: two sso-restricted stores and ExternalSecrets. - scripts/nk_bind_executor_custody.py plan|fence|verify with synthetic-fixture tests. - Source coding-agent boundary carries the KeyCape nk-bind-executor-approval path. - docs/credential-lane-designs/nk-bind-executor-go-live.md: exact founder sequence. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 523928@bnt-lap001 Assistant-Session: 6c737afb-9c14-4152-9f02-477214b9c350
426 lines
22 KiB
Python
426 lines
22 KiB
Python
#!/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())
|