#!/usr/bin/env python3 """Verify role changes with one running Authelia process; synthetic peers only.""" 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-transition-' + uuid.uuid4().hex[:12] auth = name + '-auth' values = {key: secrets.token_urlsafe(24) for key in ( 'admin_password', 'bind_password', 'old_password', 'new_password')} 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 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()})) # One Authelia process for all three phases, including rollback. probe(name, {**values, 'mode': 'bind_role', 'role': 'lldap_strict_readonly'}) docker('exec', '-d', auth, '/bin/sh', '-c', 'exec /app/authelia --config /run/fixture/configuration.yml ' '> /run/fixture/server.log 2>&1') process_marker = None tokens = set() for index, role in enumerate(['lldap_strict_readonly', 'lldap_password_manager', 'lldap_strict_readonly']): if index: probe(name, {**values, 'mode': 'bind_role', 'role': role}) assert set_password(name, prepared['admin_token'], 'alice', values['old_password']) == 0 docker('exec', auth, 'rm', '-f', '/data/notifications.txt') requested = probe(name, {**values, 'mode': 'reset_request', 'target': 'alice'}) marker = docker('exec', auth, '/bin/sh', '-c', 'for p in /proc/[0-9]*; do ' 'if [ "$(cat "$p/comm" 2>/dev/null)" = authelia ]; ' 'then cat "$p/stat"; fi; done').stdout # /proc stat changes CPU counters; compare PID and start ticks only. fields = marker.decode().split() current_marker = (fields[0], fields[21]) if process_marker is None: process_marker = current_marker assert current_marker == process_marker, 'authelia_process_restarted' 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_new_native_challenge_required_phase_' + str(index) + '_count_' + str(len(links))) token = urllib.parse.parse_qs(urllib.parse.urlsplit(links[-1]).query)['token'][0] assert token not in tokens, 'fresh_challenge_required' tokens.add(token) settled = probe(name, {**values, 'mode': 'reset_finish', 'target': 'alice', 'reset_token': token}) expected = role == 'lldap_password_manager' assert settled['new_password_accepted_after'] is expected assert settled['old_password_accepted_after'] is not expected assert (settled['status'] == 'reset_completed') is expected result['scenarios'].append({'bind_role': role, 'same_authelia_process': True, **requested, **settled}) 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-role-refresh.v1', 'observed_at': datetime.now(UTC).isoformat(), 'data': 'synthetic', 'status': 'native_role_refresh_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(), '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()