Skip to content


Subversion checkout URL

You can clone with
Download ZIP
Tree: 0d240e02eb
Fetching contributors…

Cannot retrieve contributors at this time

373 lines (293 sloc) 12.465 kB
# 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
add permissions to the profile
__all__ = ['MissingPrimaryLocationError', 'MultiplePrimaryLocationsError',
'DuplicateLocationError', 'BadPortLocationError',
'LocationsSyntaxError', 'Location', 'ServerLocations',
import codecs
import itertools
import os
import sqlite3
except ImportError:
from pysqlite2 import dbapi2 as sqlite3
import urlparse
_DEFAULT_PORTS = { 'http': '8888',
'https': '4443',
'ws': '9988',
'wss': '4443' }
class LocationError(Exception):
"""Signifies an improperly formed location."""
def __str__(self):
s = "Bad location"
if self.message:
s += ": %s" % self.message
return s
class MissingPrimaryLocationError(LocationError):
"""No primary location defined in locations file."""
def __init__(self):
LocationError.__init__(self, "missing primary location")
class MultiplePrimaryLocationsError(LocationError):
"""More than one primary location defined."""
def __init__(self):
LocationError.__init__(self, "multiple primary locations")
class DuplicateLocationError(LocationError):
"""Same location defined twice."""
def __init__(self, url):
LocationError.__init__(self, "duplicate location: %s" % url)
class BadPortLocationError(LocationError):
"""Location has invalid port value."""
def __init__(self, given_port):
LocationError.__init__(self, "bad value for port: %s" % given_port)
class LocationsSyntaxError(Exception):
"""Signifies a syntax error on a particular line in server-locations.txt."""
def __init__(self, lineno, err=None):
self.err = err
self.lineno = lineno
def __str__(self):
s = "Syntax error on line %s" % self.lineno
if self.err:
s += ": %s." % self.err
s += "."
return s
class Location(object):
"""Represents a location line in server-locations.txt."""
attrs = ('scheme', 'host', 'port')
def __init__(self, scheme, host, port, options):
for attr in self.attrs:
setattr(self, attr, locals()[attr])
self.options = options
except ValueError:
raise BadPortLocationError(self.port)
def isEqual(self, location):
"""compare scheme://host:port, but ignore options"""
return len([i for i in self.attrs if getattr(self, i) == getattr(location, i)]) == len(self.attrs)
__eq__ = isEqual
def url(self):
return '%s://%s:%s' % (self.scheme,, self.port)
def __str__(self):
return '%s %s' % (self.url(), ','.join(self.options))
class ServerLocations(object):
"""Iterable collection of locations.
Use provided functions to add new locations, rather that manipulating
_locations directly, in order to check for errors and to ensure the
callback is called, if given.
def __init__(self, filename=None, add_callback=None):
self.add_callback = add_callback
self._locations = []
self.hasPrimary = False
if filename:
def __iter__(self):
return self._locations.__iter__()
def __len__(self):
return len(self._locations)
def add(self, location, suppress_callback=False):
if "primary" in location.options:
if self.hasPrimary:
raise MultiplePrimaryLocationsError()
self.hasPrimary = True
if self.add_callback and not suppress_callback:
def add_host(self, host, port='80', scheme='http', options='privileged'):
if isinstance(options, basestring):
options = options.split(',')
self.add(Location(scheme, host, port, options))
def read(self, filename, check_for_primary=True):
Reads the file and adds all valid locations to the ``self._locations`` array.
:param filename: in the format of server-locations.txt_
:param check_for_primary: if True, a ``MissingPrimaryLocationError`` exception is raised if no primary is found
.. _server-locations.txt:
The only exception is that the port, if not defined, defaults to 80 or 443.
FIXME: Shouldn't this default to the protocol-appropriate port? Is
there any reason to have defaults at all?
locationFile =, "r", "UTF-8")
lineno = 0
new_locations = []
for line in locationFile:
line = line.strip()
lineno += 1
# check for comments and blank lines
if line.startswith("#") or not line:
# split the server from the options
server, options = line.rsplit(None, 1)
options = options.split(',')
except ValueError:
server = line
options = []
# parse the server url
if '://' not in server:
server = 'http://' + server
scheme, netloc, path, query, fragment = urlparse.urlsplit(server)
# get the host and port
host, port = netloc.rsplit(':', 1)
except ValueError:
host = netloc
port = _DEFAULT_PORTS.get(scheme, '80')
location = Location(scheme, host, port, options)
self.add(location, suppress_callback=True)
except LocationError, e:
raise LocationsSyntaxError(lineno, e)
# ensure that a primary is found
if check_for_primary and not self.hasPrimary:
raise LocationsSyntaxError(lineno + 1,
if self.add_callback:
class Permissions(object):
"""Allows handling of permissions for ``mozprofile``"""
_num_permissions = 0
def __init__(self, profileDir, locations=None):
self._profileDir = profileDir
self._locations = ServerLocations(add_callback=self.write_db)
if locations:
if isinstance(locations, ServerLocations):
self._locations = locations
self._locations.add_callback = self.write_db
elif isinstance(locations, list):
for l in locations:
elif isinstance(locations, dict):
elif os.path.exists(locations):
def write_db(self, locations):
"""write permissions to the sqlite database"""
# Open database and create table
permDB = sqlite3.connect(os.path.join(self._profileDir, "permissions.sqlite"))
cursor = permDB.cursor();
cursor.execute("PRAGMA schema_version = 3;")
# SQL copied from
cursor.execute("""CREATE TABLE IF NOT EXISTS moz_hosts (
host TEXT,
type TEXT,
permission INTEGER,
expireType INTEGER,
expireTime INTEGER)""")
for location in locations:
# set the permissions
permissions = { 'allowXULXBL': 'noxul' not in location.options }
for perm, allow in permissions.iteritems():
self._num_permissions += 1
if allow:
permission_type = 1
permission_type = 2
cursor.execute("INSERT INTO moz_hosts values(?, ?, ?, ?, 0, 0)",
(self._num_permissions,, perm,
# Commit and close
def network_prefs(self, proxy=None):
take known locations and generate preferences to handle permissions and proxy
returns a tuple of prefs, user_prefs
# Grant God-power to all the privileged servers on which tests run.
prefs = []
privileged = [i for i in self._locations if "privileged" in i.options]
for (i, l) in itertools.izip(itertools.count(1), privileged):
prefs.append(("capability.principal.codebase.p%s.granted" % i, "UniversalXPConnect"))
prefs.append(("" % i, "%s://%s:%s" %
(l.scheme,, l.port)))
prefs.append(("capability.principal.codebase.p%s.subjectName" % i, ""))
if proxy:
user_prefs = self.pac_prefs(proxy)
user_prefs = []
return prefs, user_prefs
def pac_prefs(self, user_proxy=None):
return preferences for Proxy Auto Config. originally taken from
# We need to proxy every server but the primary one.
origins = ["'%s'" % l.url()
for l in self._locations]
origins = ", ".join(origins)
proxy["origins"] = origins
for l in self._locations:
if "primary" in l.options:
proxy["remote"] =
proxy[l.scheme] = l.port
# overwrite defaults with user specified proxy
if isinstance(user_proxy, dict):
# TODO: this should live in a template!
# TODO: So changing the 5th line of the regex below from (\\\\\\\\d+)
# to (\\\\d+) makes this code work. Not sure why there would be this
# difference between and this file.
pacURL = """data:text/plain,
function FindProxyForURL(url, host)
var origins = [%(origins)s];
var regex = new RegExp('^([a-z][-a-z0-9+.]*)' +
'://' +
'(?:[^/@]*@)?' +
'(.*?)' +
var matches = regex.exec(url);
if (!matches)
return 'DIRECT';
var isHttp = matches[1] == 'http';
var isHttps = matches[1] == 'https';
var isWebSocket = matches[1] == 'ws';
var isWebSocketSSL = matches[1] == 'wss';
if (!matches[3])
if (isHttp | isWebSocket) matches[3] = '80';
if (isHttps | isWebSocketSSL) matches[3] = '443';
if (isWebSocket)
matches[1] = 'http';
if (isWebSocketSSL)
matches[1] = 'https';
var origin = matches[1] + '://' + matches[2] + ':' + matches[3];
if (origins.indexOf(origin) < 0)
return 'DIRECT';
if (isHttp) return 'PROXY %(remote)s:%(http)s';
if (isHttps) return 'PROXY %(remote)s:%(https)s';
if (isWebSocket) return 'PROXY %(remote)s:%(ws)s';
if (isWebSocketSSL) return 'PROXY %(remote)s:%(wss)s';
return 'DIRECT';
}""" % proxy
pacURL = "".join(pacURL.splitlines())
prefs = []
prefs.append(("network.proxy.type", 2))
prefs.append(("network.proxy.autoconfig_url", pacURL))
return prefs
def clean_db(self):
"""Removed permissions added by mozprofile."""
sqlite_file = os.path.join(self._profileDir, "permissions.sqlite")
if not os.path.exists(sqlite_file):
# Open database and create table
permDB = sqlite3.connect(sqlite_file)
cursor = permDB.cursor();
# TODO: only delete values that we add, this would require sending in the full permissions object
cursor.execute("DROP TABLE IF EXISTS moz_hosts");
# Commit and close
Jump to Line
Something went wrong with that request. Please try again.