coulomb-social/scripts/rehearse-native-reset-browser.py

#!/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, '<fixture-value>')
                text = re.sub(r'https?://[^\s\"\']+', '<fixture-url>', text)
                text = re.sub(r'[A-Za-z0-9_+/=-]{32,}', '<redacted>', 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))