161 lines
12 KiB
Python
161 lines
12 KiB
Python
#!/usr/bin/env python3
|
|
"""Real PHP-issued device codes / Rust tokens on a disposable Laravel SQLite copy.
|
|
No source database is modified. Device codes, keys, secrets and tokens are not printed.
|
|
"""
|
|
import argparse
|
|
import base64
|
|
import hashlib
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import socket
|
|
import sqlite3
|
|
import subprocess
|
|
import tempfile
|
|
import time
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
PHP_SETUP = r'''<?php
|
|
require getenv('APP_BASE_PATH').'/vendor/autoload.php';
|
|
$app=require getenv('APP_BASE_PATH').'/bootstrap/app.php';
|
|
$app->make(Illuminate\Contracts\Console\Kernel::class)->bootstrap();
|
|
$user=new App\Models\User;
|
|
$user->name='Device fixture';$user->email='device@example.invalid';$user->password='Synthetic-device-password-4620';$user->save();
|
|
$clients=app(Laravel\Passport\ClientRepository::class);
|
|
$client=$clients->createAuthorizationCodeGrantClient('Device fixture',[],false,null,true);
|
|
$other=$clients->createAuthorizationCodeGrantClient('Other device fixture',[],false,null,true);
|
|
$server=app(League\OAuth2\Server\AuthorizationServer::class);
|
|
$request=(new GuzzleHttp\Psr7\ServerRequest('POST',getenv('APP_URL').'/oauth/device/code'))->withParsedBody(['client_id'=>(string)$client->id,'scope'=>'*']);
|
|
$response=$server->respondToDeviceAuthorizationRequest($request,new GuzzleHttp\Psr7\Response);
|
|
$code=json_decode((string)$response->getBody(),true);
|
|
echo json_encode(['client_id'=>(string)$client->id,'other_id'=>(string)$other->id,'user_id'=>(string)$user->id,'device_code'=>$code['device_code'],'user_code'=>$code['user_code']]);
|
|
'''
|
|
PHP_OPERATION = r'''<?php
|
|
require getenv('APP_BASE_PATH').'/vendor/autoload.php';
|
|
$app=require getenv('APP_BASE_PATH').'/bootstrap/app.php';
|
|
$app->make(Illuminate\Contracts\Console\Kernel::class)->bootstrap();
|
|
$input=json_decode(file_get_contents(getenv('DEVICE_CONTROL')),true);
|
|
if ($input['operation']==='approve' || $input['operation']==='deny') {
|
|
app(League\OAuth2\Server\AuthorizationServer::class)->completeDeviceAuthorizationRequest($input['device_code'],$input['user_id'],$input['operation']==='approve');
|
|
echo json_encode(['completed'=>true]);
|
|
} elseif ($input['operation']==='read') {
|
|
$model=Laravel\Passport\Passport::deviceCode()->findOrFail($input['device_code']);
|
|
$entity=app(Laravel\Passport\Bridge\DeviceCodeRepository::class)->getDeviceCodeEntityByDeviceCode($input['device_code']);
|
|
echo json_encode(['client_id'=>(string)$model->client_id,'user_code'=>$model->user_code,'scopes'=>$model->scopes,'revoked'=>(bool)$model->revoked,'entity_readable'=>$entity!==null]);
|
|
} elseif ($input['operation']==='validate') {
|
|
$request=(new GuzzleHttp\Psr7\ServerRequest('GET',getenv('APP_URL').'/api/v1/user'))->withHeader('Authorization','Bearer '.$input['access_token']);
|
|
$request=app(League\OAuth2\Server\ResourceServer::class)->validateAuthenticatedRequest($request);
|
|
echo json_encode(['user_id'=>(string)$request->getAttribute('oauth_user_id'),'scopes'=>$request->getAttribute('oauth_scopes')]);
|
|
}
|
|
'''
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument('binary', nargs='?', type=Path, default=ROOT/'target/debug/twofauth-rs')
|
|
parser.add_argument('--php', type=Path, default=Path(os.environ.get('PHP_BINARY','/tmp/twofauth-reference-runtime/php')))
|
|
parser.add_argument('--source', type=Path, default=ROOT.parent/'upstream')
|
|
parser.add_argument('--output', type=Path)
|
|
args=parser.parse_args()
|
|
source=args.source.resolve();php=args.php.resolve();binary=args.binary.resolve()
|
|
if not (source/'vendor/autoload.php').is_file():parser.error('pinned upstream Composer dependencies required')
|
|
with tempfile.TemporaryDirectory(prefix='device-php-rust-') as directory:
|
|
work=Path(directory);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 path in ('bootstrap/cache','storage/logs','storage/framework/cache/data','storage/framework/sessions','storage/framework/views'):(app/path).mkdir(parents=True,exist_ok=True)
|
|
with socket.socket() as reserve:reserve.bind(('127.0.0.1',0));port=reserve.getsockname()[1]
|
|
base=f'http://127.0.0.1:{port}'
|
|
runtime={name:os.environ[name] for name in ('PATH','PHPRC','LD_LIBRARY_PATH') if name in os.environ}
|
|
app_key='base64:'+base64.b64encode(b'0123456789abcdef0123456789abcdef').decode()
|
|
reference_db=work/'reference.sqlite';candidate_db=work/'candidate.sqlite'
|
|
env=dict(runtime,APP_BASE_PATH=str(app),APP_ENV='local',APP_DEBUG='false',APP_KEY=app_key,APP_URL=base,DB_CONNECTION='sqlite',DB_DATABASE=str(reference_db),SESSION_DRIVER='file',CACHE_STORE='file',MAIL_MAILER='array')
|
|
def command(argv,environment=env,cwd=app):
|
|
result=subprocess.run([str(value) for value in argv],env=environment,cwd=cwd,capture_output=True)
|
|
if result.returncode:raise RuntimeError('fixture command failed')
|
|
return result.stdout
|
|
command([php,'artisan','migrate','--force']);command([php,'artisan','passport:keys','--force'])
|
|
setup=work/'setup.php';setup.write_text(PHP_SETUP)
|
|
fixture=json.loads(command([php,setup]))
|
|
with sqlite3.connect(reference_db) as original,sqlite3.connect(candidate_db) as candidate:original.backup(candidate)
|
|
rust_env=dict(runtime,DATABASE_URL='sqlite://'+str(candidate_db),APP_KEY=app_key,APP_URL=base,PASSPORT_KEY_PATH=str(app/'storage'),RUST_COOKIE_SECURE='false',RUST_PUBLIC_DIR=str(ROOT/'public'))
|
|
command([binary,'prepare-sqlite'],rust_env,work)
|
|
operation=work/'operation.php';operation.write_text(PHP_OPERATION)
|
|
control=work/'control.json'
|
|
php_env=dict(env,DB_DATABASE=str(candidate_db),DEVICE_CONTROL=str(control))
|
|
def php_operation(name,**values):
|
|
control.write_text(json.dumps({'operation':name,**values}));control.chmod(0o600)
|
|
return json.loads(command([php,operation],php_env))
|
|
def request(path,values=None):
|
|
data=None if values is None else urllib.parse.urlencode(values).encode()
|
|
req=urllib.request.Request(base+path,data=data,headers={'Accept':'application/json','Content-Type':'application/x-www-form-urlencoded'})
|
|
try:response=urllib.request.urlopen(req,timeout=15)
|
|
except urllib.error.HTTPError as error:response=error
|
|
with response:
|
|
body=response.read()
|
|
try:payload=json.loads(body)
|
|
except ValueError:payload=None
|
|
return response.status,payload,response.headers
|
|
urn='urn:ietf:params:oauth:grant-type:device_code'
|
|
def poll(code,client=None):return request('/oauth/token',{'grant_type':urn,'client_id':client or fixture['client_id'],'device_code':code})
|
|
def issue():
|
|
status,body,headers=request('/oauth/device/code',{'client_id':fixture['client_id'],'scope':'*'})
|
|
assert status==200,'Rust device issuance failed'
|
|
assert re.fullmatch('[0-9a-f]{80}',body['device_code']) and re.fullmatch('[BCDFGHJKLMNPQRSTVWXZ]{8}',body['user_code'])
|
|
assert body['verification_uri']==base+'/oauth/device' and body['verification_uri_complete']==base+'/oauth/device?user_code='+body['user_code']
|
|
assert 590<=body['expires_in']<=600 and body['interval']==5
|
|
assert headers['Cache-Control']=='no-store, private' and headers['Pragma']=='no-cache'
|
|
return body
|
|
checks={}
|
|
with (work/'rust.log').open('w') as log:
|
|
process=subprocess.Popen([str(binary),'serve','--listen',f'127.0.0.1:{port}'],env=rust_env,cwd=work,stdout=log,stderr=subprocess.STDOUT)
|
|
try:
|
|
for _ in range(100):
|
|
if process.poll() is not None:raise RuntimeError('Rust exited before readiness')
|
|
try:
|
|
if request('/up')[0]==200:break
|
|
except OSError:time.sleep(.05)
|
|
else:raise RuntimeError('Rust startup timed out')
|
|
status,body,_=poll(fixture['device_code']);assert status==400 and body['error']=='authorization_pending' and body['hint']==''
|
|
status,body,_=poll(fixture['device_code']);assert status==400 and body['error']=='slow_down' and body['hint']==''
|
|
checks['php_issued_device_pending_backoff']=True
|
|
php_operation('approve',device_code=fixture['device_code'],user_id=fixture['user_id'])
|
|
status,tokens,_=poll(fixture['device_code']);assert status==200 and tokens.get('refresh_token'),'approved device exchange failed'
|
|
validated=php_operation('validate',access_token=tokens['access_token']);assert validated=={'user_id':fixture['user_id'],'scopes':[]}
|
|
checks['php_approval_rust_token_php_resource_server']=True
|
|
status,body,_=poll(fixture['device_code']);assert status==400 and body['error']=='invalid_grant' and body['hint']==''
|
|
assert php_operation('read',device_code=fixture['device_code'])['revoked'] is True
|
|
checks['device_code_revocation_replay']=True
|
|
issued=issue();read=php_operation('read',device_code=issued['device_code'])
|
|
assert read=={'client_id':fixture['client_id'],'user_code':issued['user_code'],'scopes':['*'],'revoked':False,'entity_readable':True}
|
|
checks['rust_issued_device_php_repository_readable']=True
|
|
status,body,_=poll(issued['device_code'],fixture['other_id']);assert status==400 and body['hint']=='Device code was not issued to this client'
|
|
php_operation('approve',device_code=issued['device_code'],user_id=fixture['user_id'])
|
|
status,tokens,_=poll(issued['device_code']);assert status==200,'PHP approval of Rust-issued device failed'
|
|
assert php_operation('validate',access_token=tokens['access_token'])=={'user_id':fixture['user_id'],'scopes':[]}
|
|
checks['rust_device_php_approval_and_token_validation']=True
|
|
denied=issue();php_operation('deny',device_code=denied['device_code'],user_id=fixture['user_id'])
|
|
status,body,_=poll(denied['device_code']);assert status==401 and body['error']=='access_denied' and 'hint' not in body
|
|
checks['php_denied_device_rejected']=True
|
|
expired=issue()
|
|
with sqlite3.connect(candidate_db) as db:db.execute("UPDATE oauth_device_codes SET expires_at=datetime('now','-1 second') WHERE id=?",(expired['device_code'],))
|
|
status,body,_=poll(expired['device_code']);assert status==400 and body['error']=='expired_token' and body['hint']=='device_code'
|
|
checks['expired_device_rejected']=True
|
|
with sqlite3.connect(reference_db) as db:
|
|
assert db.execute('SELECT revoked,last_polled_at FROM oauth_device_codes WHERE id=?',(fixture['device_code'],)).fetchone()==(0,None)
|
|
checks['source_database_unchanged']=True
|
|
record={'upstream_commit':'3516c776a1fcbd83d234efc007e3a49807d6ed91','candidate_sha256':hashlib.sha256(binary.read_bytes()).hexdigest(),'php_version':command([php,'-r','echo PHP_VERSION;']).decode().strip(),'checks':checks,'passed':all(checks.values())}
|
|
if args.output:args.output.write_text(json.dumps(record,indent=2)+'\n')
|
|
print(json.dumps(record))
|
|
finally:
|
|
process.terminate()
|
|
try:process.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:process.kill();process.wait()
|
|
|
|
if __name__=='__main__':main()
|