net-kingdom/tests/lldap_groups_native_fixture.py

#!/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())