tegwick
c28651e55b
bind-lane svc-privacyidea (CCR-2026-0055) has no workload reader. Setup and verify prove the value is current by an LLDAP sign-in as svc-privacyidea, fence coding agents, write KV with CAS=0 and refuse any workload role. realm_bootstrap.py --rotate-bind rewrites only BINDPW on the established service bind, --from-openbao reads it in memory under the attended platform-admin session, and an applied rotation requires --verify-user. Lane tests: 20; realm_bootstrap tests: 102. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com> Assistant: claude-code Assistant-Model: opus Assistant-Process: 896183@bnt-lap001 Assistant-Session: 3a18c01a-3205-4d02-94db-9efff25649a2
340 lines
18 KiB
Python
340 lines
18 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'
|
|
AUTHELIA_CCR = 'CCR-2026-0053-net-kingdom-authelia-bind.yaml'
|
|
AUTHELIA_KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-authelia'
|
|
BIND_LINE = 'user: uid=svc-authelia,ou=people,dc=netkingdom,dc=local'
|
|
KEYCAPE_CCR = 'CCR-2026-0054-net-kingdom-keycape-bind.yaml'
|
|
KEYCAPE_KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-keycape'
|
|
KEYCAPE_DN = 'uid=svc-keycape,ou=people,dc=netkingdom,dc=local'
|
|
PI_CCR = 'CCR-2026-0055-net-kingdom-privacyidea-bind.yaml'
|
|
PI_KV = 'platform/data/workloads/net-kingdom/directory-binds/svc-privacyidea'
|
|
|
|
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, kv=KV, role='net-kingdom-svc-provisioner-bind-eso'):
|
|
self.kv_path = kv; self.role = role
|
|
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 = {(role, '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 == self.kv_path:
|
|
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 == self.kv_path 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': [self.role],
|
|
'lease_duration': 900}})
|
|
return out({}, 2, b'Code: 403')
|
|
self.writes.append(path)
|
|
if path == self.kv_path:
|
|
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)
|
|
|
|
|
|
def build(tmp_path, monkeypatch, name='svc-provisioner'):
|
|
authelia, keycape, pi = name == 'svc-authelia', name == 'svc-keycape', name == 'svc-privacyidea'
|
|
ccr_file = AUTHELIA_CCR if authelia else KEYCAPE_CCR if keycape else PI_CCR if pi else CCR
|
|
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')
|
|
if pi:
|
|
shutil.copy(PLATFORM / 'scripts/attended_service_delivery.py', platform / 'scripts/')
|
|
else:
|
|
shutil.copy(PLATFORM / f'openbao/policies/net-kingdom-{name}-bind-eso.hcl', platform / 'openbao/policies/')
|
|
import yaml
|
|
ccr = yaml.safe_load((PLATFORM / 'credential-change-requests' / ccr_file).read_text())
|
|
ccr['status'] = 'approved'
|
|
if not pi: ccr['openbao']['auth']['bound_claims_confirmed'] = True
|
|
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_file
|
|
path.write_text(json.dumps(ccr, indent=2) + '\n')
|
|
digest = hashlib.sha256(path.read_bytes()).hexdigest()
|
|
monkeypatch.setattr(tool, 'PLATFORM', platform)
|
|
bao = (FakeBao(AUTHELIA_KV, 'net-kingdom-svc-authelia-bind-eso') if authelia
|
|
else FakeBao(KEYCAPE_KV, 'net-kingdom-svc-keycape-bind-eso') if keycape
|
|
else FakeBao(PI_KV, 'none') if pi else FakeBao())
|
|
# LLDAP accepts exactly the account's current password.
|
|
monkeypatch.setattr(tool, 'directory_login', lambda endpoint, user, password:
|
|
user == name and password == live['value'])
|
|
monkeypatch.setattr(tool, 'bao', 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)
|
|
# KeyCape: 'value' is bindPW in keycape-config, 'delivered' the ESO Secret (None until ESO runs).
|
|
live = {'value': VALUE, 'ready': 'True', 'bind_line': BIND_LINE, 'bind_dn': KEYCAPE_DN, 'delivered': None}
|
|
|
|
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 authelia and verb == ['get', 'secret', 'authelia-secrets']:
|
|
return json.dumps({'data': {'session_secret': base64.b64encode(b'unrelated').decode(),
|
|
'ldap_password': base64.b64encode(live['value'].encode()).decode()}}).encode()
|
|
if authelia and verb == ['get', 'configmap', 'authelia-config']:
|
|
return json.dumps({'data': {'configuration.yml': 'authentication_backend:\n ldap:\n ' + live['bind_line'] + '\n'}}).encode()
|
|
if keycape and verb == ['get', 'secret', 'keycape-config']:
|
|
config = ('issuer: https://kc.example.test\nlldap:\n url: ldap://lldap:3890\n'
|
|
f' bindDN: {live["bind_dn"]}\n bindPW: "{live["value"]}"\nclients: []\n')
|
|
return json.dumps({'data': {'config.yaml': base64.b64encode(config.encode()).decode(),
|
|
'key.pem': base64.b64encode(b'unrelated').decode()}}).encode()
|
|
if keycape and verb == ['get', 'secret', 'keycape-lldap-bind']:
|
|
if live['delivered'] is None: raise subprocess.CalledProcessError(1, 'kubectl')
|
|
return json.dumps({'data': {'password': base64.b64encode(live['delivered'].encode()).decode()}}).encode()
|
|
external = 'authelia-secrets' if authelia else 'keycape-lldap-bind' if keycape else 'identity-provisioner-ldap'
|
|
if verb == ['get', 'externalsecret', external]:
|
|
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)
|
|
key = {'svc-authelia': 'LLDAP_SVC_AUTHELIA_PASS', 'svc-keycape': 'LLDAP_SVC_KEYCAPE_PASS',
|
|
'svc-privacyidea': 'LLDAP_SVC_PRIVACYIDEA_PASS'}.get(name, 'LLDAP_SVC_PROVISIONER_PASS')
|
|
env = custody / 'lldap/secrets.env'; env.write_text(f'{key}={VALUE}\n'); env.chmod(0o600)
|
|
return SimpleNamespace(tool=tool, bao=bao, live=live, digest=digest, custody=str(custody), ccr=path, name=name)
|
|
|
|
|
|
@pytest.fixture
|
|
def lane(tmp_path, monkeypatch):
|
|
return build(tmp_path, monkeypatch)
|
|
|
|
|
|
@pytest.fixture
|
|
def authelia(tmp_path, monkeypatch):
|
|
if not (PLATFORM / 'credential-change-requests' / AUTHELIA_CCR).exists(): pytest.skip('CCR-2026-0053 required')
|
|
return build(tmp_path, monkeypatch, 'svc-authelia')
|
|
|
|
|
|
@pytest.fixture
|
|
def keycape(tmp_path, monkeypatch):
|
|
if not (PLATFORM / 'credential-change-requests' / KEYCAPE_CCR).exists(): pytest.skip('CCR-2026-0054 required')
|
|
return build(tmp_path, monkeypatch, 'svc-keycape')
|
|
|
|
|
|
def run(ns, action):
|
|
receipt = {}
|
|
if action == 'setup': ns.tool.setup(ns.name, ns.digest, ns.custody, 'kc', receipt)
|
|
else: ns.tool.verify(ns.name, 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'] == {'template': {'metadata': {}}, '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'}}]
|
|
|
|
|
|
def test_setup_refuses_unconfirmed_binding(lane):
|
|
ccr = yaml_load(lane.ccr); ccr['openbao']['auth']['bound_claims_confirmed'] = False
|
|
lane.ccr.write_text(json.dumps(ccr)); digest = hashlib.sha256(lane.ccr.read_bytes()).hexdigest()
|
|
with pytest.raises(lane.tool.Refused, match='ccr_shape'):
|
|
lane.tool.setup('svc-provisioner', digest, lane.custody, 'kc', {})
|
|
|
|
|
|
def yaml_load(path):
|
|
import yaml
|
|
return yaml.safe_load(path.read_text())
|
|
|
|
|
|
def test_authelia_setup_then_verify_reads_only_its_key(authelia):
|
|
r = run(authelia, 'setup')
|
|
assert r['status'] == 'custody_and_reader_prepared' and AUTHELIA_KV in authelia.bao.writes
|
|
assert KV not in authelia.bao.writes
|
|
assert run(authelia, 'verify')['status'] == 'reader_and_delivery_verified'
|
|
|
|
|
|
def test_authelia_setup_refuses_when_config_binds_another_account(authelia):
|
|
authelia.live['bind_line'] = 'user: uid=admin,ou=people,dc=netkingdom,dc=local'
|
|
with pytest.raises(authelia.tool.Refused, match='live_bind_user'): run(authelia, 'setup')
|
|
assert authelia.bao.writes == []
|
|
|
|
|
|
def test_authelia_setup_refuses_custody_drift(authelia):
|
|
authelia.live['value'] = OTHER
|
|
with pytest.raises(authelia.tool.Refused, match='custody_drift'): run(authelia, 'setup')
|
|
assert authelia.bao.writes == []
|
|
|
|
|
|
def test_authelia_manifest_matches_contract():
|
|
import yaml
|
|
docs = [d for d in yaml.safe_load_all((ROOT / 'sso-mfa/k8s/authelia/bind-eso.yaml').read_text()) if d]
|
|
sa, store, es = docs
|
|
assert sa['metadata'] == {'name': 'net-kingdom-svc-authelia-bind-eso', 'namespace': 'sso'}
|
|
assert store['spec']['provider']['vault']['auth']['kubernetes']['role'] == 'net-kingdom-svc-authelia-bind-eso'
|
|
assert es['spec']['target'] == {'template': {'metadata': {}}, 'name': 'authelia-secrets',
|
|
'creationPolicy': 'Merge', 'deletionPolicy': 'Retain'}
|
|
assert es['spec']['data'] == [{'secretKey': 'ldap_password', 'remoteRef': {
|
|
'key': 'workloads/net-kingdom/directory-binds/svc-authelia', 'property': 'LDAP_BIND_PASSWORD'}}]
|
|
|
|
|
|
def test_keycape_setup_reads_the_config_bind_then_verify_the_dedicated_secret(keycape):
|
|
r = run(keycape, 'setup') # the dedicated Secret does not exist yet
|
|
assert r['status'] == 'custody_and_reader_prepared' and KEYCAPE_KV in keycape.bao.writes
|
|
keycape.live['delivered'] = VALUE # ESO created keycape-lldap-bind
|
|
assert run(keycape, 'verify')['status'] == 'reader_and_delivery_verified'
|
|
keycape.live['value'] = '' # bindPW cleared from keycape-config: the delivered value alone counts
|
|
assert run(keycape, 'verify')['status'] == 'reader_and_delivery_verified'
|
|
|
|
|
|
def test_keycape_verify_refuses_delivery_differing_from_the_config(keycape):
|
|
run(keycape, 'setup')
|
|
keycape.live['delivered'] = OTHER
|
|
with pytest.raises(keycape.tool.Refused, match='kv_live_mismatch'): run(keycape, 'verify')
|
|
|
|
|
|
def test_keycape_setup_refuses_another_bind_account(keycape):
|
|
keycape.live['bind_dn'] = 'uid=admin,ou=people,dc=netkingdom,dc=local'
|
|
with pytest.raises(keycape.tool.Refused, match='live_bind_user'): run(keycape, 'setup')
|
|
assert keycape.bao.writes == []
|
|
|
|
|
|
def test_keycape_manifest_creates_a_dedicated_secret():
|
|
import yaml
|
|
docs = [d for d in yaml.safe_load_all((ROOT / 'sso-mfa/k8s/keycape/bind-eso.yaml').read_text()) if d]
|
|
sa, store, es = docs
|
|
assert sa['metadata'] == {'name': 'net-kingdom-svc-keycape-bind-eso', 'namespace': 'sso'}
|
|
assert store['spec']['provider']['vault']['auth']['kubernetes']['role'] == 'net-kingdom-svc-keycape-bind-eso'
|
|
assert es['spec']['target'] == {'template': {'metadata': {}}, 'name': 'keycape-lldap-bind',
|
|
'creationPolicy': 'Owner', 'deletionPolicy': 'Retain'}
|
|
assert es['spec']['data'] == [{'secretKey': 'password', 'remoteRef': {
|
|
'key': 'workloads/net-kingdom/directory-binds/svc-keycape', 'property': 'LDAP_BIND_PASSWORD'}}]
|
|
|
|
|
|
@pytest.fixture
|
|
def privacyidea(tmp_path, monkeypatch):
|
|
if not (PLATFORM / 'credential-change-requests' / PI_CCR).exists(): pytest.skip('CCR-2026-0055 required')
|
|
return build(tmp_path, monkeypatch, 'svc-privacyidea')
|
|
|
|
|
|
def test_privacyidea_setup_and_verify_need_no_workload_access(privacyidea):
|
|
r = run(privacyidea, 'setup')
|
|
assert r['status'] == 'custody_prepared' and r['custody_is_current'] and r['workload_access'] is False
|
|
assert PI_KV in privacyidea.bao.writes
|
|
assert not any(w.startswith(('auth/kubernetes/role/', 'sys/policies/acl/net-kingdom')) for w in privacyidea.bao.writes)
|
|
assert run(privacyidea, 'verify')['status'] == 'custody_verified'
|
|
|
|
|
|
def test_privacyidea_setup_refuses_a_password_ldap_rejects(privacyidea):
|
|
privacyidea.live['value'] = OTHER # the directory has moved on; custody is stale
|
|
with pytest.raises(privacyidea.tool.Refused, match='custody_not_current'): run(privacyidea, 'setup')
|
|
assert privacyidea.bao.writes == []
|
|
|
|
|
|
def test_privacyidea_verify_refuses_a_stale_kv_value(privacyidea):
|
|
run(privacyidea, 'setup')
|
|
privacyidea.live['value'] = OTHER # rotated in LLDAP without a new KV version
|
|
with pytest.raises(privacyidea.tool.Refused, match='kv_not_current'): run(privacyidea, 'verify')
|
|
|
|
|
|
def test_privacyidea_verify_refuses_a_workload_role(privacyidea):
|
|
run(privacyidea, 'setup')
|
|
privacyidea.bao.store['auth/kubernetes/role/net-kingdom-svc-privacyidea-bind-eso'] = {'policies': ['x']}
|
|
with pytest.raises(privacyidea.tool.Refused, match='unexpected_workload_role'): run(privacyidea, 'verify')
|