#!/usr/bin/env python3 """Isolated local-Docker LLDAP proof; synthetic credentials and temporary storage only.""" import argparse import hashlib import importlib.util import json import os from pathlib import Path import re import secrets import subprocess import tempfile import time import urllib.request ROOT = Path(__file__).resolve().parents[1] SPEC = importlib.util.spec_from_file_location('native_lldap_admin', ROOT / 'sso-mfa/k8s/lldap/lldap_admin.py') admin = importlib.util.module_from_spec(SPEC) SPEC.loader.exec_module(admin) def command(args): result = subprocess.run(args, capture_output=True, text=True, timeout=60) if result.returncode: raise RuntimeError('fixture_command_failed') return result.stdout.strip() def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--image', required=True, help='Existing local image sha256; never pull') parser.add_argument('--output', required=True) args = parser.parse_args() if not re.fullmatch(r'sha256:[0-9a-f]{64}', args.image): raise SystemExit('immutable local image required') contexts = json.loads(command(['docker', 'context', 'inspect'])) host = os.environ.get('DOCKER_HOST') or contexts[0]['Endpoints']['docker']['Host'] if not host.startswith('unix://'): raise SystemExit('local Unix Docker endpoint required') assert command(['docker', 'image', 'inspect', args.image, '--format', '{{.Id}}']) == args.image checks = {} container = None result = {'image': args.image, 'synthetic_only': True, 'checks': checks, 'outcome': 'unverified'} try: with tempfile.TemporaryDirectory(prefix='nk-lldap-fixture-', dir='/tmp') as temp: directory = Path(temp) password = secrets.token_hex(24) env = directory / 'container.env' env.write_text('LLDAP_LDAP_USER_PASS=' + password + '\nLLDAP_JWT_SECRET=' + secrets.token_hex(32) + '\nLLDAP_LDAP_BASE_DN=dc=example,dc=test\n') env.chmod(0o600) root = directory / 'custody' root.mkdir(mode=0o700) (root / 'lldap').mkdir(mode=0o700) source = root / 'lldap/secrets.env' source.write_text('LLDAP_LDAP_USER_PASS=' + password + '\n') source.chmod(0o600) container = command(['docker', 'run', '--detach', '--rm', '--pull=never', '--name', 'nk-lldap-groups-' + secrets.token_hex(5), '--memory=256m', '--cpus=0.5', '--pids-limit=128', '--publish', '127.0.0.1::17170', '--tmpfs', '/data:rw,nosuid,nodev,size=32m,uid=1000,gid=1000', '--env-file', str(env), args.image]) if not re.fullmatch(r'[0-9a-f]{64}', container): raise RuntimeError('invalid_container_id') port = json.loads(command(['docker', 'inspect', container, '--format', '{{json .NetworkSettings.Ports}}']))['17170/tcp'][0] assert port['HostIp'] == '127.0.0.1' base = 'http://127.0.0.1:' + port['HostPort'] provider = admin.LLDAPProvisioner(base_url=base, admin_password=password, opener=admin.Transport(base)) for _ in range(60): try: token = provider._login() break except Exception: time.sleep(.25) else: raise RuntimeError('fixture_start_failed') initial = admin.snapshot(provider, token) preview = admin.maintain(provider, base, 'bootstrap') assert admin.snapshot(provider, token) == initial checks['preview_nonmutating'] = True result1 = admin.maintain(provider, base, 'bootstrap', apply=True, expected_digest=preview['snapshot_sha256']) assert result1['result'] == 'verified' ready = admin.snapshot(provider, token) assert all(name in ready['groups'] for name in admin.BASELINE) assert all(ready['groups'][name] == value for name, value in initial['groups'].items()) checks['bootstrap_exact_and_existing_preserved'] = True preview = admin.maintain(provider, base, 'bootstrap') assert preview['changes'] == [] assert admin.maintain(provider, base, 'bootstrap', apply=True, expected_digest=preview['snapshot_sha256'])['result'] == 'verified' checks['bootstrap_idempotent'] = True provider._gql(token, 'mutation CreateUser($id: String!, $email: String!, $display: String!) { createUser(user: {id: $id, email: $email, displayName: $display}) { id } }', {'id': 'fixture-user', 'email': 'fixture@example.test', 'display': 'Fixture'}) for action in ('add', 'remove'): kwargs = {'action': action, 'group': 'activity-core-operators', 'user': 'fixture-user'} preview = admin.maintain(provider, base, 'membership', **kwargs) updated = admin.maintain(provider, base, 'membership', **kwargs, apply=True, expected_digest=preview['snapshot_sha256']) assert updated['result'] == 'verified' listed = admin.maintain(provider, base, 'membership', action='list', group='activity-core-operators') assert ('fixture-user' in listed['members']) == (action == 'add') checks['exact_' + action + '_and_readback'] = True before = admin.snapshot(provider, token) for group in ('net-kingdom-admins', 'net-kingdom-users', 'lldap_admin'): try: admin.maintain(provider, base, 'membership', action='add', group=group, user='fixture-user') except (ValueError, admin.Refused): pass else: raise RuntimeError('reserved_group_accepted') assert admin.snapshot(provider, token) == before checks['reserved_group_refusal_preserves_state'] = True preview = admin.maintain(provider, base, 'membership', action='add', group='activity-core-operators', user='fixture-user') provider._gql(token, 'mutation CreateGroup($name: String!) { createGroup(name: $name) { id displayName } }', {'name': 'unrelated-drift'}) try: admin.maintain(provider, base, 'membership', action='add', group='activity-core-operators', user='fixture-user', apply=True, expected_digest=preview['snapshot_sha256']) except admin.Refused: pass else: raise RuntimeError('stale_preview_accepted') assert 'fixture-user' not in admin.snapshot(provider, token)['groups']['activity-core-operators']['members'] checks['stale_preview_refused'] = True complete = subprocess.run(['bash', '-x', str(ROOT / 'sso-mfa/k8s/verify-t07.sh'), '--root', str(root), '--lldap-url', base], capture_output=True, text=True, timeout=30) assert complete.returncode == 2 assert password not in complete.stdout + complete.stderr assert json.loads(complete.stdout)['native_acceptance'] == 'unverified' checks['t07_structural_only_exit_and_no_trace_disclosure'] = True result['outcome'] = 'passed' except Exception: result['outcome'] = 'failed' finally: if container and re.fullmatch(r'[0-9a-f]{64}', container): try: command(['docker', 'rm', '--force', container]) checks['fixture_removed'] = True except Exception: checks['fixture_removed'] = False result['outcome'] = 'failed' Path(args.output).write_text(json.dumps(result, indent=2) + '\n') print(json.dumps(result)) return 0 if result['outcome'] == 'passed' else 1 if __name__ == '__main__': raise SystemExit(main())