69 lines
5.3 KiB
Python
69 lines
5.3 KiB
Python
"""Supplemental semantic checks for disposable fixture servers; never print bodies."""
|
|
import base64
|
|
import http.cookiejar
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
from pathlib import Path
|
|
import re
|
|
import shutil
|
|
import subprocess
|
|
import urllib.parse
|
|
import urllib.request
|
|
|
|
|
|
def check(reference, candidate, php, source, work, php_env, timeout):
|
|
spec = importlib.util.spec_from_file_location('compat_http', Path(__file__).with_name('differential_http.py'))
|
|
harness = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(harness)
|
|
sides = {}
|
|
for side, base in (('reference', reference), ('candidate', candidate)):
|
|
jar = http.cookiejar.CookieJar()
|
|
opener = urllib.request.build_opener(harness.NoRedirect(), urllib.request.HTTPCookieProcessor(jar))
|
|
def fetch(case): return harness.fetch(base, case, side, timeout, opener, jar)
|
|
health = fetch({'path': '/up'})
|
|
csrf = fetch({'path': '/refresh-csrf'})
|
|
headers = {'X-XSRF-TOKEN': 'XSRF-TOKEN'}
|
|
login = fetch({'path': '/user/login', 'method': 'POST', 'headers_from_cookies': headers, 'json': {'email': 'compat@example.invalid', 'password': 'Synthetic-compat-password-4620'}})
|
|
created = fetch({'path': '/api/v1/twofaccounts', 'method': 'POST', 'headers_from_cookies': headers, 'json': {'service': 'compat-semantic', 'account': 'fixture@example.invalid', 'otp_type': 'hotp', 'secret': 'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ', 'digits': 6, 'algorithm': 'sha1', 'counter': 0}})
|
|
qr = {'error': 'account_not_created'}
|
|
if created.get('status') == 201:
|
|
account_id = json.loads(created['body'])['id']
|
|
qr = fetch({'path': f'/api/v1/twofaccounts/{account_id}/qrcode', 'headers_from_cookies': headers})
|
|
sides[side] = {'base': base, 'health': health, 'csrf': csrf, 'login': login, 'qr': qr}
|
|
pattern = r'Response successfully rendered in [0-9]+ms\.'
|
|
health_bodies = [sides[side]['health'].get('body', b'').decode(errors='replace').replace(sides[side]['base'], '<BASE>') for side in sides]
|
|
health = {'content_types_match': all(sides[side]['health'].get('content_type') == 'text/html' for side in sides), 'timing_field_present_once': all(len(re.findall(pattern, body)) == 1 for body in health_bodies), 'equal_except_render_ms': re.sub(pattern, 'Response successfully rendered in <MS>ms.', health_bodies[0]) == re.sub(pattern, 'Response successfully rendered in <MS>ms.', health_bodies[1]), 'statuses_ok': all(sides[side]['health'].get('status') == 200 for side in sides)}
|
|
csrf = {'text_token_shapes_valid': all(harness.is_text_token(sides[side]['csrf']) for side in sides), 'subsequent_login_succeeded': all(sides[side]['login'].get('status') == 200 for side in sides)}
|
|
qr = {'result': 'error', 'decoded_uri_equal': False, 'matches_fixture': False}
|
|
try:
|
|
node = os.environ.get('CODEX_PRIMARY_RUNTIME_NODE') or shutil.which('node')
|
|
if not node: raise RuntimeError('missing_svg_renderer')
|
|
decoded = []
|
|
for side in sides:
|
|
response = sides[side]['qr']
|
|
if response.get('status') != 200 or response.get('content_type') != 'application/json': raise RuntimeError('qr_request_failed')
|
|
image = json.loads(response['body'])['qrcode']
|
|
prefix = 'data:image/svg+xml;base64,'
|
|
if not image.startswith(prefix): raise RuntimeError('unexpected_image_format')
|
|
svg = work / (side + '-qr.svg')
|
|
png = work / (side + '-qr.png')
|
|
svg.write_bytes(base64.b64decode(image[len(prefix):], validate=True))
|
|
render = subprocess.run([node, '-e', "require('sharp')(process.argv[1],{density:300}).resize(1024,1024).flatten({background:'#fff'}).png().toFile(process.argv[2]).catch(()=>process.exit(1))", str(svg), str(png)], capture_output=True, timeout=30)
|
|
if render.returncode: raise RuntimeError('svg_render_failed')
|
|
code = "require $argv[1]; $r=new Zxing\\QrReader($argv[2]); $t=$r->text(); if (!is_string($t)||$t==='') exit(1); echo $t;"
|
|
result = subprocess.run([str(php), '-r', code, str(source / 'vendor/autoload.php'), str(png)], env=php_env, capture_output=True, timeout=30)
|
|
if result.returncode: raise RuntimeError('qr_decode_failed')
|
|
uri = result.stdout.decode()
|
|
parsed = urllib.parse.urlsplit(uri)
|
|
values = dict(urllib.parse.parse_qsl(parsed.query))
|
|
normalized = {'type': parsed.netloc.lower(), 'label': urllib.parse.unquote(parsed.path.lstrip('/')), 'secret': values.get('secret'), 'issuer': values.get('issuer'), 'algorithm': values.get('algorithm', 'SHA1').lower(), 'digits': int(values.get('digits', 6)), 'counter': int(values.get('counter', 0))}
|
|
if parsed.scheme.lower() != 'otpauth': raise RuntimeError('unexpected_qr_payload')
|
|
decoded.append(normalized)
|
|
expected = {'type': 'hotp', 'label': 'compat-semantic:fixture@example.invalid', 'secret': 'GEZDGNBVGY3TQOJQGEZDGNBVGY3TQOJQ', 'issuer': 'compat-semantic', 'algorithm': 'sha1', 'digits': 6, 'counter': 0}
|
|
qr = {'result': 'decoded', 'decoded_uri_equal': decoded[0] == decoded[1], 'matches_fixture': all(value == expected for value in decoded)}
|
|
except Exception as error:
|
|
qr['error'] = type(error).__name__
|
|
if isinstance(error, RuntimeError): qr['reason'] = str(error)
|
|
return {'health_render_timing': health, 'csrf_token_flow': csrf, 'qr_payload': qr}
|