rapp-playwright/tools/verify_consumer.py

#!/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)