#!/usr/bin/env python3 """Exercise Passport wire compatibility against a disposable copy of a PHP DB. Requires a bootable pinned PHP reference app (including vendor dependencies), its matching APP_KEY/RSA keys, and a freshly built Rust executable. Never changes its source database. Cookies, keys and bearer values are never printed. """ import argparse import http.cookies import json import os import pathlib import sqlite3 import subprocess import tempfile import time import urllib.error import urllib.request FIXTURE = r'''make(Illuminate\Contracts\Console\Kernel::class)->bootstrap(); $clients=app(Laravel\Passport\ClientRepository::class); try {$clients->personalAccessClient('users');} catch (RuntimeException $e) {$clients->createPersonalAccessGrantClient('2FAuth','users');} $user=App\Models\User::findOrFail((int)getenv('INTEROP_USER_ID')); // A synthetic fresh login also refreshes the application's inactivity clock. $user->last_seen_at=now(); $user->save(); $token=$user->createToken('php-generated-interop', []); $e=app('encrypter'); $csrf='php-interop-csrf-value-00001'; $jwt=app(Laravel\Passport\ApiTokenCookieFactory::class)->make($user->id,$csrf)->getValue(); $cookie=$e->encryptString(Illuminate\Cookie\CookieValuePrefix::create('laravel_token',$e->getKey()).$jwt); $xsrf=$e->encryptString(Illuminate\Cookie\CookieValuePrefix::create('XSRF-TOKEN',$e->getKey()).$csrf); file_put_contents(getenv('INTEROP_FIXTURE'),json_encode(['token'=>$token->accessToken,'id'=>$token->accessTokenId,'cookie'=>$cookie,'xsrf'=>$xsrf])); ''' VALIDATE = r'''make(Illuminate\Contracts\Console\Kernel::class)->bootstrap(); $token=json_decode(file_get_contents(getenv('INTEROP_ISSUED')),true)['accessToken']; $request=Symfony\Component\HttpFoundation\Request::create('/api/v1/twofaccounts','GET',[],[],[],['HTTP_AUTHORIZATION'=>'Bearer '.$token]); $psr=(new Symfony\Bridge\PsrHttpMessage\Factory\PsrHttpFactory)->createRequest($request); $validated=app(League\OAuth2\Server\ResourceServer::class)->validateAuthenticatedRequest($psr); echo json_encode(['php_validated_rust_bearer'=>(string)$validated->getAttribute('oauth_user_id')===getenv('INTEROP_USER_ID')]); ''' def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--php', required=True, type=pathlib.Path) parser.add_argument('--upstream', required=True, type=pathlib.Path) parser.add_argument('--reference-app', required=True, type=pathlib.Path) parser.add_argument('--reference-db', required=True, type=pathlib.Path) parser.add_argument('--candidate-binary', required=True, type=pathlib.Path) parser.add_argument('--user-id', default='1') parser.add_argument('--port', default=18081, type=int) args = parser.parse_args() if not os.environ.get('APP_KEY'): parser.error('APP_KEY must be set to the reference app key') with tempfile.TemporaryDirectory(prefix='twofauth-passport-interop-') as directory: work = pathlib.Path(directory) database = work / 'copy.sqlite' with sqlite3.connect(args.reference_db) as source, sqlite3.connect(database) as target: source.backup(target) target.executescript('CREATE TABLE IF NOT EXISTS rust_sessions (token_hash TEXT PRIMARY KEY,user_id INTEGER,csrf_hash TEXT NOT NULL,expires_at INTEGER NOT NULL);CREATE TABLE IF NOT EXISTS rust_login_attempts (key TEXT PRIMARY KEY,window INTEGER NOT NULL,attempts INTEGER NOT NULL);') fixture_path, issued_path = work / 'fixture.json', work / 'issued.json' fixture_php, validate_php = work / 'fixture.php', work / 'validate.php' fixture_php.write_text(FIXTURE) validate_php.write_text(VALIDATE) php_env = dict(os.environ, APP_BASE_PATH=str(args.reference_app.resolve()), APP_ENV='local', DB_CONNECTION='sqlite', DB_DATABASE=str(database), INTEROP_UPSTREAM=str(args.upstream.resolve()), INTEROP_USER_ID=args.user_id, INTEROP_FIXTURE=str(fixture_path), INTEROP_ISSUED=str(issued_path)) php_env.pop('DATABASE_URL', None) subprocess.run([str(args.php.resolve()), str(fixture_php)], env=php_env, check=True, capture_output=True) fixture = json.loads(fixture_path.read_text()) rust_env = dict(os.environ, DATABASE_URL='sqlite://' + str(database), PASSPORT_KEY_PATH=str(args.reference_app.resolve() / 'storage'), RUST_COOKIE_SECURE='false') with (work / 'server.log').open('w') as log: process = subprocess.Popen([str(args.candidate_binary.resolve()), 'serve', '--listen', f'127.0.0.1:{args.port}'], env=rust_env, stdout=log, stderr=subprocess.STDOUT) base = f'http://127.0.0.1:{args.port}' def request(path='/api/v1/twofaccounts', headers=None, method='GET', body=None, cookies=None): headers = dict(headers or {}) if cookies is not None: headers['Cookie'] = '; '.join(f'{k}={v}' for k, v in cookies.items()) data = None if body is None else json.dumps(body).encode() if data is not None: headers['Content-Type'] = 'application/json' req = urllib.request.Request(base + path, headers=headers, method=method, data=data) try: response = urllib.request.urlopen(req, timeout=10) except urllib.error.HTTPError as error: response = error if cookies is not None: for header in response.headers.get_all('Set-Cookie', []): parsed = http.cookies.SimpleCookie() parsed.load(header) for name, item in parsed.items(): if item['max-age'] == '0' or not item.value: cookies.pop(name, None) else: cookies[name] = item.value return response.status, response.read() try: for _ in range(100): try: request('/up') break except OSError: if process.poll() is not None: raise RuntimeError('candidate exited before readiness') time.sleep(.05) results = { 'php_bearer': request(headers={'Authorization': 'Bearer ' + fixture['token']})[0], 'php_cookie': request(headers={'Cookie': 'laravel_token=' + fixture['cookie'], 'X-XSRF-TOKEN': fixture['xsrf']})[0], 'cookie_missing_csrf': request(headers={'Cookie': 'laravel_token=' + fixture['cookie']})[0], 'tampered_bearer': request(headers={'Authorization': 'Bearer ' + fixture['token'][:-8] + 'AAAAAAAA'})[0], } cookies = {'laravel_token': fixture['cookie'], 'XSRF-TOKEN': fixture['xsrf']} results['php_browser_bootstrap'] = request('/', cookies=cookies)[0] results['legacy_cookie_cleared'] = 'laravel_token' not in cookies results['upgraded_browser_api'] = request(cookies=cookies, headers={'X-XSRF-TOKEN': cookies.get('XSRF-TOKEN', '')})[0] results['upgraded_browser_logout'] = request('/user/logout', cookies=cookies, headers={'X-XSRF-TOKEN': cookies.get('XSRF-TOKEN', '')})[0] results['browser_api_after_logout'] = request(cookies=cookies)[0] status, issued = request('/oauth/personal-access-tokens', headers={'Authorization': 'Bearer ' + fixture['token']}, method='POST', body={'name': 'rust-generated-interop'}) if status != 200: raise AssertionError(f'Rust PAT issuance returned {status}') issued_path.write_bytes(issued) validated = subprocess.run([str(args.php.resolve()), str(validate_php)], env=php_env, capture_output=True, text=True, check=True) results.update(json.loads(validated.stdout)) with sqlite3.connect(database) as connection: connection.execute('UPDATE oauth_access_tokens SET revoked=1 WHERE id=?', (fixture['id'],)) results['revoked_php_bearer'] = request(headers={'Authorization': 'Bearer ' + fixture['token']})[0] expected = {'php_bearer': 200, 'php_cookie': 200, 'cookie_missing_csrf': 401, 'tampered_bearer': 401, 'php_browser_bootstrap': 200, 'legacy_cookie_cleared': True, 'upgraded_browser_api': 200, 'upgraded_browser_logout': 200, 'browser_api_after_logout': 401, 'php_validated_rust_bearer': True, 'revoked_php_bearer': 401} print(json.dumps(results, indent=2)) if results != expected: raise AssertionError('Passport interoperability expectations failed') finally: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait() if __name__ == '__main__': main()