tegwick
c43afd46a7
- bounded-executor/prepare_admission.py: the live admission from the pilot example, pins read from the image's commit, window <= 2 h (audit-core's declared bound), validated. - executor-lanes.py window-open/window-close: CCR-2026-0058's window-scoped write role, created for one admitted open window under attended platform-admin and deleted afterwards (idempotent). - security-cases/nk-bind-executor-approval.yaml: the case of record, proposed (founder acceptance). - docs/evidence/2026-11-04-nk56-t15-audit-sender: 8 checks against audit-core 876c3f6. 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
227 lines
11 KiB
Python
227 lines
11 KiB
Python
"""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)
|
|
copy = root / 'credential-change-requests' / name
|
|
if not confirm: # the real CCRs are confirmed since 2026-10-31; undo it in the copy
|
|
copy.write_text(copy.read_text().replace('bound_claims_confirmed: true', 'bound_claims_confirmed: false'))
|
|
elif 'bound_claims_confirmed: true' not in copy.read_text():
|
|
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()}
|
|
|
|
|
|
def m_key(m):
|
|
return m.CAS_KEY
|
|
|
|
|
|
class FakeBao:
|
|
def __init__(self, m):
|
|
self.m = m
|
|
self.store = {m.BOUNDARY: {'policy': 'path "x" { capabilities = ["deny"] }\n' + FENCE}}
|
|
self.cas = False
|
|
self.meta = {'max_versions': 0, 'delete_version_after': '0s', 'custom_metadata': None}
|
|
self.meta_write_clobbers = 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[0] == 'read' and args[-1] == m_key(self.m):
|
|
return out({'data': {'cas_required': self.cas, **self.meta}})
|
|
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] == 'delete':
|
|
self.store.pop(args[-1], None); return out({})
|
|
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 == m_key(self.m):
|
|
assert payload == {'cas_required': True}
|
|
self.cas = True
|
|
if self.meta_write_clobbers: self.meta = {**self.meta, 'max_versions': 10}
|
|
return out({})
|
|
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'),
|
|
(lambda m, f, d: setattr(f, 'meta_write_clobbers', True), 'metadata_changed_beyond_cas_required'),
|
|
])
|
|
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)
|
|
|
|
|
|
ADMISSION = ROOT / 'examples/bounded-execution-admission/bind-rotation-svc-authelia.yaml'
|
|
|
|
|
|
def admission_at(tmp_path, m):
|
|
"""The pilot example, its window around a fixed instant."""
|
|
import datetime as dt
|
|
bea = m.load('bea_for_test', m.BEA)
|
|
doc = bea.load(ADMISSION)
|
|
path = tmp_path / 'a.yaml'
|
|
path.write_text(json.dumps(doc))
|
|
start = bea.parse_time(doc['spec']['window']['not_before'])
|
|
return path, start + dt.timedelta(minutes=10)
|
|
|
|
|
|
def test_window_open_creates_the_exact_write_role_once_then_close_removes_it(tmp_path, monkeypatch):
|
|
m, fake, d = build(tmp_path, monkeypatch)
|
|
run(m, 'setup', d)
|
|
path, now = admission_at(tmp_path, m)
|
|
r = {}
|
|
m.window_open(d, path, r, now=now)
|
|
role = 'auth/kubernetes/role/' + m.LANES['write']['role']
|
|
assert r['status'] == 'window_opened' and r['admission'] == 'BEA-2026-0001' and role in fake.store
|
|
m.check_role(fake.store[role], m.LANES['write'])
|
|
with pytest.raises(m.Refused, match='write_role_already_open'):
|
|
m.window_open(d, path, {}, now=now)
|
|
c = {}
|
|
m.window_close(path, c)
|
|
assert c['status'] == 'window_closed' and c['write_role_was_present'] and role not in fake.store
|
|
c = {}
|
|
m.window_close(path, c) # idempotent, also late
|
|
assert c['status'] == 'window_closed' and not c['write_role_was_present']
|
|
|
|
|
|
@pytest.mark.parametrize('mutate, reason', [
|
|
(lambda m, f, now: (now - __import__('datetime').timedelta(hours=1)), 'admission_window_not_open'),
|
|
(lambda m, f, now: (f.identity.update(policies=['root']), now)[1], 'attended_platform_admin_required'),
|
|
(lambda m, f, now: (setattr(f, 'cas', False), now)[1], 'cas_required_not_set'),
|
|
(lambda m, f, now: (f.store.update({m.BOUNDARY: {'policy': ''}}), now)[1], 'agent_fence_missing'),
|
|
])
|
|
def test_window_open_refusals(tmp_path, monkeypatch, mutate, reason):
|
|
m, fake, d = build(tmp_path, monkeypatch)
|
|
run(m, 'setup', d)
|
|
path, now = admission_at(tmp_path, m)
|
|
with pytest.raises(m.Refused, match=reason):
|
|
m.window_open(d, path, {}, now=mutate(m, fake, now))
|
|
assert 'auth/kubernetes/role/' + m.LANES['write']['role'] not in fake.store
|