NK-WP-0056-T04: svc-authelia bind lane (source only, CCR-2026-0053)
bind-lane.py gains per-lane Secret keys. Authelia keeps its bind user in authelia-config, so setup refuses unless the live configuration binds uid=svc-authelia, and only the ldap_password key is decoded. The ESO manifest merges ldap_password alone into authelia-secrets. The tests are parametrized per lane (12). tools/service-binds is now listed in security-tools.json, which fixes the index test that failed after the pilot. 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
This commit is contained in:
parent
436c5fe585
commit
034634ac5f
5 changed files with 164 additions and 27 deletions
|
|
@ -16,6 +16,9 @@ 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'
|
||||
|
||||
pytestmark = pytest.mark.skipif(not (PLATFORM / 'credential-change-requests' / CCR).exists(),
|
||||
reason='railiance-platform checkout with CCR-2026-0052 required')
|
||||
|
|
@ -27,10 +30,11 @@ def load():
|
|||
|
||||
|
||||
class FakeBao:
|
||||
def __init__(self):
|
||||
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 = {('net-kingdom-svc-provisioner-bind-eso', 'openbao')}
|
||||
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)
|
||||
|
|
@ -38,7 +42,7 @@ class FakeBao:
|
|||
if args[:2] == ['token', 'create']: return out({'auth': {'client_token': 'child'}})
|
||||
if args[0] == 'read':
|
||||
path = args[-1]
|
||||
if path == KV:
|
||||
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]})
|
||||
|
|
@ -47,17 +51,17 @@ class FakeBao:
|
|||
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'])
|
||||
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': ['net-kingdom-svc-provisioner-bind-eso'],
|
||||
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 == KV:
|
||||
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/'):
|
||||
|
|
@ -67,28 +71,28 @@ class FakeBao:
|
|||
raise AssertionError(args)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def lane(tmp_path, monkeypatch):
|
||||
def build(tmp_path, monkeypatch, name='svc-provisioner'):
|
||||
authelia = name == 'svc-authelia'
|
||||
ccr_file = AUTHELIA_CCR if authelia 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')
|
||||
shutil.copy(PLATFORM / 'openbao/policies/net-kingdom-svc-provisioner-bind-eso.hcl', platform / 'openbao/policies/')
|
||||
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).read_text())
|
||||
ccr = yaml.safe_load((PLATFORM / 'credential-change-requests' / ccr_file).read_text())
|
||||
ccr['status'] = 'approved'
|
||||
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
|
||||
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()
|
||||
for name in ('bao',):
|
||||
monkeypatch.setattr(tool, name, bao)
|
||||
bao = FakeBao(AUTHELIA_KV, 'net-kingdom-svc-authelia-bind-eso') if authelia else FakeBao()
|
||||
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'])
|
||||
|
|
@ -99,28 +103,45 @@ def lane(tmp_path, monkeypatch):
|
|||
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'}
|
||||
live = {'value': VALUE, 'ready': 'True', 'bind_line': BIND_LINE}
|
||||
|
||||
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']:
|
||||
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 verb == ['get', 'externalsecret', 'authelia-secrets' if authelia else '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)
|
||||
key = 'LLDAP_SVC_AUTHELIA_PASS' if authelia else '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')
|
||||
|
||||
|
||||
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)
|
||||
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
|
||||
|
||||
|
|
@ -191,3 +212,34 @@ def test_setup_refuses_unconfirmed_binding(lane):
|
|||
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'}}]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue