#!/usr/bin/env python3 """Disposable OAuth integration test against the pinned PHP cryptographic wire format.""" import base64 import hashlib import json import os from pathlib import Path import socket import sqlite3 import subprocess import sys import tempfile import time import urllib.error import urllib.parse import urllib.request ROOT=Path(__file__).resolve().parents[1] PHP=os.environ.get('PHP_BINARY','/tmp/twofauth-reference-runtime/bin/php') AUTOLOAD=ROOT.parent/'upstream/vendor/autoload.php' def php(code,*args): return subprocess.check_output([PHP,'-r',f'require {json.dumps(str(AUTOLOAD))};'+code,*args],text=True).strip() def main(): binary=Path(sys.argv[1] if len(sys.argv)>1 else ROOT/'target/debug/twofauth-rs').resolve() with tempfile.TemporaryDirectory(prefix='oauth-wire-') as d: work=Path(d) subprocess.run(['openssl','genrsa','-out',str(work/'oauth-private.key'),'2048'],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) subprocess.run(['openssl','rsa','-in',str(work/'oauth-private.key'),'-pubout','-out',str(work/'oauth-public.key')],check=True,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) app_key='0123456789abcdef0123456789abcdef' env=dict(os.environ,DATABASE_URL=f'sqlite://{work}/test.sqlite',APP_KEY='base64:'+base64.b64encode(app_key.encode()).decode(),PASSPORT_KEY_PATH=d,RUST_COOKIE_SECURE='false',RUST_PUBLIC_DIR=str(ROOT/'public')) env.pop('APP_SUBDIRECTORY',None) subprocess.run([str(binary),'init'],check=True,env=env,cwd=work,capture_output=True) with socket.socket() as sock: sock.bind(('127.0.0.1',0));port=sock.getsockname()[1] proc=subprocess.Popen([str(binary),'serve','--listen',f'127.0.0.1:{port}'],env=env,cwd=work,stdout=subprocess.DEVNULL,stderr=subprocess.DEVNULL) def request(values,path='/oauth/token'): data=urllib.parse.urlencode(values).encode() if values is not None else None req=urllib.request.Request(f'http://127.0.0.1:{port}'+path,data=data,headers={'Content-Type':'application/x-www-form-urlencoded','Accept':'application/json'}) try:r=urllib.request.urlopen(req,timeout=10) except urllib.error.HTTPError as e:r=e with r:return r.status,json.loads(r.read()) try: for _ in range(100): try: if request({'grant_type':'bogus'})[1].get('error')=='unsupported_grant_type':break except (OSError,ValueError):time.sleep(.05) else:raise AssertionError('OAuth route unavailable') con=sqlite3.connect(work/'test.sqlite') secret=php('echo password_hash("secret",PASSWORD_BCRYPT);') con.execute("INSERT INTO users(id,name,email,password) VALUES(1,'OAuth','oauth@example.invalid','unused')") con.execute("INSERT INTO oauth_clients(id,name,secret,redirect_uris,grant_types) VALUES(1,'public',NULL,'[\"https://example.test/cb\"]','[\"authorization_code\",\"refresh_token\"]')") con.execute("INSERT INTO oauth_clients(id,name,secret,redirect_uris,grant_types) VALUES(2,'machine',?,'[]','[\"client_credentials\"]')",(secret,)) verifier='a'*43 payload={'client_id':'1','auth_code_id':'php-auth-code','user_id':'1','scopes':[],'redirect_uri':'https://example.test/cb','expire_time':int(time.time())+600,'code_challenge':base64.urlsafe_b64encode(hashlib.sha256(verifier.encode()).digest()).decode().rstrip('='),'code_challenge_method':'S256'} code=php('echo Defuse\\Crypto\\Crypto::encryptWithPassword($argv[1],$argv[2]);',json.dumps(payload),app_key) con.execute("INSERT INTO oauth_auth_codes(id,user_id,client_id,scopes,revoked,expires_at) VALUES('php-auth-code',1,1,'[]',0,datetime('now','+10 minutes'))") con.commit() params={'grant_type':'authorization_code','client_id':'1','redirect_uri':payload['redirect_uri'],'code':code,'code_verifier':'b'*43} assert request(params)[1]['error']=='invalid_grant' params['code_verifier']=verifier status,tokens=request(params);assert status==200,(status,tokens) assert request(params)[1]['error']=='invalid_grant','authorization code replay accepted' claims=json.loads(base64.urlsafe_b64decode(tokens['access_token'].split('.')[1]+'==')) assert claims['sub']=='1' and claims['scopes']==[] verified=php('$p=explode(".",$argv[1]);echo openssl_verify($p[0].".".$p[1],base64_decode(strtr($p[2],"-_","+/")),file_get_contents($argv[2]),OPENSSL_ALGO_SHA256);',tokens['access_token'],str(work/'oauth-public.key')) assert verified=='1','PHP rejected Rust JWT signature' decoded=json.loads(php('echo Defuse\\Crypto\\Crypto::decryptWithPassword($argv[1],$argv[2]);',tokens['refresh_token'],app_key)) assert decoded['user_id']=='1' and decoded['access_token_id']==claims['jti'] # PHP decrypts and re-encrypts the Rust refresh payload before redemption. php_refresh=php('echo Defuse\\Crypto\\Crypto::encryptWithPassword($argv[1],$argv[2]);',json.dumps(decoded),app_key) refresh={'grant_type':'refresh_token','client_id':'1','refresh_token':php_refresh} status,rotated=request(refresh);assert status==200,(status,rotated) assert request(refresh)[1]['error']=='invalid_grant','refresh replay accepted' assert con.execute('SELECT revoked FROM oauth_access_tokens WHERE id=?',(claims['jti'],)).fetchone()[0]==1 status,machine=request({'grant_type':'client_credentials','client_id':'2','client_secret':'secret','scope':'*'}) assert status==200,(status,machine) assert 'refresh_token' not in machine machine_claims=json.loads(base64.urlsafe_b64decode(machine['access_token'].split('.')[1]+'==')) assert machine_claims['sub']=='2' and machine_claims['scopes']==['*'] assert con.execute('SELECT user_id FROM oauth_access_tokens WHERE id=?',(machine_claims['jti'],)).fetchone()[0] is None print(json.dumps({'php_authorization_code_pkce':'passed','rust_jwt_php_signature':'passed','rust_refresh_php_decryption':'passed','php_refresh_rotation_replay':'passed','client_credentials_wildcard':'passed'})) finally: proc.terminate() try:proc.wait(timeout=5) except subprocess.TimeoutExpired:proc.kill();proc.wait() if __name__=='__main__':main()