Haketilo

# SPDX-License-Identifier: CC0-1.0

"""

Haketilo unit tests - routing HTTP requests through background script

"""

# This file is part of Haketilo

#

# Copyright (C) 2022 Wojtek Kosior <koszko@koszko.org>

#

# This program is free software: you can redistribute it and/or modify

# it under the terms of the CC0 1.0 Universal License as published by

# the Creative Commons Corporation.

#

# This program is distributed in the hope that it will be useful,

# but WITHOUT ANY WARRANTY; without even the implied warranty of

# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the

# CC0 1.0 Universal License for more details.

import pytest

import json

from selenium.webdriver.support.ui import WebDriverWait

from ..script_loader import load_script

from ..world_wide_library import some_data

datas = {

    'resource':       'https://anotherdoma.in/resource/blocked/by/CORS.json',

    'nonexistent':    'https://nxdoma.in/resource.json',

    'invalid':        'w3csucks://invalid.url/',

    'redirected_ok':  'https://site.with.scripts.block.ed',

    'redirected_err': 'https://site.with.scripts.block.ed'

}

for name, url in [*datas.items()]:

    datas[name] = {'url': url}

datas['redirected_ok']['init']  = {'redirect': 'follow'}

datas['redirected_err']['init'] = {'redirect': 'error'}

content_script = '''\

const datas = %s;

async function fetch_resources() {

    const results = {};

    const promises = [];

    for (const [name, data] of Object.entries(datas)) {

        const sending = browser.runtime.sendMessage(["CORS_bypass", data]);

        promises.push(sending.then(response => results[name] = response));

    }

    await Promise.all(promises);

    window.wrappedJSObject.haketilo_fetch_results = results;

}

fetch_resources();

'''

content_script = content_script % json.dumps(datas);

@pytest.mark.ext_data({

    'content_script': content_script,

    'background_script':

    lambda: load_script('background/CORS_bypass_server.js') + '; start();'

})

@pytest.mark.usefixtures('webextension')

def test_CORS_bypass_server(driver, execute_in_page):

    """

    Test if CORS bypassing works and if errors get properly forwarded.

    """

    driver.get('https://gotmyowndoma.in/')

    # First, verify that requests without CORS bypass measures fail.

    results = execute_in_page(

        '''

        const result = {};

        let promises = [];

        for (const [name, data] of Object.entries(arguments[0])) {

            const [ok_cb, err_cb] =

                ["ok", "err"].map(status => () => result[name] = status);

            promises.push(fetch(data.url).then(ok_cb, err_cb));

        }

        // Make the promises non-failing.

        promises = promises.map(p => new Promise(cb => p.then(cb, cb)));

        returnval(Promise.all(promises).then(() => result));

        ''',

        {**datas, 'sameorigin': './nonexistent_resource'})

    assert results == dict([*[(k, 'err') for k in datas.keys()],

                            ('sameorigin', 'ok')])

    done = lambda d: d.execute_script('return window.haketilo_fetch_results;')

    results = WebDriverWait(driver, 10).until(done)

    assert set(results['invalid'].keys()) == {'error'}

    assert results['invalid']['error']['fileName'].endswith('background.js')

    assert type(results['invalid']['error']['lineNumber']) is int

    assert type(results['invalid']['error']['message']) is str

    assert results['invalid']['error']['name'] == 'TypeError'

    assert results['nonexistent']['status'] == 404

    assert results['nonexistent']['statusText'] == 'Not Found'

    assert any([name.lower() == 'content-length'

                for name, value in results['nonexistent']['headers']])

    assert bytes.fromhex(results['nonexistent']['body']) == \

        b'Handler for this URL not found.'

    assert results['resource']['status'] == 200

    assert results['resource']['statusText'] == 'OK'

    assert any([name.lower() == 'content-length'

                for name, value in results['resource']['headers']])

    assert bytes.fromhex(results['resource']['body']) == b'{"some": "data"}'

    assert results['redirected_ok']['status'] == 200

    assert results['redirected_err']['error']['name'] == 'TypeError'