Haketilo

 * Copyright (C) 2021, 2022 Wojtek Kosior <koszko@koszko.org>

#IMPORT common/indexeddb.js AS haketilodb

#FROM common/misc.js    IMPORT is_privileged_url, csp_header_regex, \

                               sha256_async AS sha256

#IF MOZILLA

/*

 * Under Mozilla-based browsers, responses are cached together with headers as

 * they appear *after* modifications by Haketilo. This means Haketilo's CSP

 * script-blocking headers might be present in responses loaded from cache. In

 * the meantime the user might have changes Haketilo settings to instead allow

 * the scripts on the page in question. This causes a problem and creates the

 * need to somehow restore the response headers to the state in which they

 * arrived from the server.

 * To cope with this, Haketilo will inject some additional headers with private

 * data. Those will include a hard-to-guess value derived from extension's

 * internal ID. It is assumed the internal ID has a longer lifetime than cached

 * responses.

 */

const settings_page_url = browser.runtime.getURL("html/settings.html");

const header_prefix_prom = sha256(settings_page_url)

      .then(hash => `X-Haketilo-${hash}`);

/*

 * Mozilla, unlike Chrome, allows webRequest callbacks to return promises. Here

 * we leverage that to be able to use asynchronous sha256 computation.

 */

async function on_headers_received(details) {

#IF NEVER

} /* Help auto-indent in editors. */

#ENDIF

#ELSE

function on_headers_received(details) {

#ENDIF

#IF MOZILLA

    const prefix = await header_prefix_prom;

    /*

     * We assume that the original CSP headers of a response are always

     * preserved under names of the form:

     *     X-Haketilo-<some_secret>-<original_name>

     * In some cases the original response may contain no CSP headers. To still

     * be able to tell whether the headers we were provided were modified by

     * Haketilo in the past, all modifications are accompanied by addition of an

     * extra header with name:

     *     X-Haketilo-<some_secret>

     */

    const restore_old_headers = details.fromCache &&

	  !!headers.filter(h => h.name === prefix).length;

    if (restore_old_headers) {

	const restored_headers = [];

	for (const h of headers) {

	    if (csp_header_regex.test(h.name) || h.name === prefix)

		continue;

	    if (h.name.startsWith(prefix)) {

		restored_headers.push({

		    name:  h.name.substring(prefix.length + 1),

		    value: h.value

		});

	    } else {

		restored_headers.push(h);

	    }

	}

	headers = restored_headers;

    }

#ENDIF

    if (!policy.allow) {

#IF MOZILLA

	const to_append = [{name: prefix, value: ":)"}];

	for (const h of headers.filter(h => csp_header_regex.test(h.name))) {

	    if (!policy.payload)

		to_append.push(Object.assign({}, h));

	    h.name = `${prefix}-${h.name}`;

	}

	headers.push(...to_append);

#ELSE

#ENDIF

	headers.push({name: "Content-Security-Policy", value: policy.csp});

    }

    /*

     * When page is meant to be viewed in the browser, use streamFilter to

     * inject a dummy <script> at the very beginning of it. This <script>

     * will cause extension's content scripts to run before page's first <meta>

     * tag is rendered so that they can prevent CSP rules inside <meta> tags

     * from blocking the payload we want to inject.

     */

    let use_stream_filter = !!policy.payload;

    if (use_stream_filter) {

	for (const header of headers) {

	    if (header.name.toLowerCase().trim() !== "content-disposition")

		continue;

	    if (/^\s*attachment\s*(;.*)$/i.test(header.value)) {

		use_stream_filter = false;

	    } else {

		use_stream_filter = true;

		break;

	    }

    use_stream_filter = use_stream_filter &&

	(details.statusCode < 300 || details.statusCode >= 400);

    if (use_stream_filter)

allowed_url = 'https://site.with.scripts.allow.ed/'

blocked_url = 'https://site.with.scripts.block.ed/'

payload_url = 'https://site.with.paylo.ad/'

            pqt.register(tree, "%(blocked)s***",

            pqt.register(tree, "%(payload)s***", "~allow", 1);

            pqt.register(tree, "%(payload)s***", "somemapping",

                         {identifier: "someresource"});

            ''' % {'blocked': blocked_url, 'payload': payload_url})

def webrequest_js_start_called():

    return webrequest_js() + ';\nstart("somesecret");'

ext_url = 'moz-extension://49de6ce9-49fc-49e1-8102-7ef35286389c/html/settings.html'

prefix = 'X-Haketilo-' + sha256(ext_url.encode()).digest().hex()

# Prepare a list of headers as could be sent by a website.

sample_csp_header = {

    'name': 'Content-Security-Policy',

    'value': "script-src 'self';"

}

sample_csp_header_idx = 7

sample_headers = [

    {'name': 'Content-Type',     'value': 'text/html;charset=utf-8'},

    {'name': 'Content-Length',   'value': '61954'},

    {'name': 'Content-Language', 'value': 'en'},

    {'name': 'Expires',          'value': 'Mon, 12 Mar 2012 11:04...'},

    {'name': 'Last-Modified',    'value': 'Fri, 26 Jul 2013 22:50...'},

    {'name': 'Cache-Control',    'value': 'max-age=0, s-maxage=86...'},

    {'name': 'Age',              'value': '224'},

    {'name': 'Server',           'value': 'nginx/1.1.19'},

    {'name': 'Date',             'value': 'Thu, 10 Mar 2022 12:09...'}

]

sample_headers.insert(sample_csp_header_idx, sample_csp_header)

# Prepare a list of headers as would be crafted by Haketilo when there is a

# payload to inject.

nonce_source = f'somemapping:someresource:{payload_url}:somesecret'.encode()

nonce = f'nonce-{sha256(nonce_source).digest().hex()}'

payload_csp_header = {

    'name': f'Content-Security-Policy',

    'value': ("prefetch-src 'none'; script-src-attr 'none'; "

              f"script-src '{nonce}'; script-src-elem '{nonce}';")

}

sample_payload_headers = [

    *sample_headers,

    {'name': prefix, 'value': ':)'},

    payload_csp_header

]

sample_payload_headers[sample_csp_header_idx] = {

    **sample_csp_header,

    'name': f'{prefix}-{sample_csp_header["name"]}',

}

# Prepare a list of headers as would be crafted by Haketilo when scripts are

# blocked.

sample_blocked_headers = [*sample_payload_headers]

sample_blocked_headers.pop()

sample_blocked_headers.append(sample_csp_header)

sample_blocked_headers.append({

    'name': f'Content-Security-Policy',

    'value': ("prefetch-src 'none'; script-src-attr 'none'; "

              f"script-src 'none'; script-src-elem 'none';")

})

@pytest.mark.get_page('https://gotmyowndoma.in')

@pytest.mark.parametrize('params', [

    (sample_headers,         allowed_url),

    (sample_blocked_headers, blocked_url),

    (sample_payload_headers, payload_url),

])

def test_webrequest_on_headers_received(driver, execute_in_page, params):

    """Unit-test the on_headers_received() function."""

    headers_out, url = params

    execute_in_page(

        '''{

        // Mock browser object.

        const url = arguments[0];

        this.browser = {runtime: {getURL: () => url}};

        }''',

        ext_url)

    execute_in_page(webrequest_js())

    execute_in_page('secret = "somesecret";')

    for headers_in in [

            sample_headers,

            sample_blocked_headers,

            sample_payload_headers

    ]:

        details = {'url': url, 'responseHeaders': headers_in, 'fromCache': True}

        res = execute_in_page('returnval(on_headers_received(arguments[0]));',

                              details)

        assert res == {'responseHeaders': headers_out}

@pytest.mark.ext_data({'background_script': webrequest_js_start_called})

def test_webrequest_real_pages(driver, execute_in_page):

    """

    Test webRequest-based header modifications by loading actual pages and

    attempting to run scripts within them.

    """

    driver.get(allowed_url)

    driver.get(payload_url)