OpenWPM/test/test_extension.py

import pytest # NOQA
import os
import utilities
import expected
from openwpmtest import OpenWPMTest
from ..automation import TaskManager
from datetime import datetime
# TODO: add test for setter instrumentation


class TestExtension(OpenWPMTest):
    NUM_BROWSERS = 1

    def get_config(self, data_dir):
        manager_params, browser_params = TaskManager.load_default_params(self.NUM_BROWSERS)
        manager_params['data_directory'] = data_dir
        manager_params['log_directory'] = data_dir
        browser_params[0]['headless'] = True
        browser_params[0]['js_instrument'] = True
        manager_params['db'] = os.path.join(manager_params['data_directory'],
                                            manager_params['database_name'])
        return manager_params, browser_params

    def test_property_enumeration(self, tmpdir):
        test_url = utilities.BASE_TEST_URL + '/property_enumeration.html'
        db = self.visit(test_url, str(tmpdir))
        rows = utilities.query_db(db,
                                  "SELECT script_url, symbol FROM javascript")
        observed_symbols = set()
        for script_url, symbol in rows:
            assert script_url == test_url
            observed_symbols.add(symbol)
        assert expected.properties == observed_symbols

    def test_canvas_fingerprinting(self, tmpdir):
        db = self.visit('/canvas_fingerprinting.html', str(tmpdir))
        # Check that all calls and methods are recorded
        rows = utilities.get_javascript_entries(db)
        observed_rows = set()
        for item in rows:
            observed_rows.add(item)
        assert expected.canvas == observed_rows

    def test_extension_gets_correct_visit_id(self, tmpdir):
        manager_params, browser_params = self.get_config(str(tmpdir))
        manager = TaskManager.TaskManager(manager_params, browser_params)

        url_a = utilities.BASE_TEST_URL + '/simple_a.html'
        url_b = utilities.BASE_TEST_URL + '/simple_b.html'

        manager.get(url_a)
        manager.get(url_b)
        manager.close()
        qry_res = utilities.query_db(manager_params['db'],
                                     "SELECT visit_id, site_url FROM site_visits")

        # Construct dict mapping site_url to visit_id
        visit_ids = dict()
        for row in qry_res:
            visit_ids[row[1]] = row[0]

        simple_a_visit_id = utilities.query_db(
                                    manager_params['db'],
                                    "SELECT visit_id FROM javascript WHERE "
                                    "symbol=?", ("window.navigator.userAgent",))

        simple_b_visit_id = utilities.query_db(
                                    manager_params['db'],
                                    "SELECT visit_id FROM javascript WHERE "
                                    "symbol=?", ("window.navigator.platform",))

        assert visit_ids[url_a] == simple_a_visit_id[0][0]
        assert visit_ids[url_b] == simple_b_visit_id[0][0]

    def check_webrtc_sdp_offer(self, sdp_str):
        """Make sure the SDP offer includes expected fields/strings.

        SDP offer contains randomly generated strings (e.g. GUID). That's why
        we don't expect a fixed string but only check the presence of certain
        protocol fields.
        """
        for expected_str in expected.webrtc_sdp_offer_strings:
            assert expected_str in sdp_str

    def test_webrtc_localip(self, tmpdir):
        db = self.visit('/webrtc_localip.html', str(tmpdir))
        # Check that all calls and methods are recorded
        rows = utilities.get_javascript_entries(db)
        observed_rows = set()
        for item in rows:
            if item[1] == "RTCPeerConnection.setLocalDescription":
                assert item[2:5] == (u'call', u'', 0)
                sdp_offer = item[5]
                self.check_webrtc_sdp_offer(sdp_offer)
            else:
                observed_rows.add(item)
        assert set(expected.webrtc_calls) == observed_rows

    @pytest.mark.skipif("TRAVIS" in os.environ and os.environ["TRAVIS"] == "true", reason='Flaky on Travis CI')
    def test_audio_fingerprinting(self, tmpdir):
        db = self.visit('/audio_fingerprinting.html', str(tmpdir))
        # Check that all calls and methods are recorded
        rows = utilities.get_javascript_entries(db)
        observed_symbols = set()
        for item in rows:
            observed_symbols.add(item[1])
        assert expected.audio == observed_symbols

    def test_js_call_stack(self, tmpdir):
        db = self.visit('/js_call_stack.html', str(tmpdir))
        # Check that all stack info are recorded
        rows = utilities.get_javascript_entries(db, all_columns=True)
        observed_rows = set()
        for item in rows:
            observed_rows.add(item[3:11])
        assert set(expected.js_stack_calls) == observed_rows

    def test_js_time_stamp(self, tmpdir):
        # Check that timestamp is recorded correctly for the javascript table
        MAX_TIMEDELTA = 30  # max time diff in seconds
        db = self.visit('/canvas_fingerprinting.html', str(tmpdir))
        utc_now = datetime.utcnow()  # OpenWPM stores timestamp in UTC time
        rows = utilities.get_javascript_entries(db, all_columns=True)
        assert len(rows)  # make sure we have some JS events captured
        for row in rows:
            js_time = datetime.strptime(row[14], "%Y-%m-%dT%H:%M:%S.%fZ")
            # compare UTC now and the timestamp recorded at the visit
            assert (utc_now - js_time).seconds < MAX_TIMEDELTA
        assert not utilities.any_command_failed(db)