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
341 lines
19 KiB
Python
Executable file
341 lines
19 KiB
Python
Executable file
#!/usr/bin/env python3
|
|
"""Attended platform-custody lane for one LLDAP service bind (NK-WP-0056-T04).
|
|
|
|
setup: under an approved, sha256-bound CCR and an attended platform-admin session,
|
|
fence coding agents, write the bind from the decrypted custody copy to KV
|
|
(CAS=0, or confirm an identical existing value), then policy and role. Refuses
|
|
unless custody equals the live Secret, so no rotation happens here.
|
|
verify: after the ExternalSecret is applied, native reader login as the ESO
|
|
ServiceAccount, exact policy/TTL, denials, and KV == live Secret in memory.
|
|
|
|
Receipts hold closed booleans only; values are never printed or written.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import sys
|
|
|
|
HERE = Path(__file__).resolve().parent
|
|
REPO = HERE.parents[1]
|
|
PLATFORM = Path(os.environ.get('NK_PLATFORM_ROOT', REPO.parent / 'railiance-platform'))
|
|
sys.path.insert(0, str(PLATFORM / 'scripts'))
|
|
from state_hub_preflight_lane import bao, data, capabilities, revoke # noqa: E402
|
|
from repair_eso_kubernetes_auth import role_payload, check_role # noqa: E402
|
|
from telemetry_smtp_delivery import ensure # noqa: E402
|
|
|
|
LANES = {
|
|
'svc-provisioner': {
|
|
'ccr': 'CCR-2026-0052-net-kingdom-identity-provisioner-bind.yaml',
|
|
'custody_key': 'LLDAP_SVC_PROVISIONER_PASS',
|
|
'secret': 'identity-provisioner-ldap', 'username': 'svc-provisioner',
|
|
'external_secret': 'identity-provisioner-ldap',
|
|
},
|
|
# Authelia keeps the bind user in its configuration, not in the Secret, and reads
|
|
# ldap_password from a file at start-up only.
|
|
'svc-authelia': {
|
|
'ccr': 'CCR-2026-0053-net-kingdom-authelia-bind.yaml',
|
|
'custody_key': 'LLDAP_SVC_AUTHELIA_PASS',
|
|
'secret': 'authelia-secrets', 'secret_key': 'ldap_password', 'username': 'svc-authelia',
|
|
'bind_user_config': ('authelia-config', 'configuration.yml',
|
|
'user: uid=svc-authelia,ou=people,dc=netkingdom,dc=local'),
|
|
'external_secret': 'authelia-secrets',
|
|
},
|
|
# KeyCape reads the bind from config.yaml in keycape-config until it takes
|
|
# KEYCAPE_LLDAP_BIND_PW (key-cape KEY-WP-0046); ESO owns a new dedicated Secret.
|
|
'svc-keycape': {
|
|
'ccr': 'CCR-2026-0054-net-kingdom-keycape-bind.yaml',
|
|
'custody_key': 'LLDAP_SVC_KEYCAPE_PASS',
|
|
'secret': 'keycape-lldap-bind', 'secret_key': 'password', 'username': 'svc-keycape',
|
|
'bind_in_config_secret': ('keycape-config', 'config.yaml',
|
|
'uid=svc-keycape,ou=people,dc=netkingdom,dc=local'),
|
|
'external_secret': 'keycape-lldap-bind',
|
|
},
|
|
# privacyIDEA stores the resolver password in its own database and accepts it only from
|
|
# pi-admin with MFA: no workload reader, no ESO. The value is proven current by an LLDAP
|
|
# sign-in and delivered by realm_bootstrap.py --rotate-bind --from-openbao (attended).
|
|
'svc-privacyidea': {
|
|
'ccr': 'CCR-2026-0055-net-kingdom-privacyidea-bind.yaml',
|
|
'custody_key': 'LLDAP_SVC_PRIVACYIDEA_PASS', 'username': 'svc-privacyidea',
|
|
'attended': True, 'directory': 'https://lldap.coulomb.social',
|
|
'procedure': 'net-kingdom sso-mfa/k8s/privacyidea/realm_bootstrap.py --rotate-bind --from-openbao',
|
|
},
|
|
}
|
|
ATTENDED_AUTH = {'method': 'attended-oidc', 'mount': 'netkingdom', 'role': 'platform-admin'}
|
|
BIND_NAMES = ('svc-authelia', 'svc-keycape', 'svc-privacyidea', 'svc-provisioner')
|
|
FIELD = 'LDAP_BIND_PASSWORD'
|
|
BOUNDARY = 'sys/policies/acl/agent-high-risk-boundary'
|
|
APPROVERS = {'platform-owner', 'net-kingdom-owner'}
|
|
|
|
|
|
class Refused(Exception):
|
|
pass
|
|
|
|
|
|
def load(name, path):
|
|
spec = importlib.util.spec_from_file_location(name, path)
|
|
module = importlib.util.module_from_spec(spec); spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
def contract(name, expected):
|
|
spec = LANES[name]
|
|
path = PLATFORM / 'credential-change-requests' / spec['ccr']
|
|
raw = path.read_bytes()
|
|
if hashlib.sha256(raw).hexdigest() != expected: raise Refused('ccr_digest')
|
|
ccr, errors, _ = load('credential_change', PLATFORM / 'scripts/credential-change.py').validate_ccr(path)
|
|
if errors or ccr['status'] not in {'approved', 'applied', 'verified', 'active'}: raise Refused('ccr_status')
|
|
approved = {r['reviewer'].removeprefix('User (').removesuffix(')') for r in ccr['review'].get('comments', [])
|
|
if r.get('decision') == 'approved' and (r.get('source_ref') or r.get('at'))}
|
|
if not APPROVERS <= approved: raise Refused('ccr_approval')
|
|
role = 'net-kingdom-' + name + '-bind-eso'
|
|
kv = 'platform/workloads/net-kingdom/directory-binds/' + name
|
|
if spec.get('attended'):
|
|
a, d = ccr['openbao'], ccr['delivery']
|
|
if (ccr['request_type'] != 'attended-service-credential-delivery' or a['kv_path'] != kv
|
|
or a['fields'] != [FIELD] or a['auth'] != ATTENDED_AUTH or a['runtime_access'] is not False
|
|
or d['procedure'] != spec['procedure']):
|
|
raise Refused('ccr_shape')
|
|
return ccr, {'kv_path': kv.replace('platform/', 'platform/data/', 1), 'role': role, **spec}
|
|
a = ccr['openbao']; auth = a['auth']; d = ccr['delivery']
|
|
if (a['kv_path'] != kv or a['fields'] != [FIELD] or a['policy_name'] != role or auth['role'] != role
|
|
or auth['policies'] != [role] or auth.get('audience') != 'openbao'
|
|
or auth['bound_claims'] != {'service_account_names': [role], 'service_account_namespaces': ['sso']}
|
|
or auth.get('bound_claims_confirmed') is not True or auth.get('token_no_default_policy') is not True or a.get('metadata_read') is not False
|
|
or any(auth.get(k) != '15m' for k in ('ttl', 'token_max_ttl', 'token_explicit_max_ttl'))
|
|
or d['namespace'] != 'sso' or d['secret'] != spec['secret'] or d['key'] != spec.get('secret_key', 'password')
|
|
or d['service_account'] != role):
|
|
raise Refused('ccr_shape')
|
|
lane = {'service_account': role, 'namespace': 'sso', 'role': role, 'policy': role,
|
|
'kv_path': kv.replace('platform/', 'platform/data/', 1), **spec}
|
|
policy = ('path "' + lane['kv_path'] + '" { capabilities = ["read"] }\n'
|
|
'path "auth/token/lookup-self" { capabilities = ["read"] }\n'
|
|
'path "auth/token/revoke-self" { capabilities = ["update"] }\n')
|
|
if (PLATFORM / a['policy_file']).read_text() != policy: raise Refused('policy_file')
|
|
lane['policy_text'] = policy
|
|
return ccr, lane
|
|
|
|
|
|
def kube(kubeconfig, args, payload=None):
|
|
return subprocess.run(['kubectl', '--kubeconfig', kubeconfig, *args], input=payload,
|
|
check=True, capture_output=True).stdout
|
|
|
|
|
|
def secret_data(kubeconfig, name):
|
|
return json.loads(kube(kubeconfig, ['-n', 'sso', 'get', 'secret', name, '-o', 'json'])).get('data', {})
|
|
|
|
|
|
def decoded(raw, key):
|
|
return base64.b64decode(raw.get(key, ''), validate=True).decode()
|
|
|
|
|
|
def configured_bind(kubeconfig, lane):
|
|
"""The bindPW in a service's config Secret, after checking it binds the lane's account."""
|
|
import yaml
|
|
name, entry, dn = lane['bind_in_config_secret']
|
|
config = yaml.safe_load(decoded(secret_data(kubeconfig, name), entry))
|
|
lldap = config.get('lldap') if isinstance(config, dict) else None
|
|
if not isinstance(lldap, dict) or lldap.get('bindDN') != dn: raise Refused('live_bind_user')
|
|
return lldap.get('bindPW') or ''
|
|
|
|
|
|
def live_secret(kubeconfig, lane, delivered=True):
|
|
"""The bind value in use. For a config-embedded bind, setup (delivered=False) reads the
|
|
config; verify reads the delivered Secret, which must equal the config while both exist."""
|
|
key = lane.get('secret_key', 'password')
|
|
if 'bind_in_config_secret' in lane:
|
|
configured = configured_bind(kubeconfig, lane)
|
|
if not delivered:
|
|
if not configured: raise Refused('live_secret_shape')
|
|
return configured
|
|
value = decoded(secret_data(kubeconfig, lane['secret']), key)
|
|
if not value: raise Refused('live_secret_shape')
|
|
if configured and configured != value: raise Refused('kv_live_mismatch')
|
|
return value
|
|
raw = secret_data(kubeconfig, lane['secret'])
|
|
if 'bind_user_config' in lane:
|
|
name, entry, line = lane['bind_user_config']
|
|
config = json.loads(kube(kubeconfig, ['-n', 'sso', 'get', 'configmap', name, '-o', 'json'])).get('data', {})
|
|
if line not in [l.strip() for l in config.get(entry, '').splitlines()]: raise Refused('live_bind_user')
|
|
elif base64.b64decode(raw.get('username', ''), validate=True).decode() != lane['username']:
|
|
raise Refused('live_secret_shape')
|
|
value = base64.b64decode(raw.get(key, ''), validate=True).decode()
|
|
if not value: raise Refused('live_secret_shape')
|
|
return value
|
|
|
|
|
|
def custody_value(root, lane):
|
|
io = load('custody_io', REPO / 'sso-mfa/bootstrap/apply-service-secrets.py')
|
|
with io.custody(root, ['lldap']) as dirs:
|
|
return io.literal(io.read_env(dirs['lldap']), lane['custody_key'])
|
|
|
|
|
|
def attended_platform_admin():
|
|
ident = data(bao(['token', 'lookup', '-format=json']))['data']
|
|
if ('platform-admin' not in ident.get('policies', []) or 'root' in ident.get('policies', [])
|
|
or not ident.get('path', '').startswith('auth/netkingdom/')):
|
|
raise Refused('attended_platform_admin_required')
|
|
|
|
|
|
def directory_login(endpoint, user, password):
|
|
"""True if LLDAP accepts this password for the account: LLDAP is the authority for the current value."""
|
|
split = load('bind_split', REPO / 'sso-mfa/k8s/lldap/bind-split.py')
|
|
return split.Directory(endpoint).login(user, password) is not None
|
|
|
|
|
|
def fence_paths(lane):
|
|
return (lane['kv_path'], lane['kv_path'].replace('/data/', '/metadata/', 1))
|
|
|
|
|
|
def fence_present(lane):
|
|
current = data(bao(['read', '-format=json', BOUNDARY]))['data']['policy']
|
|
return all('path "' + p + '" { capabilities = ["deny"] }' in current for p in fence_paths(lane))
|
|
|
|
|
|
def setup(name, expected, root, kubeconfig, receipt, endpoint=None):
|
|
receipt['phase'] = 'contract'; ccr, lane = contract(name, expected)
|
|
receipt['phase'] = 'identity'; attended_platform_admin()
|
|
value = custody_value(root, lane)
|
|
if lane.get('attended'):
|
|
receipt['phase'] = 'custody_is_current'
|
|
if not directory_login(endpoint or lane['directory'], lane['username'], value): raise Refused('custody_not_current')
|
|
else:
|
|
receipt['phase'] = 'custody_equals_live'
|
|
if live_secret(kubeconfig, lane, delivered=False) != value: raise Refused('custody_drift')
|
|
receipt['phase'] = 'coding_agent_fence'
|
|
current = data(bao(['read', '-format=json', BOUNDARY]))['data']['policy']; added = ''
|
|
for path in (lane['kv_path'], lane['kv_path'].replace('/data/', '/metadata/', 1)):
|
|
stanza = 'path "' + path + '" { capabilities = ["deny"] }\n'
|
|
if '"' + path + '"' not in current: added += stanza
|
|
elif stanza.strip() not in current: raise Refused('fence_conflict')
|
|
if added:
|
|
if data(bao(['read', '-format=json', BOUNDARY]))['data']['policy'] != current: raise Refused('fence_race')
|
|
bao(['write', BOUNDARY, '-'], payload={'policy': current + '\n' + added})
|
|
if data(bao(['read', '-format=json', BOUNDARY]))['data']['policy'] != current + '\n' + added:
|
|
raise Refused('fence_readback')
|
|
receipt['phase'] = 'kv'
|
|
existing = bao(['read', '-format=json', lane['kv_path']], allow_failure=True)
|
|
if existing.returncode == 0:
|
|
if data(existing)['data']['data'] != {FIELD: value}: raise Refused('kv_differs')
|
|
receipt['kv_version'] = data(existing)['data']['metadata']['version']; receipt['existing_identical_custody'] = True
|
|
else:
|
|
if b'No value found' not in existing.stdout + existing.stderr: raise Refused('kv_absence_unproven')
|
|
written = data(bao(['write', '-format=json', lane['kv_path'], '-'],
|
|
payload={'options': {'cas': 0}, 'data': {FIELD: value}}))
|
|
receipt['kv_version'] = written['data']['version']
|
|
if lane.get('attended'):
|
|
receipt['phase'] = 'fence_present'
|
|
if not fence_present(lane): raise Refused('fence_readback')
|
|
receipt.update(phase='done', status='custody_prepared', custody_is_current=True, coding_agents_denied=True,
|
|
workload_access=False, source_ccr=ccr['id'], ccr_sha256=expected)
|
|
return
|
|
receipt['phase'] = 'policy_role'
|
|
policy = lane['policy_text']
|
|
ensure('sys/policies/acl/' + lane['policy'], {'policy': policy},
|
|
lambda actual: None if actual['policy'] == policy else (_ for _ in ()).throw(Refused('policy_drift')))
|
|
ensure('auth/kubernetes/role/' + lane['role'], role_payload(lane), lambda actual: check_role(actual, lane))
|
|
receipt['phase'] = 'fence_effective'
|
|
child = data(bao(['token', 'create', '-format=json', '-policy=' + lane['policy'], '-policy=agent-high-risk-boundary',
|
|
'-no-default-policy', '-ttl=60s']))['auth']['client_token']
|
|
try:
|
|
paths = [lane['kv_path'], lane['kv_path'].replace('/data/', '/metadata/', 1)]
|
|
if any(v != ['deny'] for v in capabilities(child, paths).values()): raise Refused('fence_ineffective')
|
|
finally:
|
|
revoke(child)
|
|
receipt.update(phase='done', status='custody_and_reader_prepared', custody_equals_live=True,
|
|
coding_agents_denied=True, source_ccr=ccr['id'], ccr_sha256=expected,
|
|
external_secret_apply_and_verify_pending=True)
|
|
|
|
|
|
def verify_attended(ccr, lane, expected, endpoint, receipt):
|
|
receipt['phase'] = 'kv_is_current'
|
|
native = data(bao(['read', '-format=json', lane['kv_path']]))['data']
|
|
if set(native['data']) != {FIELD} or not directory_login(endpoint or lane['directory'], lane['username'], native['data'][FIELD]):
|
|
raise Refused('kv_not_current')
|
|
receipt['kv_version'] = native['metadata']['version']
|
|
receipt['phase'] = 'fence_present'
|
|
if not fence_present(lane): raise Refused('fence_missing')
|
|
receipt['phase'] = 'no_workload_role'
|
|
if bao(['read', '-format=json', 'auth/kubernetes/role/' + lane['role']], allow_failure=True).returncode == 0:
|
|
raise Refused('unexpected_workload_role')
|
|
receipt.update(phase='done', status='custody_verified', kv_is_current=True, coding_agents_denied=True,
|
|
workload_access=False, source_ccr=ccr['id'], ccr_sha256=expected)
|
|
|
|
|
|
def verify(name, expected, kubeconfig, receipt, endpoint=None):
|
|
receipt['phase'] = 'contract'; ccr, lane = contract(name, expected)
|
|
receipt['phase'] = 'identity'; attended_platform_admin()
|
|
if lane.get('attended'): return verify_attended(ccr, lane, expected, endpoint, receipt)
|
|
receipt['phase'] = 'external_secret_ready'
|
|
es = json.loads(kube(kubeconfig, ['-n', 'sso', 'get', 'externalsecret', lane['external_secret'], '-o', 'json']))
|
|
if not any(c.get('type') == 'Ready' and c.get('status') == 'True' for c in es.get('status', {}).get('conditions', [])):
|
|
raise Refused('external_secret_not_ready')
|
|
|
|
def login(sa, audience, negative=False):
|
|
jwt = kube(kubeconfig, ['-n', 'sso', 'create', 'token', sa, '--audience=' + audience, '--duration=10m']).decode().strip()
|
|
result = bao(['write', '-format=json', 'auth/kubernetes/login', '-'],
|
|
payload={'role': lane['role'], 'jwt': jwt}, allow_failure=negative)
|
|
if negative:
|
|
if result.returncode == 0:
|
|
revoke(data(result)['auth']['client_token']); raise Refused('negative_login_accepted')
|
|
if b'403' not in result.stderr and b'400' not in result.stderr: raise Refused('negative_login_unclear')
|
|
return None
|
|
return data(result)['auth']
|
|
|
|
receipt['phase'] = 'native_reader_login'
|
|
auth = login(lane['service_account'], 'openbao'); token = auth['client_token']
|
|
try:
|
|
if (auth['token_policies'] != [lane['policy']] or auth.get('identity_policies')
|
|
or not 0 < auth['lease_duration'] <= 900):
|
|
raise Refused('reader_policy_or_ttl')
|
|
receipt['phase'] = 'reader_capabilities'
|
|
others = ['platform/data/workloads/net-kingdom/directory-binds/' + n for n in BIND_NAMES if n != name]
|
|
paths = [lane['kv_path'], lane['kv_path'].replace('/data/', '/metadata/', 1), *others, 'auth/token/create']
|
|
cap = capabilities(token, paths)
|
|
if cap[paths[0]] != ['read'] or any(cap[p] != ['deny'] for p in paths[1:]): raise Refused('reader_capabilities')
|
|
receipt['phase'] = 'kv_equals_live'
|
|
native = data(bao(['read', '-format=json', lane['kv_path']], token=token))['data']
|
|
if set(native['data']) != {FIELD} or native['data'][FIELD] != live_secret(kubeconfig, lane):
|
|
raise Refused('kv_live_mismatch')
|
|
receipt['kv_version'] = native['metadata']['version']
|
|
finally:
|
|
revoke(token)
|
|
receipt['phase'] = 'negative_logins'
|
|
login('default', 'openbao', True); login(lane['service_account'], 'wrong-audience', True)
|
|
check_role(data(bao(['read', '-format=json', 'auth/kubernetes/role/' + lane['role']]))['data'], lane)
|
|
receipt.update(phase='done', status='reader_and_delivery_verified', external_secret_ready=True,
|
|
exact_policy_and_ttl_verified=True, sibling_metadata_and_token_create_denied=True,
|
|
wrong_service_account_denied=True, wrong_audience_denied=True,
|
|
eso_delivery_equality_verified=True, temporary_reader_revoked=True,
|
|
source_ccr=ccr['id'], ccr_sha256=expected)
|
|
|
|
|
|
def main():
|
|
p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
|
|
p.add_argument('action', choices=['setup', 'verify']); p.add_argument('lane', choices=sorted(LANES))
|
|
p.add_argument('--ccr-sha256', required=True); p.add_argument('--kubeconfig', help='required for ESO lanes')
|
|
p.add_argument('--endpoint', help='LLDAP web endpoint for attended lanes (default: the lane\'s directory)')
|
|
p.add_argument('--custody', help='decrypted custody root (setup only)'); p.add_argument('--receipt', required=True)
|
|
a = p.parse_args()
|
|
if (a.action == 'setup') != bool(a.custody): p.error('--custody is required for setup and refused for verify')
|
|
if not LANES[a.lane].get('attended') and not a.kubeconfig: p.error('--kubeconfig is required for this lane')
|
|
fd = os.open(a.receipt, os.O_WRONLY | os.O_CREAT | os.O_EXCL | os.O_NOFOLLOW, 0o600)
|
|
r = {'status': 'failed', 'action': a.action, 'lane': a.lane, 'values_recorded': False,
|
|
'source_sha256': hashlib.sha256(Path(__file__).read_bytes()).hexdigest()}
|
|
try:
|
|
if a.action == 'setup': setup(a.lane, a.ccr_sha256, a.custody, a.kubeconfig, r, a.endpoint)
|
|
else: verify(a.lane, a.ccr_sha256, a.kubeconfig, r, a.endpoint)
|
|
except Exception as error:
|
|
r.update(status='failed', reason=str(error) if isinstance(error, Refused) else type(error).__name__)
|
|
finally:
|
|
with os.fdopen(fd, 'w') as out: json.dump(r, out, indent=2, sort_keys=True); out.write('\n')
|
|
print(json.dumps({k: r[k] for k in ('status', 'phase', 'reason') if k in r}))
|
|
return int(r['status'] == 'failed')
|
|
|
|
|
|
if __name__ == '__main__':
|
|
raise SystemExit(main())
|