#!/usr/bin/env python3 """Verify the browser API through an already established consumer connection. This command does not establish identity, create access grants or restart services. """ import argparse from datetime import datetime, timezone import json from pathlib import Path import struct import sys import time sys.path.insert(0, str(Path(__file__).resolve().parents[1] / 'client')) from rapp_playwright import Client, ClientError, collect, save TERMINAL = {'passed', 'failed', 'cancelled', 'interrupted'} def require(condition, message): if not condition: raise ClientError(message) def denied(operation, expected): try: operation() except ClientError as exc: require(str(exc) == expected, 'unexpected denial: ' + str(exc)) else: raise ClientError('request unexpectedly permitted') def wait_for(client, run, *, running=False): deadline = time.monotonic() + 90 while time.monotonic() < deadline: result = client.request('/v1/runs/' + run['id'], token=run['token']) if running and result['status'] == 'running': return result if result['status'] in TERMINAL: require(not running, 'cancellation case terminated before running') return result time.sleep(0.1) raise ClientError('acceptance_wait_timeout') def verify(args): args.out.mkdir(parents=True, mode=0o700, exist_ok=False) report = {'started_at': datetime.now(timezone.utc).isoformat(), 'service_url': args.url, 'argv': sys.argv, 'expected_release': args.expected_release, 'expected_location': args.expected_location, 'checks': [], 'status': 'failed', 'identity_authentication_verified': False, 'scope': 'browser API only; pair with native consumer identity and permission receipts'} client = Client(args.url) runs = [] def passed(name): report['checks'].append(name) def submit(spec, name): out = args.out / name out.mkdir(mode=0o700) run = client.request('/v1/runs', 'POST', spec) runs.append(run) save(out / 'run.json', run) return run, out try: pf = client.request('/v1/preflight') save(args.out / 'preflight.json', pf) require(pf.get('available') is True and pf.get('api_version') == '1', 'provider_not_ready') require(pf.get('release') == args.expected_release, 'release_mismatch') require(pf.get('execution_location') == args.expected_location, 'execution_location_mismatch') passed('preflight_identity') spec = client.request('/v1/trial') spec['viewports'] = [{'name': 'desktop', 'width': 1280, 'height': 800}, {'name': 'narrow', 'width': 390, 'height': 844}] save(args.out / 'trial-spec.json', spec) run, out = submit(spec, 'trial') wait_for(client, run) receipt = collect(client, run, out) require(receipt['status'] == 'passed' and receipt['error'] is None, 'trial_failed') require(receipt['completed_steps'] == len(spec['steps']) * len(spec['viewports']), 'incomplete_trial') require(receipt['release'] == args.expected_release and receipt['execution_location'] == args.expected_location, 'trial_identity_mismatch') require(len(receipt['downloads']) == 4, 'expected_four_trial_screenshots') sizes = {v['name']: (v['width'], v['height']) for v in spec['viewports']} for item in receipt['downloads']: data = (out / item['name']).read_bytes() require(data[:8] == b'\x89PNG\r\n\x1a\n' and data[12:16] == b'IHDR', 'invalid_png') viewport = next((v for v in sizes if item['name'].startswith(v + '-')), None) require(viewport is not None and struct.unpack('>II', data[16:24]) == sizes[viewport], 'wrong_png_dimensions') passed('responsive_keyboard_trial_and_downloads') path = '/v1/runs/' + run['id'] denied(lambda: client.request(path, token='0' * 64), 'HTTP 404: not_found') denied(lambda: client.request(path + '/artifacts/' + receipt['downloads'][0]['name'], token='0' * 64), 'HTTP 404: not_found') passed('foreign_run_and_artifact_denied') bad = {**spec, 'viewports': spec['viewports'][:1], 'steps': [ {'action': 'screenshot', 'name': 'before-failure'}, {'action': 'text', 'selector': '#count', 'value': 'acceptance-deliberately-wrong'}]} failed, failure_out = submit(bad, 'ui-failure') wait_for(client, failed) failure = collect(client, failed, failure_out) require(failure['status'] == 'failed' and failure['error'] == 'ui_assertion_failed', 'failure_not_classified') require(len(failure['downloads']) >= 1, 'failure_artifact_missing') passed('ui_failure_preserves_evidence') # An undeclared private origin must fail before browser execution. denied(lambda: client.request('/v1/runs', 'POST', {**spec, 'target': 'http://127.0.0.1:1/'}), 'HTTP 403: target_not_allowed') passed('private_target_denied') slow = {**spec, 'viewports': spec['viewports'][:1], 'timeout_ms': 60000, 'steps': [{'action': 'visible', 'selector': '#acceptance-never-present'}]} cancelled, cancel_out = submit(slow, 'cancel') wait_for(client, cancelled, running=True) start = time.monotonic() client.request('/v1/runs/' + cancelled['id'], 'DELETE', token=cancelled['token']) result = wait_for(client, cancelled) require(result['status'] == 'cancelled', 'cancellation_failed') require(time.monotonic() - start < 10, 'cancellation_exceeded_10s') collect(client, cancelled, cancel_out) passed('running_job_cancellation') report['status'] = 'passed' except KeyboardInterrupt: report['error'] = 'cancelled_by_client' except (ClientError, OSError, ValueError, KeyError, struct.error) as exc: report['error'] = str(exc) finally: # Cancel only runs created here. Completed runs remain downloadable until expiry. for run in runs: try: client.request('/v1/runs/' + run['id'], 'DELETE', token=run['token']) except ClientError: pass report['finished_at'] = datetime.now(timezone.utc).isoformat() save(args.out / 'verification.json', report) print(json.dumps(report, indent=2)) return 0 if report['status'] == 'passed' else 3 def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('--url', required=True) parser.add_argument('--expected-release', required=True) parser.add_argument('--expected-location', required=True) parser.add_argument('--out', required=True, type=Path) return verify(parser.parse_args()) if __name__ == '__main__': try: sys.exit(main()) except (ClientError, OSError, ValueError) as exc: print(json.dumps({'status': 'failed', 'error': str(exc)}), file=sys.stderr) sys.exit(3)