tegwick
c6df34c4de
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 673304@bnt-lap001 Assistant-Session: 959b12c7-cf2e-440b-a60d-91229cc9e96e
178 lines
9.3 KiB
Python
178 lines
9.3 KiB
Python
"""bind-lane.py against an in-memory OpenBao and Kubernetes; no values leave the fakes."""
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import shutil
|
|
import subprocess
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PLATFORM = ROOT.parent / 'railiance-platform'
|
|
CCR = 'CCR-2026-0052-net-kingdom-identity-provisioner-bind.yaml'
|
|
VALUE = 'SyntheticProvisionerBind1234567890abcdef'
|
|
OTHER = 'DifferentProvisionerBind1234567890abcdef'
|
|
KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-provisioner'
|
|
|
|
pytestmark = pytest.mark.skipif(not (PLATFORM / 'credential-change-requests' / CCR).exists(),
|
|
reason='railiance-platform checkout with CCR-2026-0052 required')
|
|
|
|
|
|
def load():
|
|
spec = importlib.util.spec_from_file_location('bind_lane', ROOT / 'tools/service-binds/bind-lane.py')
|
|
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module); return module
|
|
|
|
|
|
class FakeBao:
|
|
def __init__(self):
|
|
self.store = {'sys/policies/acl/agent-high-risk-boundary': {'policy': 'path "x" { capabilities = ["deny"] }\n'}}
|
|
self.kv = {}; self.writes = []; self.identity = {'policies': ['platform-admin'], 'path': 'auth/netkingdom/oidc/callback'}
|
|
self.login_ok = {('net-kingdom-svc-provisioner-bind-eso', 'openbao')}
|
|
|
|
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)
|
|
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':
|
|
path = args[-1]
|
|
if path == KV:
|
|
return out({'data': {'data': self.kv['value'], 'metadata': {'version': self.kv['version']}}}) if self.kv \
|
|
else out({}, 2, b'No value found at ' + path.encode())
|
|
if path in self.store: return out({'data': self.store[path]})
|
|
return out({}, 2, b'No value found')
|
|
if args[0] == 'write':
|
|
path = args[-2] if args[-1] == '-' else args[-1]
|
|
if path == 'auth/token/revoke': return out({})
|
|
if path == 'sys/capabilities':
|
|
caps = {p: (['read'] if p == KV and token is None and payload['token'] == 'reader' else ['deny'])
|
|
for p in payload['paths']}
|
|
return out({'data': caps})
|
|
if path == 'auth/kubernetes/login':
|
|
sa, aud = payload['jwt'].split('|')
|
|
if (sa, aud) in self.login_ok:
|
|
return out({'auth': {'client_token': 'reader', 'token_policies': ['net-kingdom-svc-provisioner-bind-eso'],
|
|
'lease_duration': 900}})
|
|
return out({}, 2, b'Code: 403')
|
|
self.writes.append(path)
|
|
if path == KV:
|
|
assert payload['options'] == {'cas': 0}
|
|
self.kv = {'value': payload['data'], 'version': 1}; return out({'data': {'version': 1}})
|
|
if path.startswith('auth/kubernetes/role/'):
|
|
stored = {k: (900 if v == '15m' else v) for k, v in payload.items()}
|
|
self.store[path] = stored; return out({})
|
|
self.store[path] = payload; return out({})
|
|
raise AssertionError(args)
|
|
|
|
|
|
@pytest.fixture
|
|
def lane(tmp_path, monkeypatch):
|
|
tool = load()
|
|
platform = tmp_path / 'platform'
|
|
(platform / 'credential-change-requests').mkdir(parents=True); (platform / 'openbao/policies').mkdir(parents=True)
|
|
(platform / 'scripts').mkdir()
|
|
shutil.copy(PLATFORM / 'scripts/credential-change.py', platform / 'scripts/credential-change.py')
|
|
shutil.copy(PLATFORM / 'openbao/policies/net-kingdom-svc-provisioner-bind-eso.hcl', platform / 'openbao/policies/')
|
|
ccr = json.loads((PLATFORM / 'credential-change-requests' / CCR).read_text())
|
|
ccr['status'] = 'approved'
|
|
ccr['review']['comments'] = [{'reviewer': f'User ({r})', 'decision': 'approved', 'source_ref': 'statehub:decision:synthetic',
|
|
'comment': 'synthetic', 'reviewed_at': '2026-10-30T00:00:00+00:00'}
|
|
for r in ('platform-owner', 'net-kingdom-owner')]
|
|
path = platform / 'credential-change-requests' / CCR
|
|
path.write_text(json.dumps(ccr, indent=2) + '\n')
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
monkeypatch.setattr(tool, 'PLATFORM', platform)
|
|
bao = FakeBao()
|
|
for name in ('bao',):
|
|
monkeypatch.setattr(tool, name, bao)
|
|
monkeypatch.setattr(tool, 'data', lambda r: json.loads(r.stdout))
|
|
monkeypatch.setattr(tool, 'capabilities', lambda token, paths: json.loads(
|
|
bao(['write', '-format=json', 'sys/capabilities', '-'], payload={'paths': paths, 'token': token}).stdout)['data'])
|
|
monkeypatch.setattr(tool, 'revoke', lambda token: None)
|
|
|
|
def ensure(path, wanted, check):
|
|
old = bao(['read', '-format=json', path], allow_failure=True)
|
|
if old.returncode != 0: bao(['write', path, '-'], payload=wanted)
|
|
check(json.loads(bao(['read', '-format=json', path]).stdout)['data'])
|
|
monkeypatch.setattr(tool, 'ensure', ensure)
|
|
live = {'value': VALUE, 'ready': 'True'}
|
|
|
|
def kube(kubeconfig, args, payload=None):
|
|
verb = args[2:5]
|
|
if verb == ['get', 'secret', 'identity-provisioner-ldap']:
|
|
return json.dumps({'data': {'username': base64.b64encode(b'svc-provisioner').decode(),
|
|
'password': base64.b64encode(live['value'].encode()).decode()}}).encode()
|
|
if verb == ['get', 'externalsecret', 'identity-provisioner-ldap']:
|
|
return json.dumps({'status': {'conditions': [{'type': 'Ready', 'status': live['ready']}]}}).encode()
|
|
if verb[:2] == ['create', 'token']:
|
|
return (args[4] + '|' + args[5].split('=', 1)[1]).encode()
|
|
raise AssertionError(args)
|
|
monkeypatch.setattr(tool, 'kube', kube)
|
|
custody = tmp_path / 'custody'; (custody / 'lldap').mkdir(parents=True, mode=0o700); custody.chmod(0o700)
|
|
env = custody / 'lldap/secrets.env'; env.write_text(f'LLDAP_SVC_PROVISIONER_PASS={VALUE}\n'); env.chmod(0o600)
|
|
return SimpleNamespace(tool=tool, bao=bao, live=live, digest=digest, custody=str(custody), ccr=path)
|
|
|
|
|
|
def run(ns, action):
|
|
receipt = {}
|
|
if action == 'setup': ns.tool.setup('svc-provisioner', ns.digest, ns.custody, 'kc', receipt)
|
|
else: ns.tool.verify('svc-provisioner', ns.digest, 'kc', receipt)
|
|
assert VALUE not in json.dumps(receipt)
|
|
return receipt
|
|
|
|
|
|
def test_setup_then_verify(lane):
|
|
r = run(lane, 'setup')
|
|
assert r['status'] == 'custody_and_reader_prepared' and r['kv_version'] == 1
|
|
assert lane.bao.writes[0] == 'sys/policies/acl/agent-high-risk-boundary' and KV in lane.bao.writes
|
|
assert KV in lane.bao.store['sys/policies/acl/agent-high-risk-boundary']['policy']
|
|
assert run(lane, 'setup')['existing_identical_custody'] is True # rerun is idempotent
|
|
assert run(lane, 'verify')['status'] == 'reader_and_delivery_verified'
|
|
|
|
|
|
def test_setup_refuses_custody_drift_before_any_write(lane):
|
|
lane.live['value'] = OTHER
|
|
with pytest.raises(lane.tool.Refused, match='custody_drift'): run(lane, 'setup')
|
|
assert lane.bao.writes == []
|
|
|
|
|
|
def test_setup_refuses_unapproved_or_changed_ccr(lane):
|
|
with pytest.raises(lane.tool.Refused, match='ccr_digest'):
|
|
lane.tool.setup('svc-provisioner', '0' * 64, lane.custody, 'kc', {})
|
|
ccr = json.loads(lane.ccr.read_text()); ccr['status'] = 'proposed'; lane.ccr.write_text(json.dumps(ccr))
|
|
digest = hashlib.sha256(lane.ccr.read_bytes()).hexdigest()
|
|
with pytest.raises(lane.tool.Refused, match='ccr_status'):
|
|
lane.tool.setup('svc-provisioner', digest, lane.custody, 'kc', {})
|
|
assert lane.bao.writes == []
|
|
|
|
|
|
def test_setup_refuses_non_admin_session(lane):
|
|
lane.bao.identity = {'policies': ['default'], 'path': 'auth/netkingdom/oidc/callback'}
|
|
with pytest.raises(lane.tool.Refused, match='attended_platform_admin_required'): run(lane, 'setup')
|
|
|
|
|
|
def test_verify_refuses_unsynced_or_mismatched_delivery(lane):
|
|
run(lane, 'setup')
|
|
lane.live['ready'] = 'False'
|
|
with pytest.raises(lane.tool.Refused, match='external_secret_not_ready'): run(lane, 'verify')
|
|
lane.live.update(ready='True', value=OTHER)
|
|
with pytest.raises(lane.tool.Refused, match='kv_live_mismatch'): run(lane, 'verify')
|
|
|
|
|
|
def test_verify_refuses_when_wrong_service_account_can_log_in(lane):
|
|
run(lane, 'setup')
|
|
lane.bao.login_ok.add(('default', 'openbao'))
|
|
with pytest.raises(lane.tool.Refused, match='negative_login_accepted'): run(lane, 'verify')
|
|
|
|
|
|
def test_manifest_matches_contract():
|
|
import yaml
|
|
docs = [d for d in yaml.safe_load_all((ROOT / 'sso-mfa/k8s/identity-provisioner/bind-eso.yaml').read_text()) if d]
|
|
sa, store, es = docs
|
|
assert sa['metadata'] == {'name': 'net-kingdom-svc-provisioner-bind-eso', 'namespace': 'sso'}
|
|
assert store['spec']['provider']['vault']['auth']['kubernetes']['role'] == 'net-kingdom-svc-provisioner-bind-eso'
|
|
assert es['spec']['target'] == {'name': 'identity-provisioner-ldap', 'creationPolicy': 'Merge', 'deletionPolicy': 'Retain'}
|
|
assert es['spec']['data'] == [{'secretKey': 'password', 'remoteRef': {
|
|
'key': 'workloads/net-kingdom/directory-binds/svc-provisioner', 'property': 'LDAP_BIND_PASSWORD'}}]
|