#!/usr/bin/env python3 """Run HTTP comparisons using fresh, disposable PHP and Rust fixture databases. Requires PHP 8.4+ with required extensions and Composer-installed upstream/vendor. No production configuration is copied. No response bodies or cookies are printed. """ import argparse import hashlib import importlib.util import json import os from pathlib import Path import shutil import socket import sqlite3 import subprocess import sys import tempfile import time import urllib.request ROOT = Path(__file__).resolve().parents[1] def command(argv, env, cwd): result = subprocess.run([str(x) for x in argv], env=env, cwd=cwd, capture_output=True) if result.returncode: raise RuntimeError('fixture_command_failed') return result.stdout.decode(errors='replace') def port(): with socket.socket() as sock: sock.bind(('127.0.0.1', 0)) return sock.getsockname()[1] def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--php', required=True, type=Path) parser.add_argument('--source', type=Path, default=ROOT.parent / 'upstream') parser.add_argument('--candidate-binary', type=Path, default=ROOT / 'target/release/twofauth-rs') parser.add_argument('--output', type=Path) parser.add_argument('--candidate-db', choices=('native', 'laravel-copy'), default='native', help='native init or SQLite backup of real migrated PHP fixture') parser.add_argument('--timeout', type=float, default=30) parser.add_argument('--semantic-checks', action='store_true', help='also verify health timing, CSRF login and rendered QR content; QR needs Node sharp') parser.add_argument('--passport-interop', action='store_true', help='also exercise PHP-issued and Rust-issued Passport tokens against the disposable migrated database') parser.add_argument('--extra-cases', type=Path, action='append', default=[], help='additional stateful JSON case files to compare using the disposable fixtures') args = parser.parse_args() php, source, binary = (p.resolve() for p in (args.php, args.source, args.candidate_binary)) if not (source / 'vendor/autoload.php').is_file(): parser.error('install the pinned upstream Composer dependencies first') processes = [] try: with tempfile.TemporaryDirectory(prefix='twofauth-reference-') as directory: work = Path(directory) candidate_source = str(binary) shutil.copy2(binary, work / 'candidate-binary') binary = work / 'candidate-binary' app = work / 'reference' shutil.copytree(source, app, ignore=shutil.ignore_patterns('vendor', 'node_modules', '.env*', 'cache', 'logs', 'sessions', 'testing')) (app / 'vendor').symlink_to(source / 'vendor', target_is_directory=True) for subdir in ('bootstrap/cache', 'storage/logs', 'storage/framework/cache/data', 'storage/framework/sessions', 'storage/framework/views'): (app / subdir).mkdir(parents=True, exist_ok=True) reference_port, candidate_port = port(), port() while candidate_port == reference_port: candidate_port = port() reference_base = f'http://127.0.0.1:{reference_port}' candidate_base = f'http://127.0.0.1:{candidate_port}' # Start with an explicit environment so deployment secrets/config cannot leak into fixtures. runtime = {name: os.environ[name] for name in ('PATH', 'PHPRC', 'PHP_INI_SCAN_DIR', 'LD_LIBRARY_PATH', 'SYSTEMROOT') if name in os.environ} reference_env = dict(runtime, APP_BASE_PATH=str(app), APP_ENV='local', APP_NAME='2FAuth', APP_KEY='base64:AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=', APP_URL=reference_base, DB_CONNECTION='sqlite', DB_DATABASE=str(work / 'reference.sqlite'), SESSION_DRIVER='file', SESSION_SECURE_COOKIE='false', CACHE_STORE='file', MAIL_MAILER='array', THROTTLE_API='1000') candidate_env = dict(runtime, APP_URL=candidate_base, APP_KEY=reference_env['APP_KEY'], THROTTLE_API='1000', DATABASE_URL='sqlite://' + str(work / 'candidate.sqlite'), RUST_COOKIE_SECURE='false', RUST_PUBLIC_DIR=str(ROOT / 'public')) php_version = command([php, '-r', 'echo PHP_VERSION;'], reference_env, app) command([php, 'artisan', 'migrate', '--force'], reference_env, app) command([php, 'artisan', 'passport:keys', '--force'], reference_env, app) fixture = work / 'fixture.php' fixture.write_text('''make(Illuminate\\Contracts\\Console\\Kernel::class)->bootstrap(); foreach ([['Compat','compat@example.invalid',false],['Admin','admin@example.invalid',true],['Recipient','recipient@example.invalid',false]] as [$name,$email,$admin]) { $user = new App\\Models\\User; $user->name = $name; $user->email = $email; $user->password = 'Synthetic-compat-password-4620'; $user->is_admin = $admin; $user->save(); } ''') command([php, fixture], reference_env, app) if args.candidate_db == 'laravel-copy': with sqlite3.connect(work / 'reference.sqlite') as reference_db, sqlite3.connect(work / 'candidate.sqlite') as candidate_db: reference_db.backup(candidate_db) candidate_env['APP_KEY'] = reference_env['APP_KEY'] command([binary, 'prepare-sqlite'], candidate_env, work) else: command([binary, 'init'], candidate_env, work) password = work / 'password' password.write_text('Synthetic-compat-password-4620') password.chmod(0o600) for name, email in (('Compat', 'compat@example.invalid'), ('Admin', 'admin@example.invalid'), ('Recipient', 'recipient@example.invalid')): command([binary, 'create-user', '--name', name, '--email', email, '--password-file', password], candidate_env, work) with sqlite3.connect(work / 'candidate.sqlite') as database: database.execute("UPDATE users SET is_admin=1 WHERE email='admin@example.invalid'") password.unlink() commands = [([php, '-S', f'127.0.0.1:{reference_port}', '-t', app / 'public', source / 'vendor/laravel/framework/src/Illuminate/Foundation/resources/server.php'], reference_env, app / 'public'), ([binary, 'serve', '--listen', f'127.0.0.1:{candidate_port}'], candidate_env, work)] for argv, env, cwd in commands: processes.append(subprocess.Popen([str(x) for x in argv], env=env, cwd=cwd, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)) for base, process in zip((reference_base, candidate_base), processes): for _ in range(100): if process.poll() is not None: raise RuntimeError('server_start_failed') try: with urllib.request.urlopen(base + '/up', timeout=1): break except Exception: time.sleep(.05) else: raise RuntimeError('server_start_timeout') record = {'candidate_source': candidate_source, 'candidate_database_mode': args.candidate_db, 'reference_php': php_version, 'candidate_sha256': hashlib.sha256(binary.read_bytes()).hexdigest(), 'database': 'fresh disposable SQLite; upstream migrations applied', 'fixture_config': {'THROTTLE_API': 1000, 'note': 'bulk fixture sequence; does not establish default throttle equivalence'}, 'composer_lock_sha256': hashlib.sha256((source / 'composer.lock').read_bytes()).hexdigest(), 'harness_sha256': hashlib.sha256((ROOT / 'tools/differential_http.py').read_bytes()).hexdigest(), 'results': {}} suites = [('default', None)] + [(name, ROOT / f'compatibility/cases.{name}.json') for name in ('stateful', 'extended', 'admin-sharing')] def compare_suite(name, cases_path): argv = [sys.executable, str(ROOT / 'tools/differential_http.py'), '--reference', reference_base, '--candidate', candidate_base, '--timeout', str(args.timeout)] if cases_path is not None: argv += ['--cases', str(cases_path), '--sessions', '--allow-side-effects'] result = subprocess.run(argv, capture_output=True, text=True) if result.returncode not in (0, 1): raise RuntimeError('comparison_failed') record['results'][name] = {'exit_code': result.returncode, 'cases': [json.loads(line) for line in result.stdout.splitlines()]} if cases_path is not None: record['results'][name]['cases_sha256'] = hashlib.sha256(cases_path.read_bytes()).hexdigest() for name, cases_path in suites: compare_suite(name, cases_path) if args.output: args.output.write_text(json.dumps(record, indent=2) + '\n') if args.semantic_checks: spec = importlib.util.spec_from_file_location('semantic_http', ROOT / 'tools/semantic_http.py') semantic = importlib.util.module_from_spec(spec) spec.loader.exec_module(semantic) record['semantic_results'] = semantic.check(reference_base, candidate_base, php, source, work, reference_env, args.timeout) checks = record['semantic_results'] record['semantic_pass'] = all(checks['health_render_timing'].values()) and all(checks['csrf_token_flow'].values()) and checks['qr_payload'].get('result') == 'decoded' and checks['qr_payload'].get('decoded_uri_equal') and checks['qr_payload'].get('matches_fixture') # OAuth uses Laravel's unnamed web throttle, shared with password login. # Run login/QR semantics before the extra token-error burst consumes it. record['extra_cases_order'] = 'after core suites and supplemental semantic checks' for index, path in enumerate(args.extra_cases): compare_suite('extra-' + str(index), path.resolve()) if args.passport_interop: interop_env = dict(reference_env, RUST_PUBLIC_DIR=str(ROOT / 'public')) interop = subprocess.run([sys.executable, str(ROOT / 'tools/passport_interop.py'), '--php', str(php), '--upstream', str(source), '--reference-app', str(app), '--reference-db', str(work / 'reference.sqlite'), '--candidate-binary', str(binary), '--port', str(port())], env=interop_env, capture_output=True, text=True) record['passport_interop'] = {'exit_code': interop.returncode} if interop.returncode == 0: record['passport_interop']['checks'] = json.loads(interop.stdout) # Stop the servers before their temporary application files are removed. for process in processes: process.terminate() process.wait(timeout=10) serialized = json.dumps(record, indent=2) + '\n' if args.output: args.output.write_text(serialized) print(serialized, end='') return int(any(result['exit_code'] for result in record['results'].values()) or (args.semantic_checks and not record.get('semantic_pass', False)) or record.get('passport_interop', {}).get('exit_code', 0)) except Exception as error: print(json.dumps({'error': type(error).__name__}), file=sys.stderr) return 2 finally: for process in processes: if process.poll() is None: process.kill() process.wait() if __name__ == '__main__': raise SystemExit(main())