| #!/usr/bin/env python |
| |
| # This Source Code Form is subject to the terms of the Mozilla Public |
| # License, v. 2.0. If a copy of the MPL was not distributed with this file, |
| # You can obtain one at http://mozilla.org/MPL/2.0/. |
| |
| import mozfile |
| import os |
| import shutil |
| import tempfile |
| import unittest |
| from mozprofile.permissions import Permissions |
| try: |
| import sqlite3 |
| except ImportError: |
| from pysqlite2 import dbapi2 as sqlite3 |
| |
| class PermissionsTest(unittest.TestCase): |
| |
| locations = """http://mochi.test:8888 primary,privileged |
| http://127.0.0.1:80 noxul |
| http://127.0.0.1:8888 privileged |
| """ |
| |
| profile_dir = None |
| locations_file = None |
| |
| def setUp(self): |
| self.profile_dir = tempfile.mkdtemp() |
| self.locations_file = mozfile.NamedTemporaryFile() |
| self.locations_file.write(self.locations) |
| self.locations_file.flush() |
| |
| def tearDown(self): |
| if self.profile_dir: |
| shutil.rmtree(self.profile_dir) |
| if self.locations_file: |
| self.locations_file.close() |
| |
| def write_perm_db(self, version=3): |
| permDB = sqlite3.connect(os.path.join(self.profile_dir, "permissions.sqlite")) |
| cursor = permDB.cursor() |
| |
| cursor.execute("PRAGMA user_version=%d;" % version) |
| |
| if version == 3: |
| cursor.execute("""CREATE TABLE IF NOT EXISTS moz_hosts ( |
| id INTEGER PRIMARY KEY, |
| host TEXT, |
| type TEXT, |
| permission INTEGER, |
| expireType INTEGER, |
| expireTime INTEGER, |
| appId INTEGER, |
| isInBrowserElement INTEGER)""") |
| elif version == 2: |
| cursor.execute("""CREATE TABLE IF NOT EXISTS moz_hosts ( |
| id INTEGER PRIMARY KEY, |
| host TEXT, |
| type TEXT, |
| permission INTEGER, |
| expireType INTEGER, |
| expireTime INTEGER)""") |
| else: |
| raise Exception("version must be 2 or 3") |
| |
| permDB.commit() |
| cursor.close() |
| |
| def test_create_permissions_db(self): |
| perms = Permissions(self.profile_dir, self.locations_file.name) |
| perms_db_filename = os.path.join(self.profile_dir, 'permissions.sqlite') |
| |
| select_stmt = 'select host, type, permission from moz_hosts' |
| |
| con = sqlite3.connect(perms_db_filename) |
| cur = con.cursor() |
| cur.execute(select_stmt) |
| entries = cur.fetchall() |
| |
| self.assertEqual(len(entries), 3) |
| |
| self.assertEqual(entries[0][0], 'mochi.test') |
| self.assertEqual(entries[0][1], 'allowXULXBL') |
| self.assertEqual(entries[0][2], 1) |
| |
| self.assertEqual(entries[1][0], '127.0.0.1') |
| self.assertEqual(entries[1][1], 'allowXULXBL') |
| self.assertEqual(entries[1][2], 2) |
| |
| self.assertEqual(entries[2][0], '127.0.0.1') |
| self.assertEqual(entries[2][1], 'allowXULXBL') |
| self.assertEqual(entries[2][2], 1) |
| |
| perms._locations.add_host('a.b.c', options='noxul') |
| |
| cur.execute(select_stmt) |
| entries = cur.fetchall() |
| |
| self.assertEqual(len(entries), 4) |
| self.assertEqual(entries[3][0], 'a.b.c') |
| self.assertEqual(entries[3][1], 'allowXULXBL') |
| self.assertEqual(entries[3][2], 2) |
| |
| # when creating a DB we should default to user_version==2 |
| cur.execute('PRAGMA user_version') |
| entries = cur.fetchall() |
| self.assertEqual(entries[0][0], 2) |
| |
| perms.clean_db() |
| # table should be removed |
| cur.execute("select * from sqlite_master where type='table'") |
| entries = cur.fetchall() |
| self.assertEqual(len(entries), 0) |
| |
| def test_nw_prefs(self): |
| perms = Permissions(self.profile_dir, self.locations_file.name) |
| |
| prefs, user_prefs = perms.network_prefs(False) |
| |
| self.assertEqual(len(user_prefs), 0) |
| self.assertEqual(len(prefs), 6) |
| |
| self.assertEqual(prefs[0], ('capability.principal.codebase.p1.granted', |
| 'UniversalXPConnect')) |
| self.assertEqual(prefs[1], ('capability.principal.codebase.p1.id', |
| 'http://mochi.test:8888')) |
| self.assertEqual(prefs[2], ('capability.principal.codebase.p1.subjectName', '')) |
| |
| self.assertEqual(prefs[3], ('capability.principal.codebase.p2.granted', |
| 'UniversalXPConnect')) |
| self.assertEqual(prefs[4], ('capability.principal.codebase.p2.id', |
| 'http://127.0.0.1:8888')) |
| self.assertEqual(prefs[5], ('capability.principal.codebase.p2.subjectName', '')) |
| |
| prefs, user_prefs = perms.network_prefs(True) |
| self.assertEqual(len(user_prefs), 2) |
| self.assertEqual(user_prefs[0], ('network.proxy.type', 2)) |
| self.assertEqual(user_prefs[1][0], 'network.proxy.autoconfig_url') |
| |
| origins_decl = "var origins = ['http://mochi.test:8888', 'http://127.0.0.1:80', 'http://127.0.0.1:8888'];" |
| self.assertTrue(origins_decl in user_prefs[1][1]) |
| |
| proxy_check = ("if (isHttp) return 'PROXY mochi.test:8888';", |
| "if (isHttps) return 'PROXY mochi.test:4443';", |
| "if (isWebSocket) return 'PROXY mochi.test:4443';", |
| "if (isWebSocketSSL) return 'PROXY mochi.test:4443';") |
| self.assertTrue(all(c in user_prefs[1][1] for c in proxy_check)) |
| |
| def verify_user_version(self, version): |
| """Verifies that we call INSERT statements using the correct number |
| of columns for existing databases. |
| """ |
| self.write_perm_db(version=version) |
| Permissions(self.profile_dir, self.locations_file.name) |
| perms_db_filename = os.path.join(self.profile_dir, 'permissions.sqlite') |
| |
| select_stmt = 'select * from moz_hosts' |
| |
| con = sqlite3.connect(perms_db_filename) |
| cur = con.cursor() |
| cur.execute(select_stmt) |
| entries = cur.fetchall() |
| |
| self.assertEqual(len(entries), 3) |
| |
| columns = 8 if version == 3 else 6 |
| self.assertEqual(len(entries[0]), columns) |
| for x in range(4, columns): |
| self.assertEqual(entries[0][x], 0) |
| |
| def test_existing_permissions_db_v2(self): |
| self.verify_user_version(2) |
| |
| def test_existing_permissions_db_v3(self): |
| self.verify_user_version(3) |
| |
| |
| if __name__ == '__main__': |
| unittest.main() |
