coulomb-social/scripts/rehearse-reset-role-refresh.py

#!/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()