#!/usr/bin/env python3 """Measure reset rights on disposable native peers; never contact production.""" import argparse import hashlib import json import re import runpy import secrets import urllib.parse import uuid from datetime import UTC, datetime from pathlib import Path HERE = Path(__file__).resolve().parent BASE = runpy.run_path(str(HERE / 'rehearse-native-password-reset.py')) docker, container, archive, probe, configuration = ( BASE[n] for n in ('docker', 'container', 'archive', 'probe', 'configuration')) def set_password(anchor, token, target, password): assert target in {'alice', 'bob', 'operator', 'appoperator', 'svc-other', 'svc-auth'} return docker('exec', '-i', anchor, '/bin/sh', '-c', 'read -r token; read -r LLDAP_USER_PASSWORD; export LLDAP_USER_PASSWORD; ' 'exec /app/lldap_set_password --base-url http://127.0.0.1:17170 ' '--token "$token" --username ' + target, data=(token + '\n' + password + '\n').encode(), check=False).returncode def rehearse(): if not __debug__: raise RuntimeError('acceptance_assertions_must_be_enabled') context = json.loads(docker('context', 'inspect', 'default').stdout)[0] assert context['Endpoints']['docker']['Host'] == 'unix:///var/run/docker.sock' for image in (BASE['LLDAP'], BASE['AUTHELIA'], BASE['CLIENT']): assert json.loads(docker('image', 'inspect', image).stdout)[0]['Id'] == image name = 'csoc-reset-authority-' + uuid.uuid4().hex[:12] auth = name + '-auth' values = {key: secrets.token_urlsafe(24) for key in ( 'admin_password', 'bind_password', 'old_password', 'new_password')} targets = ['alice', 'bob', 'operator', 'appoperator', 'svc-other'] result = {'scenarios': []} try: container(name, BASE['LLDAP'], 'none', { 'LLDAP_JWT_SECRET_FILE': '/run/fixture/jwt', 'LLDAP_LDAP_USER_PASS_FILE': '/run/fixture/admin-password', 'LLDAP_LDAP_BASE_DN': 'dc=example,dc=test', 'LLDAP_LDAP_HOST': '127.0.0.1', 'LLDAP_HTTP_HOST': '127.0.0.1'}) docker('exec', '-i', name, 'tar', '-xf', '-', '-C', '/run/fixture', data=archive({ 'jwt': secrets.token_hex(32).encode(), 'admin-password': values['admin_password'].encode(), 'lldap.toml': ('database_url = "sqlite:///data/users.db?mode=rwc"\nkey_seed = ' + json.dumps(secrets.token_hex(32)) + '\n').encode()})) docker('exec', '-d', name, '/bin/sh', '-c', 'exec /app/lldap run --config-file /run/fixture/lldap.toml ' '> /run/fixture/server.log 2>&1') prepared = probe(name, {**values, 'mode': 'bootstrap', 'authority_matrix': True}) assert set_password(name, prepared['admin_token'], 'svc-auth', values['bind_password']) == 0 for role in ['lldap_strict_readonly', 'lldap_password_manager']: probe(name, {**values, 'mode': 'bind_role', 'role': role}) context = probe(name, {**values, 'mode': 'authority_context'}) writes = probe(name, {**values, 'mode': 'write_boundaries'}) scenario = {'bind_role': role, 'directory_writes': writes, 'direct_password_changes': [], 'authelia_resets': []} for target in targets: assert set_password(name, context['admin_token'], target, values['old_password']) == 0 before = probe(name, {**values, 'mode': 'directory_authentication', 'target': target}) assert before == {'old_password_accepted': True, 'new_password_accepted': False} code = set_password(name, context['service_token'], target, values['new_password']) after = probe(name, {**values, 'mode': 'directory_authentication', 'target': target}) expected = role == 'lldap_password_manager' and target != 'operator' assert (code == 0) is expected, 'unexpected_direct_password_right' assert after == {'old_password_accepted': not expected, 'new_password_accepted': expected} scenario['direct_password_changes'].append({ 'target': target, 'password_changed': expected, **after}) # Separate fresh stores make reset/link checks independent of earlier scenarios. for target in ['alice', 'bob', 'operator', 'appoperator']: assert set_password(name, context['admin_token'], target, values['old_password']) == 0 container(auth, BASE['AUTHELIA'], 'container:' + name, {}) docker('exec', '-i', auth, 'tar', '-xf', '-', '-C', '/run/fixture', data=archive({ 'configuration.yml': json.dumps( configuration(values['bind_password'])).encode()})) docker('exec', '-d', auth, '/bin/sh', '-c', 'exec /app/authelia --config /run/fixture/configuration.yml ' '> /run/fixture/server.log 2>&1') requested = probe(name, {**values, 'mode': 'reset_request', 'target': target}) notification = docker( 'exec', auth, 'cat', '/data/notifications.txt').stdout.decode() links = re.findall(r'https://auth\.recovery\.test/reset-password/step2\?[^\s<>]+', notification) assert len(links) == 1, 'one_native_challenge_required' token = urllib.parse.parse_qs(urllib.parse.urlsplit(links[0]).query)['token'][0] settled = probe(name, {**values, 'mode': 'reset_finish', 'target': target, 'reset_token': token}) expected = role == 'lldap_password_manager' and target != 'operator' assert settled['new_password_accepted_after'] is expected assert settled['old_password_accepted_after'] is not expected assert (settled['status'] == 'reset_completed') is expected scenario['authelia_resets'].append({'target': target, **requested, **settled}) docker('rm', '-f', '-v', auth) result['scenarios'].append(scenario) finally: for current in [name + '-probe', auth, name]: docker('rm', '-f', '-v', current, check=False) result['cleanup_verified'] = all(not docker( 'ps', '-aq', '--filter', 'name=^/' + n + '$').stdout.strip() for n in [name + '-probe', auth, name]) if not result['cleanup_verified']: raise RuntimeError('fixture_cleanup_requires_review') assert not any(value in json.dumps(result) for value in values.values()) return result def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--output', type=Path, required=True) args = parser.parse_args() manifest_path = args.output.with_suffix('.sources.txt') if args.output.exists() or manifest_path.exists(): raise SystemExit('Fresh receipt and source manifest required') findings = rehearse() paths = [Path(__file__), HERE / 'rehearse-native-password-reset.py', HERE / 'native_recovery/probe.py'] manifest = ''.join(str(p.relative_to(HERE.parent)) + ': ' + hashlib.sha256(p.read_bytes()).hexdigest() + '\n' for p in paths).encode() receipt = {'schema': 'coulomb.native-reset-authority-matrix.v1', 'observed_at': datetime.now(UTC).isoformat(), 'data': 'synthetic', 'status': 'native_authority_matrix_verified', 'images': {n: BASE[n.upper()] for n in ('lldap', 'authelia', 'client')}, 'production_access': False, 'provider_smtp_contacted': False, 'owner_runtime_or_permission_change_performed': False, 'no_ports_host_mounts_or_external_routes': True, 'backend_container_flags_verified_before_fixture_import': True, 'source_manifest_sha256': hashlib.sha256(manifest).hexdigest(), 'target_classes': {'alice': 'ordinary user', 'bob': 'other ordinary user', 'operator': 'lldap_admin member', 'appoperator': 'application-admins member, not lldap_admin', 'svc-other': 'other service, lldap_strict_readonly'}, 'findings': findings} with manifest_path.open('xb') as stream: stream.write(manifest) with args.output.open('x') as stream: json.dump(receipt, stream, indent=2) stream.write('\n') print(json.dumps({'status': receipt['status'], 'roles': len(findings['scenarios']), 'cleanup_verified': findings['cleanup_verified']})) if __name__ == '__main__': main()