summaryrefslogtreecommitdiff
path: root/src/leap/base
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/base')
-rw-r--r--src/leap/base/auth.py355
-rw-r--r--src/leap/base/checks.py198
-rw-r--r--src/leap/base/config.py139
-rw-r--r--src/leap/base/connection.py10
-rw-r--r--src/leap/base/constants.py36
-rw-r--r--src/leap/base/exceptions.py39
-rw-r--r--src/leap/base/network.py39
-rw-r--r--src/leap/base/pluggableconfig.py38
-rw-r--r--src/leap/base/providers.py14
-rw-r--r--src/leap/base/specs.py22
-rw-r--r--src/leap/base/tests/__init__.py0
-rw-r--r--src/leap/base/tests/test_auth.py58
-rw-r--r--src/leap/base/tests/test_checks.py116
-rw-r--r--src/leap/base/tests/test_providers.py33
14 files changed, 918 insertions, 179 deletions
diff --git a/src/leap/base/auth.py b/src/leap/base/auth.py
new file mode 100644
index 00000000..c2d3f424
--- /dev/null
+++ b/src/leap/base/auth.py
@@ -0,0 +1,355 @@
+import binascii
+import json
+import logging
+#import urlparse
+
+import requests
+import srp
+
+from PyQt4 import QtCore
+
+from leap.base import constants as baseconstants
+from leap.crypto import leapkeyring
+from leap.util.misc import null_check
+from leap.util.web import get_https_domain_and_port
+
+logger = logging.getLogger(__name__)
+
+SIGNUP_TIMEOUT = getattr(baseconstants, 'SIGNUP_TIMEOUT', 5)
+
+"""
+Registration and authentication classes for the
+SRP auth mechanism used in the leap platform.
+
+We're using the srp library which uses a c-based implementation
+of the protocol if the c extension is available, and a python-based
+one if not.
+"""
+
+
+class SRPAuthenticationError(Exception):
+ """
+ exception raised
+ for authentication errors
+ """
+
+
+safe_unhexlify = lambda x: binascii.unhexlify(x) \
+ if (len(x) % 2 == 0) else binascii.unhexlify('0' + x)
+
+
+class LeapSRPRegister(object):
+
+ def __init__(self,
+ schema="https",
+ provider=None,
+ verify=True,
+ register_path="1/users",
+ method="POST",
+ fetcher=requests,
+ srp=srp,
+ hashfun=srp.SHA256,
+ ng_constant=srp.NG_1024):
+
+ null_check(provider, "provider")
+
+ self.schema = schema
+
+ domain, port = get_https_domain_and_port(provider)
+ self.provider = domain
+ self.port = port
+
+ self.verify = verify
+ self.register_path = register_path
+ self.method = method
+ self.fetcher = fetcher
+ self.srp = srp
+ self.HASHFUN = hashfun
+ self.NG = ng_constant
+
+ self.init_session()
+
+ def init_session(self):
+ self.session = self.fetcher.session()
+
+ def get_registration_uri(self):
+ # XXX assert is https!
+ # use urlparse
+ if self.port:
+ uri = "%s://%s:%s/%s" % (
+ self.schema,
+ self.provider,
+ self.port,
+ self.register_path)
+ else:
+ uri = "%s://%s/%s" % (
+ self.schema,
+ self.provider,
+ self.register_path)
+
+ return uri
+
+ def register_user(self, username, password, keep=False):
+ """
+ @rtype: tuple
+ @rparam: (ok, request)
+ """
+ salt, vkey = self.srp.create_salted_verification_key(
+ username,
+ password,
+ self.HASHFUN,
+ self.NG)
+
+ user_data = {
+ 'user[login]': username,
+ 'user[password_verifier]': binascii.hexlify(vkey),
+ 'user[password_salt]': binascii.hexlify(salt)}
+
+ uri = self.get_registration_uri()
+ logger.debug('post to uri: %s' % uri)
+
+ # XXX get self.method
+ req = self.session.post(
+ uri, data=user_data,
+ timeout=SIGNUP_TIMEOUT,
+ verify=self.verify)
+ # we catch it in the form
+ #req.raise_for_status()
+ return (req.ok, req)
+
+
+class SRPAuth(requests.auth.AuthBase):
+
+ def __init__(self, username, password, server=None, verify=None):
+ # sanity check
+ null_check(server, 'server')
+ self.username = username
+ self.password = password
+ self.server = server
+ self.verify = verify
+
+ logger.debug('SRPAuth. verify=%s' % verify)
+ logger.debug('server: %s. username=%s' % (server, username))
+
+ self.init_data = None
+ self.session = requests.session()
+
+ self.init_srp()
+
+ def init_srp(self):
+ usr = srp.User(
+ self.username,
+ self.password,
+ srp.SHA256,
+ srp.NG_1024)
+ uname, A = usr.start_authentication()
+
+ self.srp_usr = usr
+ self.A = A
+
+ def get_auth_data(self):
+ return {
+ 'login': self.username,
+ 'A': binascii.hexlify(self.A)
+ }
+
+ def get_init_data(self):
+ try:
+ init_session = self.session.post(
+ self.server + '/1/sessions/',
+ data=self.get_auth_data(),
+ verify=self.verify)
+ except requests.exceptions.ConnectionError:
+ raise SRPAuthenticationError(
+ "No connection made (salt).")
+ except:
+ raise SRPAuthenticationError(
+ "Unknown error (salt).")
+ if init_session.status_code not in (200, ):
+ raise SRPAuthenticationError(
+ "No valid response (salt).")
+
+ self.init_data = init_session.json
+ return self.init_data
+
+ def get_server_proof_data(self):
+ try:
+ auth_result = self.session.put(
+ #self.server + '/1/sessions.json/' + self.username,
+ self.server + '/1/sessions/' + self.username,
+ data={'client_auth': binascii.hexlify(self.M)},
+ verify=self.verify)
+ except requests.exceptions.ConnectionError:
+ raise SRPAuthenticationError(
+ "No connection made (HAMK).")
+
+ if auth_result.status_code not in (200, ):
+ raise SRPAuthenticationError(
+ "No valid response (HAMK).")
+
+ self.auth_data = auth_result.json
+ return self.auth_data
+
+ def authenticate(self):
+ logger.debug('start authentication...')
+
+ init_data = self.get_init_data()
+ salt = init_data.get('salt', None)
+ B = init_data.get('B', None)
+
+ # XXX refactor this function
+ # move checks and un-hex
+ # to routines
+
+ if not salt or not B:
+ raise SRPAuthenticationError(
+ "Server did not send initial data.")
+
+ try:
+ unhex_salt = safe_unhexlify(salt)
+ except TypeError:
+ raise SRPAuthenticationError(
+ "Bad data from server (salt)")
+ try:
+ unhex_B = safe_unhexlify(B)
+ except TypeError:
+ raise SRPAuthenticationError(
+ "Bad data from server (B)")
+
+ self.M = self.srp_usr.process_challenge(
+ unhex_salt,
+ unhex_B
+ )
+
+ proof_data = self.get_server_proof_data()
+
+ HAMK = proof_data.get("M2", None)
+ if not HAMK:
+ errors = proof_data.get('errors', None)
+ if errors:
+ logger.error(errors)
+ raise SRPAuthenticationError("Server did not send HAMK.")
+
+ try:
+ unhex_HAMK = safe_unhexlify(HAMK)
+ except TypeError:
+ raise SRPAuthenticationError(
+ "Bad data from server (HAMK)")
+
+ self.srp_usr.verify_session(
+ unhex_HAMK)
+
+ try:
+ assert self.srp_usr.authenticated()
+ logger.debug('user is authenticated!')
+ except (AssertionError):
+ raise SRPAuthenticationError(
+ "Auth verification failed.")
+
+ def __call__(self, req):
+ self.authenticate()
+ req.cookies = self.session.cookies
+ return req
+
+
+def srpauth_protected(user=None, passwd=None, server=None, verify=True):
+ """
+ decorator factory that accepts
+ user and password keyword arguments
+ and add those to the decorated request
+ """
+ def srpauth(fn):
+ def wrapper(*args, **kwargs):
+ if user and passwd:
+ auth = SRPAuth(user, passwd, server, verify)
+ kwargs['auth'] = auth
+ kwargs['verify'] = verify
+ if not args:
+ logger.warning('attempting to get from empty uri!')
+ return fn(*args, **kwargs)
+ return wrapper
+ return srpauth
+
+
+def get_leap_credentials():
+ settings = QtCore.QSettings()
+ full_username = settings.value('username')
+ username, domain = full_username.split('@')
+ seed = settings.value('%s_seed' % domain, None)
+ password = leapkeyring.leap_get_password(full_username, seed=seed)
+ return (username, password)
+
+
+# XXX TODO
+# Pass verify as single argument,
+# in srpauth_protected style
+
+def magick_srpauth(fn):
+ """
+ decorator that gets user and password
+ from the config file and adds those to
+ the decorated request
+ """
+ logger.debug('magick srp auth decorator called')
+
+ def wrapper(*args, **kwargs):
+ #uri = args[0]
+ # XXX Ugh!
+ # Problem with this approach.
+ # This won't work when we're using
+ # api.foo.bar
+ # Unless we keep a table with the
+ # equivalencies...
+ user, passwd = get_leap_credentials()
+
+ # XXX pass verify and server too
+ # (pop)
+ auth = SRPAuth(user, passwd)
+ kwargs['auth'] = auth
+ return fn(*args, **kwargs)
+ return wrapper
+
+
+if __name__ == "__main__":
+ """
+ To test against test_provider (twisted version)
+ Register an user: (will be valid during the session)
+ >>> python auth.py add test password
+
+ Test login with that user:
+ >>> python auth.py login test password
+ """
+
+ import sys
+
+ if len(sys.argv) not in (4, 5):
+ print 'Usage: auth <add|login> <user> <pass> [server]'
+ sys.exit(0)
+
+ action = sys.argv[1]
+ user = sys.argv[2]
+ passwd = sys.argv[3]
+
+ if len(sys.argv) == 5:
+ SERVER = sys.argv[4]
+ else:
+ SERVER = "https://localhost:8443"
+
+ if action == "login":
+
+ @srpauth_protected(
+ user=user, passwd=passwd, server=SERVER, verify=False)
+ def test_srp_protected_get(*args, **kwargs):
+ req = requests.get(*args, **kwargs)
+ req.raise_for_status
+ return req
+
+ #req = test_srp_protected_get('https://localhost:8443/1/cert')
+ req = test_srp_protected_get('%s/1/cert' % SERVER)
+ #print 'cert :', req.content[:200] + "..."
+ print req.content
+ sys.exit(0)
+
+ if action == "add":
+ auth = LeapSRPRegister(provider=SERVER, verify=False)
+ auth.register_user(user, passwd)
diff --git a/src/leap/base/checks.py b/src/leap/base/checks.py
index 84f9dd46..0bf44f59 100644
--- a/src/leap/base/checks.py
+++ b/src/leap/base/checks.py
@@ -1,122 +1,171 @@
# -*- coding: utf-8 -*-
import logging
import platform
+import re
+import socket
import netifaces
-import ping
-import requests
+import sh
from leap.base import constants
from leap.base import exceptions
logger = logging.getLogger(name=__name__)
+_platform = platform.system()
+
+#EVENTS OF NOTE
+EVENT_CONNECT_REFUSED = "[ECONNREFUSED]: Connection refused (code=111)"
+
+ICMP_TARGET = "8.8.8.8"
class LeapNetworkChecker(object):
"""
all network related checks
"""
- # #718
- # XXX get provider gateway as a parameter
- # for constructor.
- # def __init__(self, *args, **kwargs):
- # ...
- # provider_gw = kwargs.pop('provider_gw', None)
- # self.provider_gateway = provider_gw
+ def __init__(self, *args, **kwargs):
+ provider_gw = kwargs.pop('provider_gw', None)
+ self.provider_gateway = provider_gw
def run_all(self, checker=None):
if not checker:
checker = self
- self.error = None # ?
+ #self.error = None # ?
# for MVS
checker.check_tunnel_default_interface()
checker.check_internet_connection()
checker.is_internet_up()
- # XXX We are pinging the default gateway for our connection right?
- # kali: 2012-10-05 20:59 -- I think we should get
- # also the default gateway and ping it instead.
- checker.ping_gateway()
+ if self.provider_gateway:
+ checker.ping_gateway(self.provider_gateway)
- # something like: ?
- # see __init__ above
- # if self.provider_gateway:
- # checker.ping_gateway(self.provider_gateway)
+ checker.parse_log_and_react([], ())
def check_internet_connection(self):
- try:
- # XXX remove this hardcoded random ip
- # ping leap.se or eip provider instead...?
- requests.get('http://216.172.161.165')
-
- except (requests.HTTPError, requests.RequestException) as e:
- raise exceptions.NoInternetConnection(e.message)
- except requests.ConnectionError as e:
- error = "Unidentified Connection Error"
- if e.message == "[Errno 113] No route to host":
+ if _platform == "Linux":
+ try:
+ output = sh.ping("-c", "5", "-w", "5", ICMP_TARGET)
+ # XXX should redirect this to netcheck logger.
+ # and don't clutter main log.
+ logger.debug('Network appears to be up.')
+ except sh.ErrorReturnCode_1 as e:
+ packet_loss = re.findall("\d+% packet loss", e.message)[0]
+ logger.debug("Unidentified Connection Error: " + packet_loss)
if not self.is_internet_up():
error = "No valid internet connection found."
else:
error = "Provider server appears to be down."
- logger.error(error)
- raise exceptions.NoInternetConnection(error)
- logger.debug('Network appears to be up.')
- def is_internet_up(self):
- iface, gateway = self.get_default_interface_gateway()
- self.ping_gateway(self)
+ logger.error(error)
+ raise exceptions.NoInternetConnection(error)
- def check_tunnel_default_interface(self):
- """
- Raises an TunnelNotDefaultRouteError
- (including when no routes are present)
- """
- if not platform.system() == "Linux":
+ else:
raise NotImplementedError
+ def is_internet_up(self):
+ iface, gateway = self.get_default_interface_gateway()
+ try:
+ self.ping_gateway(self.provider_gateway)
+ except exceptions.NoConnectionToGateway:
+ return False
+ return True
+
+ def _get_route_table_linux(self):
+ # do not use context manager, tests pass a StringIO
f = open("/proc/net/route")
route_table = f.readlines()
f.close()
#toss out header
route_table.pop(0)
-
if not route_table:
- raise exceptions.TunnelNotDefaultRouteError()
+ raise exceptions.NoDefaultInterfaceFoundError
+ return route_table
+ def _get_def_iface_osx(self):
+ default_iface = None
+ #gateway = None
+ routes = list(sh.route('-n', 'get', ICMP_TARGET, _iter=True))
+ iface = filter(lambda l: "interface" in l, routes)
+ if not iface:
+ return None, None
+ def_ifacel = re.findall('\w+\d', iface[0])
+ default_iface = def_ifacel[0] if def_ifacel else None
+ if not default_iface:
+ return None, None
+ _gw = filter(lambda l: "gateway" in l, routes)
+ gw = re.findall('\d+\.\d+\.\d+\.\d+', _gw[0])[0]
+ return default_iface, gw
+
+ def _get_tunnel_iface_linux(self):
+ # XXX review.
+ # valid also when local router has a default entry?
+ route_table = self._get_route_table_linux()
line = route_table.pop(0)
iface, destination = line.split('\t')[0:2]
if not destination == '00000000' or not iface == 'tun0':
raise exceptions.TunnelNotDefaultRouteError()
+ return True
- def get_default_interface_gateway(self):
- """only impletemented for linux so far."""
- if not platform.system() == "Linux":
+ def check_tunnel_default_interface(self):
+ """
+ Raises an TunnelNotDefaultRouteError
+ if tun0 is not the chosen default route
+ (including when no routes are present)
+ """
+ #logger.debug('checking tunnel default interface...')
+
+ if _platform == "Linux":
+ valid = self._get_tunnel_iface_linux()
+ return valid
+ elif _platform == "Darwin":
+ default_iface, gw = self._get_def_iface_osx()
+ #logger.debug('iface: %s', default_iface)
+ if default_iface != "tun0":
+ logger.debug('tunnel not default route! gw: %s', default_iface)
+ # XXX should catch this and act accordingly...
+ # but rather, this test should only be launched
+ # when we have successfully completed a connection
+ # ... TRIGGER: Connection stablished (or whatever it is)
+ # in the logs
+ raise exceptions.TunnelNotDefaultRouteError
+ else:
+ #logger.debug('PLATFORM !!! %s', _platform)
raise NotImplementedError
- # XXX use psutil
- f = open("/proc/net/route")
- route_table = f.readlines()
- f.close()
- #toss out header
- route_table.pop(0)
-
+ def _get_def_iface_linux(self):
default_iface = None
gateway = None
+
+ route_table = self._get_route_table_linux()
while route_table:
line = route_table.pop(0)
iface, destination, gateway = line.split('\t')[0:3]
if destination == '00000000':
default_iface = iface
break
+ return default_iface, gateway
+
+ def get_default_interface_gateway(self):
+ """
+ gets the interface we are going thru.
+ (this should be merged with check tunnel default interface,
+ imo...)
+ """
+ if _platform == "Linux":
+ default_iface, gw = self._get_def_iface_linux()
+ elif _platform == "Darwin":
+ default_iface, gw = self._get_def_iface_osx()
+ else:
+ raise NotImplementedError
if not default_iface:
raise exceptions.NoDefaultInterfaceFoundError
if default_iface not in netifaces.interfaces():
raise exceptions.InterfaceNotFoundError
-
- return default_iface, gateway
+ logger.debug('-- default iface %s', default_iface)
+ return default_iface, gw
def ping_gateway(self, gateway):
# TODO: Discuss how much packet loss (%) is acceptable.
@@ -125,15 +174,40 @@ class LeapNetworkChecker(object):
# -- is it a valid ip? (there's something in util)
# -- is it a domain?
# -- can we resolve? -- raise NoDNSError if not.
- packet_loss = ping.quiet_ping(gateway)[0]
+
+ # XXX -- sh.ping implemtation needs review!
+ try:
+ output = sh.ping("-c", "10", gateway).stdout
+ except sh.ErrorReturnCode_1 as e:
+ output = e.message
+ finally:
+ packet_loss = int(re.findall("(\d+)% packet loss", output)[0])
+
+ logger.debug('packet loss %s%%' % packet_loss)
if packet_loss > constants.MAX_ICMP_PACKET_LOSS:
raise exceptions.NoConnectionToGateway
- # XXX check for name resolution servers
- # dunno what's the best way to do this...
- # check for etc/resolv entries or similar?
- # just try to resolve?
- # is there something in psutil?
+ def check_name_resolution(self, domain_name):
+ try:
+ socket.gethostbyname(domain_name)
+ return True
+ except socket.gaierror:
+ raise exceptions.CannotResolveDomainError
- # def check_name_resolution(self):
- # pass
+ def parse_log_and_react(self, log, error_matrix=None):
+ """
+ compares the recent openvpn status log to
+ strings passed in and executes the callbacks passed in.
+ @param log: openvpn log
+ @type log: list of strings
+ @param error_matrix: tuples of strings and tuples of callbacks
+ @type error_matrix: tuples strings and call backs
+ """
+ for line in log:
+ # we could compile a regex here to save some cycles up -- kali
+ for each in error_matrix:
+ error, callbacks = each
+ if error in line:
+ for cb in callbacks:
+ if callable(cb):
+ cb()
diff --git a/src/leap/base/config.py b/src/leap/base/config.py
index cf01d1aa..6a13db7d 100644
--- a/src/leap/base/config.py
+++ b/src/leap/base/config.py
@@ -4,12 +4,15 @@ Configuration Base Class
import grp
import json
import logging
+import re
import socket
-import tempfile
+import time
import os
logger = logging.getLogger(name=__name__)
+from dateutil import parser as dateparser
+from xdg import BaseDirectory
import requests
from leap.base import exceptions
@@ -118,23 +121,50 @@ class JSONLeapConfig(BaseLeapConfig):
" derived class")
assert issubclass(self.spec, PluggableConfig)
+ self.domain = kwargs.pop('domain', None)
self._config = self.spec(format="json")
self._config.load()
self.fetcher = kwargs.pop('fetcher', requests)
# mandatory baseconfig interface
- def save(self, to=None):
- if to is None:
- to = self.filename
- folder, filename = os.path.split(to)
- if folder and not os.path.isdir(folder):
- mkdir_p(folder)
- self._config.serialize(to)
+ def save(self, to=None, force=False):
+ """
+ force param will skip the dirty check.
+ :type force: bool
+ """
+ # XXX this force=True does not feel to right
+ # but still have to look for a better way
+ # of dealing with dirtiness and the
+ # trick of loading remote config only
+ # when newer.
+
+ if force:
+ do_save = True
+ else:
+ do_save = self._config.is_dirty()
+
+ if do_save:
+ if to is None:
+ to = self.filename
+ folder, filename = os.path.split(to)
+ if folder and not os.path.isdir(folder):
+ mkdir_p(folder)
+ self._config.serialize(to)
+ return True
+
+ else:
+ return False
+
+ def load(self, fromfile=None, from_uri=None, fetcher=None,
+ force_download=False, verify=True):
- def load(self, fromfile=None, from_uri=None, fetcher=None, verify=False):
if from_uri is not None:
- fetched = self.fetch(from_uri, fetcher=fetcher, verify=verify)
+ fetched = self.fetch(
+ from_uri,
+ fetcher=fetcher,
+ verify=verify,
+ force_dl=force_download)
if fetched:
return
if fromfile is None:
@@ -145,33 +175,68 @@ class JSONLeapConfig(BaseLeapConfig):
logger.error('tried to load config from non-existent path')
logger.error('Not Found: %s', fromfile)
- def fetch(self, uri, fetcher=None, verify=True):
+ def fetch(self, uri, fetcher=None, verify=True, force_dl=False):
if not fetcher:
fetcher = self.fetcher
- logger.debug('verify: %s', verify)
- logger.debug('uri: %s', uri)
- request = fetcher.get(uri, verify=verify)
- # XXX should send a if-modified-since header
- # XXX get 404, ...
- # and raise a UnableToFetch...
+ logger.debug('uri: %s (verify: %s)' % (uri, verify))
+
+ rargs = (uri, )
+ rkwargs = {'verify': verify}
+ headers = {}
+
+ curmtime = self.get_mtime() if not force_dl else None
+ if curmtime:
+ logger.debug('requesting with if-modified-since %s' % curmtime)
+ headers['if-modified-since'] = curmtime
+ rkwargs['headers'] = headers
+
+ #request = fetcher.get(uri, verify=verify)
+ request = fetcher.get(*rargs, **rkwargs)
request.raise_for_status()
- fd, fname = tempfile.mkstemp(suffix=".json")
- if request.json:
- self._config.load(json.dumps(request.json))
+ if request.status_code == 304:
+ logger.debug('...304 Not Changed')
+ # On this point, we have to assume that
+ # we HAD the filename. If that filename is corruct,
+ # we should enforce a force_download in the load
+ # method above.
+ self._config.load(fromfile=self.filename)
+ return True
+ if request.json:
+ mtime = None
+ last_modified = request.headers.get('last-modified', None)
+ if last_modified:
+ _mtime = dateparser.parse(last_modified)
+ mtime = int(_mtime.strftime("%s"))
+ if callable(request.json):
+ _json = request.json()
+ else:
+ # back-compat
+ _json = request.json
+ self._config.load(json.dumps(_json), mtime=mtime)
+ self._config.set_dirty()
else:
# not request.json
# might be server did not announce content properly,
# let's try deserializing all the same.
try:
self._config.load(request.content)
+ self._config.set_dirty()
except ValueError:
raise eipexceptions.LeapBadConfigFetchedError
return True
+ def get_mtime(self):
+ try:
+ _mtime = os.stat(self.filename)[8]
+ mtime = time.strftime("%c GMT", time.gmtime(_mtime))
+ return mtime
+ except OSError:
+ return None
+
def get_config(self):
return self._config.config
@@ -216,15 +281,16 @@ def get_config_dir():
@rparam: config path
@rtype: string
"""
- # TODO
- # check for $XDG_CONFIG_HOME var?
- # get a more sensible path for win/mac
- # kclair: opinion? ^^
-
- return os.path.expanduser(
- os.path.join('~',
- '.config',
- 'leap'))
+ home = os.path.expanduser("~")
+ if re.findall("leap_tests-[a-zA-Z0-9]{6}", home):
+ # we're inside a test! :)
+ return os.path.join(home, ".config/leap")
+ else:
+ # XXX dirspec is cross-platform,
+ # we should borrow some of those
+ # routines for osx/win and wrap this call.
+ return os.path.join(BaseDirectory.xdg_config_home,
+ 'leap')
def get_config_file(filename, folder=None):
@@ -252,6 +318,15 @@ def get_default_provider_path():
return default_provider_path
+def get_provider_path(domain):
+ # XXX if not domain, return get_default_provider_path
+ default_subpath = os.path.join("providers", domain)
+ provider_path = get_config_file(
+ '',
+ folder=default_subpath)
+ return provider_path
+
+
def validate_ip(ip_str):
"""
raises exception if the ip_str is
@@ -261,7 +336,11 @@ def validate_ip(ip_str):
def get_username():
- return os.getlogin()
+ try:
+ return os.getlogin()
+ except OSError as e:
+ import pwd
+ return pwd.getpwuid(os.getuid())[0]
def get_groupname():
diff --git a/src/leap/base/connection.py b/src/leap/base/connection.py
index e478538d..41d13935 100644
--- a/src/leap/base/connection.py
+++ b/src/leap/base/connection.py
@@ -37,11 +37,11 @@ class Connection(Authentication):
"""
pass
- def shutdown(self):
- """
- shutdown and quit
- """
- self.desired_con_state = self.status.DISCONNECTED
+ #def shutdown(self):
+ #"""
+ #shutdown and quit
+ #"""
+ #self.desired_con_state = self.status.DISCONNECTED
def connection_state(self):
"""
diff --git a/src/leap/base/constants.py b/src/leap/base/constants.py
index f7be8d98..f5665e5f 100644
--- a/src/leap/base/constants.py
+++ b/src/leap/base/constants.py
@@ -1,6 +1,7 @@
"""constants to be used in base module"""
from leap import __branding
-APP_NAME = __branding.get("short_name", "leap")
+APP_NAME = __branding.get("short_name", "leap-client")
+OPENVPN_BIN = "openvpn"
# default provider placeholder
# using `example.org` we make sure that this
@@ -14,18 +15,27 @@ DEFAULT_PROVIDER = __branding.get(
DEFINITION_EXPECTED_PATH = "provider.json"
DEFAULT_PROVIDER_DEFINITION = {
- u'api_uri': u'https://api.%s/' % DEFAULT_PROVIDER,
- u'api_version': u'0.1.0',
- u'ca_cert_fingerprint': u'8aab80ae4326fd30721689db813733783fe0bd7e',
- u'ca_cert_uri': u'https://%s/cacert.pem' % DEFAULT_PROVIDER,
- u'description': {u'en': u'This is a test provider'},
- u'display_name': {u'en': u'Test Provider'},
- u'domain': u'%s' % DEFAULT_PROVIDER,
- u'enrollment_policy': u'open',
- u'public_key': u'cb7dbd679f911e85bc2e51bd44afd7308ee19c21',
- u'serial': 1,
- u'services': [u'eip'],
- u'version': u'0.1.0'}
+ u"api_uri": "https://api.%s/" % DEFAULT_PROVIDER,
+ u"api_version": u"1",
+ u"ca_cert_fingerprint": "SHA256: fff",
+ u"ca_cert_uri": u"https://%s/ca.crt" % DEFAULT_PROVIDER,
+ u"default_language": u"en",
+ u"description": {
+ u"en": u"A demonstration service provider using the LEAP platform"
+ },
+ u"domain": "%s" % DEFAULT_PROVIDER,
+ u"enrollment_policy": u"open",
+ u"languages": [
+ u"en"
+ ],
+ u"name": {
+ u"en": u"Test Provider"
+ },
+ u"services": [
+ "openvpn"
+ ]
+}
+
MAX_ICMP_PACKET_LOSS = 10
diff --git a/src/leap/base/exceptions.py b/src/leap/base/exceptions.py
index f12a49d5..2e31b33b 100644
--- a/src/leap/base/exceptions.py
+++ b/src/leap/base/exceptions.py
@@ -14,6 +14,7 @@ Exception attributes and their meaning/uses
* usermessage: the message that will be passed to user in ErrorDialogs
in Qt-land.
"""
+from leap.util.translations import translate
class LeapException(Exception):
@@ -22,6 +23,7 @@ class LeapException(Exception):
sets some parameters that we will check
during error checking routines
"""
+
critical = False
failfirst = False
warning = False
@@ -46,27 +48,50 @@ class ImproperlyConfigured(Exception):
pass
-class NoDefaultInterfaceFoundError(LeapException):
- message = "no default interface found"
- usermessage = "Looks like your computer is not connected to the internet"
+# NOTE: "Errors" (context) has to be a explicit string!
class InterfaceNotFoundError(LeapException):
# XXX should take iface arg on init maybe?
message = "interface not found"
+ usermessage = translate(
+ "Errors",
+ "Interface not found")
+
+
+class NoDefaultInterfaceFoundError(LeapException):
+ message = "no default interface found"
+ usermessage = translate(
+ "Errors",
+ "Looks like your computer "
+ "is not connected to the internet")
class NoConnectionToGateway(CriticalError):
message = "no connection to gateway"
- usermessage = "Looks like there are problems with your internet connection"
+ usermessage = translate(
+ "Errors",
+ "Looks like there are problems "
+ "with your internet connection")
class NoInternetConnection(CriticalError):
message = "No Internet connection found"
- usermessage = "It looks like there is no internet connection."
+ usermessage = translate(
+ "Errors",
+ "It looks like there is no internet connection.")
# and now we try to connect to our web to troubleshoot LOL :P
-class TunnelNotDefaultRouteError(CriticalError):
+class CannotResolveDomainError(LeapException):
+ message = "Cannot resolve domain"
+ usermessage = translate(
+ "Errors",
+ "Domain cannot be found")
+
+
+class TunnelNotDefaultRouteError(LeapException):
message = "Tunnel connection dissapeared. VPN down?"
- usermessage = "The Encrypted Connection was lost. Shutting down..."
+ usermessage = translate(
+ "Errors",
+ "The Encrypted Connection was lost.")
diff --git a/src/leap/base/network.py b/src/leap/base/network.py
index e90139c4..d841e692 100644
--- a/src/leap/base/network.py
+++ b/src/leap/base/network.py
@@ -3,9 +3,11 @@ from __future__ import (print_function)
import logging
import threading
+from leap.eip import config as eipconfig
from leap.base.checks import LeapNetworkChecker
from leap.base.constants import ROUTE_CHECK_INTERVAL
from leap.base.exceptions import TunnelNotDefaultRouteError
+from leap.util.misc import null_check
from leap.util.coroutines import (launch_thread, process_events)
from time import sleep
@@ -19,23 +21,34 @@ class NetworkCheckerThread(object):
connection.
"""
def __init__(self, *args, **kwargs):
+
self.status_signals = kwargs.pop('status_signals', None)
- #self.watcher_cb = kwargs.pop('status_signals', None)
self.error_cb = kwargs.pop(
'error_cb',
lambda exc: logger.error("%s", exc.message))
self.shutdown = threading.Event()
- # XXX get provider_gateway and pass it to checker
- # see in eip.config for function
- # #718
- self.checker = LeapNetworkChecker()
+ # XXX get provider passed here
+ provider = kwargs.pop('provider', None)
+ null_check(provider, 'provider')
+
+ eipconf = eipconfig.EIPConfig(domain=provider)
+ eipconf.load()
+ eipserviceconf = eipconfig.EIPServiceConfig(domain=provider)
+ eipserviceconf.load()
+
+ gw = eipconfig.get_eip_gateway(
+ eipconfig=eipconf,
+ eipserviceconfig=eipserviceconf)
+ self.checker = LeapNetworkChecker(
+ provider_gw=gw)
def start(self):
self.process_handle = self._launch_recurrent_network_checks(
(self.error_cb,))
def stop(self):
+ self.process_handle.join(timeout=0.1)
self.shutdown.set()
logger.debug("network checked stopped.")
@@ -47,6 +60,7 @@ class NetworkCheckerThread(object):
#here all the observers in fail_callbacks expect one positional argument,
#which is exception so we can try by passing a lambda with logger to
#check it works.
+
def _network_checks_thread(self, fail_callbacks):
#TODO: replace this with waiting for a signal from openvpn
while True:
@@ -55,11 +69,17 @@ class NetworkCheckerThread(object):
break
except TunnelNotDefaultRouteError:
# XXX ??? why do we sleep here???
+ # aa: If the openvpn isn't up and running yet,
+ # let's give it a moment to breath.
+ #logger.error('NOT DEFAULT ROUTE!----')
+ # Instead of this, we should flag when the
+ # iface IS SUPPOSED to be up imo. -- kali
sleep(1)
fail_observer_dict = dict(((
observer,
process_events(observer)) for observer in fail_callbacks))
+
while not self.shutdown.is_set():
try:
self.checker.check_tunnel_default_interface()
@@ -69,11 +89,18 @@ class NetworkCheckerThread(object):
for obs in fail_observer_dict:
fail_observer_dict[obs].send(exc)
sleep(ROUTE_CHECK_INTERVAL)
+
#reset event
+ # I see a problem with this. You cannot stop it, it
+ # resets itself forever. -- kali
+
+ # XXX use QTimer for the recurrent triggers,
+ # and ditch the sleeps.
+ logger.debug('resetting event')
self.shutdown.clear()
def _launch_recurrent_network_checks(self, fail_callbacks):
- #we need to wrap the fail callback in a tuple
+ # XXX reimplement using QTimer -- kali
watcher = launch_thread(
self._network_checks_thread,
(fail_callbacks,))
diff --git a/src/leap/base/pluggableconfig.py b/src/leap/base/pluggableconfig.py
index b8615ad8..3517db6b 100644
--- a/src/leap/base/pluggableconfig.py
+++ b/src/leap/base/pluggableconfig.py
@@ -10,6 +10,8 @@ import urlparse
import jsonschema
+from leap.util.translations import LEAPTranslatable
+
logger = logging.getLogger(__name__)
@@ -118,7 +120,6 @@ adaptors['json'] = JSONAdaptor()
# to proper python types.
# TODO:
-# - multilingual object.
# - HTTPS uri
@@ -132,6 +133,20 @@ class DateType(object):
return time.strftime(self.fmt, data)
+class TranslatableType(object):
+ """
+ a type that casts to LEAPTranslatable objects.
+ Used for labels we get from providers and stuff.
+ """
+
+ def to_python(self, data):
+ return LEAPTranslatable(data)
+
+ # needed? we already have an extended dict...
+ #def get_prep_value(self, data):
+ #return dict(data)
+
+
class URIType(object):
def to_python(self, data):
@@ -164,6 +179,7 @@ types = {
'date': DateType(),
'uri': URIType(),
'https-uri': HTTPSURIType(),
+ 'translatable': TranslatableType(),
}
@@ -180,6 +196,8 @@ class PluggableConfig(object):
self.adaptors = adaptors
self.types = types
self._format = format
+ self.mtime = None
+ self.dirty = False
@property
def option_dict(self):
@@ -319,6 +337,13 @@ class PluggableConfig(object):
serializable = self.prep_value(config)
adaptor.write(serializable, filename)
+ if self.mtime:
+ self.touch_mtime(filename)
+
+ def touch_mtime(self, filename):
+ mtime = self.mtime
+ os.utime(filename, (mtime, mtime))
+
def deserialize(self, string=None, fromfile=None, format=None):
"""
load configuration from a file or string
@@ -364,6 +389,12 @@ class PluggableConfig(object):
content = _try_deserialize()
return content
+ def set_dirty(self):
+ self.dirty = True
+
+ def is_dirty(self):
+ return self.dirty
+
def load(self, *args, **kwargs):
"""
load from string or file
@@ -373,6 +404,8 @@ class PluggableConfig(object):
"""
string = args[0] if args else None
fromfile = kwargs.get("fromfile", None)
+ mtime = kwargs.pop("mtime", None)
+ self.mtime = mtime
content = None
# start with defaults, so we can
@@ -402,7 +435,8 @@ class PluggableConfig(object):
return True
-def testmain():
+def testmain(): # pragma: no cover
+
from tests import test_validation as t
import pprint
diff --git a/src/leap/base/providers.py b/src/leap/base/providers.py
index 7b219cc7..d41f3695 100644
--- a/src/leap/base/providers.py
+++ b/src/leap/base/providers.py
@@ -7,20 +7,20 @@ class LeapProviderDefinition(baseconfig.JSONLeapConfig):
spec = specs.leap_provider_spec
def _get_slug(self):
- provider_path = baseconfig.get_default_provider_path()
+ domain = getattr(self, 'domain', None)
+ if domain:
+ path = baseconfig.get_provider_path(domain)
+ else:
+ path = baseconfig.get_default_provider_path()
+
return baseconfig.get_config_file(
- 'provider.json',
- folder=provider_path)
+ 'provider.json', folder=path)
def _set_slug(self, *args, **kwargs):
raise AttributeError("you cannot set slug")
slug = property(_get_slug, _set_slug)
- # TODO (MVS+)
- # we will construct slug from providers/%s/definition.json
- # where %s is domain name. we can get that on __init__
-
class LeapProviderSet(object):
# we gather them from the filesystem
diff --git a/src/leap/base/specs.py b/src/leap/base/specs.py
index b4bb8dcf..f57d7e9c 100644
--- a/src/leap/base/specs.py
+++ b/src/leap/base/specs.py
@@ -2,28 +2,36 @@ leap_provider_spec = {
'description': 'provider definition',
'type': 'object',
'properties': {
- 'serial': {
- 'type': int,
- 'default': 1,
- 'required': True,
- },
+ #'serial': {
+ #'type': int,
+ #'default': 1,
+ #'required': True,
+ #},
'version': {
'type': unicode,
'default': '0.1.0'
#'required': True
},
+ "default_language": {
+ 'type': unicode,
+ 'default': 'en'
+ },
'domain': {
'type': unicode, # XXX define uri type
'default': 'testprovider.example.org'
#'required': True,
},
- 'display_name': {
- 'type': dict, # XXX multilingual object?
+ 'name': {
+ #'type': LEAPTranslatable,
+ 'type': dict,
+ 'format': 'translatable',
'default': {u'en': u'Test Provider'}
#'required': True
},
'description': {
+ #'type': LEAPTranslatable,
'type': dict,
+ 'format': 'translatable',
'default': {u'en': u'Test provider'}
},
'enrollment_policy': {
diff --git a/src/leap/base/tests/__init__.py b/src/leap/base/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/src/leap/base/tests/__init__.py
diff --git a/src/leap/base/tests/test_auth.py b/src/leap/base/tests/test_auth.py
new file mode 100644
index 00000000..b3009a9b
--- /dev/null
+++ b/src/leap/base/tests/test_auth.py
@@ -0,0 +1,58 @@
+from BaseHTTPServer import BaseHTTPRequestHandler
+import urlparse
+try:
+ import unittest2 as unittest
+except ImportError:
+ import unittest
+
+import requests
+#from mock import Mock
+
+from leap.base import auth
+#from leap.base import exceptions
+from leap.eip.tests.test_checks import NoLogRequestHandler
+from leap.testing.basetest import BaseLeapTest
+from leap.testing.https_server import BaseHTTPSServerTestCase
+
+
+class LeapSRPRegisterTests(BaseHTTPSServerTestCase, BaseLeapTest):
+ __name__ = "leap_srp_register_test"
+ provider = "testprovider.example.org"
+
+ class request_handler(NoLogRequestHandler, BaseHTTPRequestHandler):
+ responses = {
+ '/': ['OK', '']}
+
+ def do_GET(self):
+ path = urlparse.urlparse(self.path)
+ message = '\n'.join(self.responses.get(
+ path.path, None))
+ self.send_response(200)
+ self.end_headers()
+ self.wfile.write(message)
+
+ def setUp(self):
+ pass
+
+ def tearDown(self):
+ pass
+
+ def test_srp_auth_should_implement_check_methods(self):
+ SERVER = "https://localhost:8443"
+ srp_auth = auth.LeapSRPRegister(provider=SERVER, verify=False)
+
+ self.assertTrue(hasattr(srp_auth, "init_session"),
+ "missing meth")
+ self.assertTrue(hasattr(srp_auth, "get_registration_uri"),
+ "missing meth")
+ self.assertTrue(hasattr(srp_auth, "register_user"),
+ "missing meth")
+
+ def test_srp_auth_basic_functionality(self):
+ SERVER = "https://localhost:8443"
+ srp_auth = auth.LeapSRPRegister(provider=SERVER, verify=False)
+
+ self.assertIsInstance(srp_auth.session, requests.sessions.Session)
+ self.assertEqual(
+ srp_auth.get_registration_uri(),
+ "https://localhost:8443/1/users")
diff --git a/src/leap/base/tests/test_checks.py b/src/leap/base/tests/test_checks.py
index bec09ce6..8126755b 100644
--- a/src/leap/base/tests/test_checks.py
+++ b/src/leap/base/tests/test_checks.py
@@ -3,13 +3,11 @@ try:
except ImportError:
import unittest
import os
+import sh
from mock import (patch, Mock)
from StringIO import StringIO
-import ping
-import requests
-
from leap.base import checks
from leap.base import exceptions
from leap.testing.basetest import BaseLeapTest
@@ -21,6 +19,7 @@ class LeapNetworkCheckTest(BaseLeapTest):
__name__ = "leap_network_check_tests"
def setUp(self):
+ os.environ['PATH'] += ':/bin'
pass
def tearDown(self):
@@ -37,16 +36,27 @@ class LeapNetworkCheckTest(BaseLeapTest):
"missing meth")
self.assertTrue(hasattr(checker, "ping_gateway"),
"missing meth")
+ self.assertTrue(hasattr(checker, "parse_log_and_react"),
+ "missing meth")
def test_checker_should_actually_call_all_tests(self):
checker = checks.LeapNetworkChecker()
+ mc = Mock()
+ checker.run_all(checker=mc)
+ self.assertTrue(mc.check_internet_connection.called, "not called")
+ self.assertTrue(mc.check_tunnel_default_interface.called, "not called")
+ self.assertTrue(mc.is_internet_up.called, "not called")
+ self.assertTrue(mc.parse_log_and_react.called, "not called")
+ # ping gateway only called if we pass provider_gw
+ checker = checks.LeapNetworkChecker(provider_gw="0.0.0.0")
mc = Mock()
checker.run_all(checker=mc)
self.assertTrue(mc.check_internet_connection.called, "not called")
self.assertTrue(mc.check_tunnel_default_interface.called, "not called")
self.assertTrue(mc.ping_gateway.called, "not called")
self.assertTrue(mc.is_internet_up.called, "not called")
+ self.assertTrue(mc.parse_log_and_react.called, "not called")
def test_get_default_interface_no_interface(self):
checker = checks.LeapNetworkChecker()
@@ -65,14 +75,6 @@ class LeapNetworkCheckTest(BaseLeapTest):
mock_open.return_value = StringIO(
"Iface\tDestination Gateway\t"
"Flags\tRefCntd\tUse\tMetric\t"
- "Mask\tMTU\tWindow\tIRTT")
- checker.check_tunnel_default_interface()
-
- with patch('leap.base.checks.open', create=True) as mock_open:
- with self.assertRaises(exceptions.TunnelNotDefaultRouteError):
- mock_open.return_value = StringIO(
- "Iface\tDestination Gateway\t"
- "Flags\tRefCntd\tUse\tMetric\t"
"Mask\tMTU\tWindow\tIRTT\n"
"wlan0\t00000000\t0102A8C0\t"
"0003\t0\t0\t0\t00000000\t0\t0\t0")
@@ -88,30 +90,88 @@ class LeapNetworkCheckTest(BaseLeapTest):
def test_ping_gateway_fail(self):
checker = checks.LeapNetworkChecker()
- with patch.object(ping, "quiet_ping") as mocked_ping:
+ with patch.object(sh, "ping") as mocked_ping:
with self.assertRaises(exceptions.NoConnectionToGateway):
- mocked_ping.return_value = [11, "", ""]
+ mocked_ping.return_value = Mock
+ mocked_ping.return_value.stdout = "11% packet loss"
checker.ping_gateway("4.2.2.2")
- def test_check_internet_connection_failures(self):
+ def test_ping_gateway(self):
checker = checks.LeapNetworkChecker()
- with patch.object(requests, "get") as mocked_get:
- mocked_get.side_effect = requests.HTTPError
- with self.assertRaises(exceptions.NoInternetConnection):
- checker.check_internet_connection()
+ with patch.object(sh, "ping") as mocked_ping:
+ mocked_ping.return_value = Mock
+ mocked_ping.return_value.stdout = """
+PING 4.2.2.2 (4.2.2.2) 56(84) bytes of data.
+64 bytes from 4.2.2.2: icmp_req=1 ttl=54 time=33.8 ms
+64 bytes from 4.2.2.2: icmp_req=2 ttl=54 time=30.6 ms
+64 bytes from 4.2.2.2: icmp_req=3 ttl=54 time=31.4 ms
+64 bytes from 4.2.2.2: icmp_req=4 ttl=54 time=36.1 ms
+64 bytes from 4.2.2.2: icmp_req=5 ttl=54 time=30.8 ms
+64 bytes from 4.2.2.2: icmp_req=6 ttl=54 time=30.4 ms
+64 bytes from 4.2.2.2: icmp_req=7 ttl=54 time=30.7 ms
+64 bytes from 4.2.2.2: icmp_req=8 ttl=54 time=32.7 ms
+64 bytes from 4.2.2.2: icmp_req=9 ttl=54 time=31.4 ms
+64 bytes from 4.2.2.2: icmp_req=10 ttl=54 time=33.3 ms
+
+--- 4.2.2.2 ping statistics ---
+10 packets transmitted, 10 received, 0% packet loss, time 9016ms
+rtt min/avg/max/mdev = 30.497/32.172/36.161/1.755 ms"""
+ checker.ping_gateway("4.2.2.2")
- with patch.object(requests, "get") as mocked_get:
- mocked_get.side_effect = requests.RequestException
+ def test_check_internet_connection_failures(self):
+ checker = checks.LeapNetworkChecker()
+ TimeoutError = get_ping_timeout_error()
+ with patch.object(sh, "ping") as mocked_ping:
+ mocked_ping.side_effect = TimeoutError
with self.assertRaises(exceptions.NoInternetConnection):
- checker.check_internet_connection()
+ with patch.object(checker, "ping_gateway") as mock_gateway:
+ mock_gateway.side_effect = exceptions.NoConnectionToGateway
+ checker.check_internet_connection()
- #TODO: Mock possible errors that can be raised by is_internet_up
- with patch.object(requests, "get") as mocked_get:
- mocked_get.side_effect = requests.ConnectionError
+ with patch.object(sh, "ping") as mocked_ping:
+ mocked_ping.side_effect = TimeoutError
with self.assertRaises(exceptions.NoInternetConnection):
- checker.check_internet_connection()
+ with patch.object(checker, "ping_gateway") as mock_gateway:
+ mock_gateway.return_value = True
+ checker.check_internet_connection()
- @unittest.skipUnless(_uid == 0, "root only")
- def test_ping_gateway(self):
+ def test_parse_log_and_react(self):
checker = checks.LeapNetworkChecker()
- checker.ping_gateway("4.2.2.2")
+ to_call = Mock()
+ log = [("leap.openvpn - INFO - Mon Nov 19 13:36:24 2012 "
+ "read UDPv4 [ECONNREFUSED]: Connection refused (code=111)")]
+ err_matrix = [(checks.EVENT_CONNECT_REFUSED, (to_call, ))]
+ checker.parse_log_and_react(log, err_matrix)
+ self.assertTrue(to_call.called)
+
+ log = [("2012-11-19 13:36:26,177 - leap.openvpn - INFO - "
+ "Mon Nov 19 13:36:24 2012 ERROR: Linux route delete command "
+ "failed: external program exited"),
+ ("2012-11-19 13:36:26,178 - leap.openvpn - INFO - "
+ "Mon Nov 19 13:36:24 2012 ERROR: Linux route delete command "
+ "failed: external program exited"),
+ ("2012-11-19 13:36:26,180 - leap.openvpn - INFO - "
+ "Mon Nov 19 13:36:24 2012 ERROR: Linux route delete command "
+ "failed: external program exited"),
+ ("2012-11-19 13:36:26,181 - leap.openvpn - INFO - "
+ "Mon Nov 19 13:36:24 2012 /sbin/ifconfig tun0 0.0.0.0"),
+ ("2012-11-19 13:36:26,182 - leap.openvpn - INFO - "
+ "Mon Nov 19 13:36:24 2012 Linux ip addr del failed: external "
+ "program exited with error stat"),
+ ("2012-11-19 13:36:26,183 - leap.openvpn - INFO - "
+ "Mon Nov 19 13:36:26 2012 SIGTERM[hard,] received, process"
+ "exiting"), ]
+ to_call.reset_mock()
+ checker.parse_log_and_react(log, err_matrix)
+ self.assertFalse(to_call.called)
+
+ to_call.reset_mock()
+ checker.parse_log_and_react([], err_matrix)
+ self.assertFalse(to_call.called)
+
+
+def get_ping_timeout_error():
+ try:
+ sh.ping("-c", "1", "-w", "1", "8.8.7.7")
+ except Exception as e:
+ return e
diff --git a/src/leap/base/tests/test_providers.py b/src/leap/base/tests/test_providers.py
index 8d3b8847..f257f54d 100644
--- a/src/leap/base/tests/test_providers.py
+++ b/src/leap/base/tests/test_providers.py
@@ -8,18 +8,22 @@ import os
import jsonschema
-from leap import __branding as BRANDING
+#from leap import __branding as BRANDING
from leap.testing.basetest import BaseLeapTest
from leap.base import providers
EXPECTED_DEFAULT_CONFIG = {
u"api_version": u"0.1.0",
- u"description": {u'en': u"Test provider"},
- u"display_name": {u'en': u"Test Provider"},
+ #u"description": "LEAPTranslatable<{u'en': u'Test provider'}>",
+ u"description": {u'en': u'Test provider'},
+ u"default_language": u"en",
+ #u"display_name": {u'en': u"Test Provider"},
u"domain": u"testprovider.example.org",
+ #u'name': "LEAPTranslatable<{u'en': u'Test Provider'}>",
+ u'name': {u'en': u'Test Provider'},
u"enrollment_policy": u"open",
- u"serial": 1,
+ #u"serial": 1,
u"services": [
u"eip"
],
@@ -30,9 +34,11 @@ EXPECTED_DEFAULT_CONFIG = {
class TestLeapProviderDefinition(BaseLeapTest):
def setUp(self):
- self.definition = providers.LeapProviderDefinition()
- self.definition.save()
- self.definition.load()
+ self.domain = "testprovider.example.org"
+ self.definition = providers.LeapProviderDefinition(
+ domain=self.domain)
+ self.definition.save(force=True)
+ self.definition.load() # why have to load after save??
self.config = self.definition.config
def tearDown(self):
@@ -51,7 +57,7 @@ class TestLeapProviderDefinition(BaseLeapTest):
os.path.join(
self.home,
'.config', 'leap', 'providers',
- '%s' % BRANDING.get('provider_domain'),
+ '%s' % self.domain,
'provider.json'))
with self.assertRaises(AttributeError):
self.definition.slug = 23
@@ -59,9 +65,10 @@ class TestLeapProviderDefinition(BaseLeapTest):
def test_provider_dump(self):
# check a good provider definition is dumped to disk
self.testfile = self.get_tempfile('test.json')
- self.definition.save(to=self.testfile)
+ self.definition.save(to=self.testfile, force=True)
deserialized = json.load(open(self.testfile, 'rb'))
self.maxDiff = None
+ #import ipdb;ipdb.set_trace()
self.assertEqual(deserialized, EXPECTED_DEFAULT_CONFIG)
def test_provider_dump_to_slug(self):
@@ -80,13 +87,15 @@ class TestLeapProviderDefinition(BaseLeapTest):
with open(self.testfile, 'w') as wf:
wf.write(json.dumps(EXPECTED_DEFAULT_CONFIG))
self.definition.load(fromfile=self.testfile)
- self.assertDictEqual(self.config,
- EXPECTED_DEFAULT_CONFIG)
+ #self.assertDictEqual(self.config,
+ #EXPECTED_DEFAULT_CONFIG)
+ self.assertItemsEqual(self.config, EXPECTED_DEFAULT_CONFIG)
def test_provider_validation(self):
self.definition.validate(self.config)
_config = copy.deepcopy(self.config)
- _config['serial'] = 'aaa'
+ # bad type, raise validation error
+ _config['domain'] = 111
with self.assertRaises(jsonschema.ValidationError):
self.definition.validate(_config)
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-03 11:34:15 +0000