security: repair LDAP group callers and fail incomplete T07 acceptance

Assistant: codex
Assistant-Model: gpt-6-astra
Assistant-Session: 01a1346b-72a4-7d40-b8bd-8b8ca346cfa2

tests/lldap_groups_native_fixture.py

@@ -0,0 +1,141 @@
+#!/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())