#!/usr/bin/env python3 """Isolated native reset, SMTP, issuer and app browser journey. No production or MFA. All peers share a network-none namespace. Only generated synthetic fixture credentials are used. Runtime assets enter private tmpfs through tar pipes. """ import argparse import hashlib import ipaddress import json import os import re import runpy import secrets import subprocess import tarfile import tempfile import uuid from datetime import UTC, datetime, timedelta from pathlib import Path from cryptography import x509 from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import rsa from cryptography.x509.oid import NameOID HERE = Path(__file__).resolve().parent ROOT = HERE.parent BASE = runpy.run_path(str(HERE / 'rehearse-native-password-reset.py')) docker, archive, container, probe = [BASE[n] for n in ['docker', 'archive', 'container', 'probe']] BROWSER = 'sha256:06a9939e57531807f8d5fd76ce44b53165ffb7d7501d87ab10e285c20b1e971f' ASSETS = HERE / 'native_recovery' APP_STAGE = runpy.run_path(str(ASSETS / 'app_stage.py')) def certificate(): key = rsa.generate_private_key(public_exponent=65537, key_size=2048) subject = x509.Name([x509.NameAttribute(NameOID.COMMON_NAME, 'auth.recovery.test')]) now = datetime.now(UTC) cert = (x509.CertificateBuilder().subject_name(subject).issuer_name(subject) .public_key(key.public_key()).serial_number(x509.random_serial_number()) .not_valid_before(now - timedelta(minutes=1)).not_valid_after(now + timedelta(hours=1)) .add_extension(x509.BasicConstraints(ca=True, path_length=0), critical=True) .add_extension(x509.SubjectAlternativeName([ x509.DNSName('auth.recovery.test'), x509.IPAddress(ipaddress.ip_address('127.0.0.1'))]), critical=False).sign(key, hashes.SHA256())) return cert.public_bytes(serialization.Encoding.PEM), key.private_bytes( serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()) def browser_assets(name): lock = json.loads((HERE / 'package-lock.json').read_text()) version = lock['packages']['node_modules/playwright']['version'] runtime = json.loads((HERE / 'node_modules/playwright/package.json').read_text())['version'] if version != runtime: raise RuntimeError('locked_playwright_required') spec = json.loads((HERE / 'node_modules/playwright-core/browsers.json').read_text()) revision = next(b['revision'] for b in spec['browsers'] if b['name'] == 'chromium-headless-shell') browser = Path.home() / '.cache/ms-playwright' / ('chromium_headless_shell-' + revision) roots = [(HERE / 'node_modules/playwright', 'node_modules/playwright'), (HERE / 'node_modules/playwright-core', 'node_modules/playwright-core'), (browser, 'browser')] manifest = {} with subprocess.Popen(['docker', '--context', 'default', 'exec', '-i', name, 'tar', '-xf', '-', '-C', '/run/fixture'], stdin=subprocess.PIPE, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, env={'PATH': os.environ['PATH']}) as process: with tarfile.open(fileobj=process.stdin, mode='w|') as tar: for root, prefix in roots: if root.is_symlink() or not root.is_dir(): raise RuntimeError('retained_runtime_assets_required') for path in sorted(root.rglob('*')): if path.is_symlink() or not (path.is_dir() or path.is_file()): raise RuntimeError('runtime_asset_links_refused') if path.is_dir(): continue member = prefix + '/' + path.relative_to(root).as_posix() digest = hashlib.sha256() with path.open('rb') as stream: while data := stream.read(1024 * 1024): digest.update(data) stream.seek(0) item = tarfile.TarInfo(member) item.size, item.mode = path.stat().st_size, path.stat().st_mode & 0o777 item.uid = item.gid = 1000 tar.addfile(item, stream) manifest[member] = digest.hexdigest() process.stdin.close() if process.wait(timeout=60): raise RuntimeError('runtime_asset_import_failed') return {'playwright_version': version, 'chromium_revision': revision, 'runtime_assets_sha256': hashlib.sha256(json.dumps( manifest, sort_keys=True).encode()).hexdigest(), 'runtime_asset_files': len(manifest)} def browser_container(name, anchor): docker('run', '-d', '--pull=never', '--name', name, '--network', 'container:' + anchor, '--read-only', '--cap-drop', 'ALL', '--security-opt', 'no-new-privileges', '--user', '1000:1000', '--memory', '1536m', '--cpus', '1', '--pids-limit', '256', '--env', 'HOME=/tmp', '--tmpfs', '/run/fixture:rw,exec,nosuid,nodev,size=805306368,uid=1000,gid=1000,mode=0700', '--tmpfs', '/tmp:rw,noexec,nosuid,nodev,size=536870912,uid=1000,gid=1000,mode=0700', '--entrypoint', '/bin/sh', BROWSER, '-c', 'sleep 1800') BASE['confined'](name, 'container:' + anchor) def peers(anchor, value): done = docker('run', '-i', '--rm', '--name', anchor + '-probe', '--pull=never', '--network', 'container:' + anchor, '--read-only', '--cap-drop', 'ALL', '--security-opt', 'no-new-privileges', '--memory', '128m', '--cpus', '1', '--entrypoint', 'python', BASE['CLIENT'], '-c', (ASSETS / 'check_peers.py').read_text(), data=json.dumps(value).encode(), check=False) if done.returncode: raise RuntimeError('native_browser_peer_check_failed') return json.loads(done.stdout) 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'], BROWSER]: assert json.loads(docker('image', 'inspect', image).stdout)[0]['Id'] == image name = 'csoc-native-browser-' + uuid.uuid4().hex[:12] auth, mail, browser = [name + suffix for suffix in ['-auth', '-mail', '-browser']] names = [name + '-probe', browser, auth, mail, name + '-app', name + '-portal', name + '-extract', name + '-hasher', name] values = {n: secrets.token_urlsafe(24) for n in [ 'admin_password', 'bind_password', 'old_password', 'new_password', 'expired_password']} result = {'scenarios': []} cert, key = certificate() 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'}) for user, password in [('alice', values['old_password']), ('svc-auth', values['bind_password'])]: docker('exec', '-i', name, '/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 ' + user, data=(prepared['admin_token'] + '\n' + password + '\n').encode()) container(mail, BASE['CLIENT'], 'container:' + name, {}) docker('exec', '-i', mail, 'tar', '-xf', '-', '-C', '/run/fixture', data=archive({ 'mail_sink.py': (ASSETS / 'mail_sink.py').read_bytes()})) docker('exec', '-d', mail, '/bin/sh', '-c', 'exec python /run/fixture/mail_sink.py > /run/fixture/server.log 2>&1') browser_container(browser, name) result.update(browser_assets(browser)) for case, role in [('read_only_refusal', 'lldap_strict_readonly'), ('successful_reset', 'lldap_password_manager'), ('expired_handle', 'lldap_password_manager')]: assigned = probe(name, {**values, 'mode': 'bind_role', 'role': role}) config = BASE['configuration'](values['bind_password']) config['server']['tls'] = {'certificate': '/run/fixture/cert.pem', 'key': '/run/fixture/key.pem'} config['session']['cookies'][0]['authelia_url'] = 'https://auth.recovery.test:9091' if case == 'expired_handle': config['identity_validation']['reset_password']['jwt_lifespan'] = '1s' config['notifier'] = {'disable_startup_check': True, 'smtp': { 'address': 'smtp://127.0.0.1:2525', 'sender': 'no-reply@example.test', 'identifier': 'auth.recovery.test', 'disable_require_tls': True, 'disable_starttls': True}} container(auth, BASE['AUTHELIA'], 'container:' + name, {}) docker('exec', '-i', auth, 'tar', '-xf', '-', '-C', '/run/fixture', data=archive({ 'configuration.yml': json.dumps(config).encode(), 'cert.pem': cert, 'key.pem': key})) docker('exec', '-d', auth, '/bin/sh', '-c', 'exec /app/authelia --config /run/fixture/configuration.yml ' '> /run/fixture/server.log 2>&1') ready = peers(name, {'mode': 'ready', 'certificate': cert.decode()}) docker('exec', '-i', browser, 'tar', '-xf', '-', '-C', '/run/fixture', data=archive({ 'reset_browser.mjs': (ASSETS / 'reset_browser.mjs').read_bytes(), 'browser-input.json': json.dumps({ 'new_password': values['new_password'], 'expect_expired': case == 'expired_handle', 'expected_success': case == 'successful_reset'}).encode()})) checked = docker('exec', browser, 'node', '/run/fixture/reset_browser.mjs', check=False) browser_result = json.loads(checked.stdout) if checked.returncode: raise RuntimeError('browser_' + browser_result.get('stage', 'failed')) observed_values = (values if case != 'expired_handle' else { **values, 'old_password': values['new_password'], 'new_password': values['expired_password']}) directory = peers(name, {**observed_values, 'mode': 'directory_check', 'certificate': cert.decode()}) assert directory['new_password_accepted'] is (case == 'successful_reset') assert directory['old_password_accepted'] is (case != 'successful_reset') result['scenarios'].append({'case': case, 'bind_role': role, **ready, **browser_result, **directory, 'directory_group_creation_refused': assigned['directory_group_creation_refused']}) docker('rm', '-f', '-v', auth) result['app_journey'] = APP_STAGE['run'](name, browser, values, cert, key, peers) except Exception: fd, diagnostic = tempfile.mkstemp(prefix='csoc-native-browser-diagnostic-', suffix='.txt') with os.fdopen(fd, 'w') as stream: for current, filename in [(browser, 'browser-diagnostic.txt'), (browser, 'issuer.log'), (auth, 'server.log'), (mail, 'server.log'), (name + '-portal', 'server.log'), (name + '-app', 'server.log')]: text = docker('exec', current, 'cat', '/run/fixture/' + filename, check=False).stdout.decode(errors='replace') for value in values.values(): text = text.replace(value, '') text = re.sub(r'https?://[^\s\"\']+', '', text) text = re.sub(r'[A-Za-z0-9_+/=-]{32,}', '', text) stream.write(current + '\n' + text + '\n') print('Private redacted fixture diagnostic: ' + diagnostic) raise finally: for current in names: docker('rm', '-f', '-v', current, check=False) assert all(not docker('ps', '-aq', '--filter', 'name=^/' + n + '$').stdout.strip() for n in names), 'fixture_cleanup' result['cleanup_verified'] = True assert not any(value in json.dumps(result) for value in values.values()) return result if __name__ == '__main__': parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--output', type=Path, required=True) args = parser.parse_args() source_path = args.output.with_suffix('.sources.txt') if args.output.exists() or source_path.exists(): raise SystemExit('Fresh receipt and source manifest required') findings = rehearse() sources = [Path(__file__), HERE / 'rehearse-native-password-reset.py', ASSETS / 'probe.py', ASSETS / 'mail_sink.py', ASSETS / 'check_peers.py', ASSETS / 'reset_browser.mjs', ASSETS / 'app_browser.mjs', ASSETS / 'app_stage.py', ASSETS / 'user_portal.py', HERE / 'package-lock.json', ASSETS / 'test_journey.py', HERE / 'check-native-recovery.py'] manifest = ''.join(str(p.relative_to(ROOT)) + ': ' + hashlib.sha256(p.read_bytes()).hexdigest() + '\n' for p in sources).encode() receipt = {'schema': 'coulomb.native-account-recovery-browser.v1', 'status': 'isolated_native_recovery_browser_verified', 'observed_at': datetime.now(UTC).isoformat(), 'data': 'synthetic', 'production_access': False, 'provider_smtp_contacted': False, 'native_oidc_and_app_journey_verified': True, 'production_or_mfa_acceptance': False, 'fixture_tls_browser_errors_ignored': True, 'images': {'authelia': BASE['AUTHELIA'], 'lldap': BASE['LLDAP'], 'probe': BASE['CLIENT'], 'browser': BROWSER, 'issuer': APP_STAGE['ISSUER_IMAGE'], 'user_portal': APP_STAGE['USER_IMAGE']}, 'source_manifest_sha256': hashlib.sha256(manifest).hexdigest(), 'findings': findings} with source_path.open('xb') as stream: stream.write(manifest) with args.output.open('x') as stream: stream.write(json.dumps(receipt, indent=2) + '\n') print(json.dumps(findings))