147 lines
9.2 KiB
Python
147 lines
9.2 KiB
Python
#!/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'''<?php
|
|
require getenv('INTEROP_UPSTREAM').'/vendor/autoload.php';
|
|
$app=require getenv('INTEROP_UPSTREAM').'/bootstrap/app.php';
|
|
$app->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'''<?php
|
|
require getenv('INTEROP_UPSTREAM').'/vendor/autoload.php';
|
|
$app=require getenv('INTEROP_UPSTREAM').'/bootstrap/app.php';
|
|
$app->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()
|