Merge remote-tracking branch 'kali/feature/sanitize-config' into develop

4 files changed, 240 insertions, 13 deletions

diff --git a/src/leap/services/eip/eipbootstrapper.py b/src/leap/services/eip/eipbootstrapper.py
index af13ab8c..a881f235 100644
--- a/src/leap/services/eip/eipbootstrapper.py
+++ b/src/leap/services/eip/eipbootstrapper.py
@@ -102,11 +102,14 @@ class EIPBootstrapper(QtCore.QObject):
             if self._download_if_needed and mtime:
                 headers['if-modified-since'] = mtime
 
-            res = self._session.get("%s/%s/%s/%s" %
-                                    (self._provider_config.get_api_uri(),
-                                     self._provider_config.get_api_version(),
-                                     "config",
-                                     "eip-service.json"),
+            # there is some confusion with this uri,
+            # it's in 1/config/eip, config/eip and config/1/eip...
+            config_uri = "%s/%s/config/eip-service.json" % (
+                self._provider_config.get_api_uri(),
+                self._provider_config.get_api_version())
+            logger.debug('Downloading eip config from: %s' % config_uri)
+
+            res = self._session.get(config_uri,
                                     verify=self._provider_config
                                     .get_ca_cert_path(),
                                     headers=headers)
@@ -176,15 +179,15 @@ class EIPBootstrapper(QtCore.QObject):
             cookies = None
             if session_id:
                 cookies = {"_session_id": session_id}
-            res = self._session.get("%s/%s/%s/" %
-                                    (self._provider_config.get_api_uri(),
-                                     self._provider_config.get_api_version(),
-                                     "cert"),
+            cert_uri = "%s/%s/cert" % (
+                self._provider_config.get_api_uri(),
+                self._provider_config.get_api_version())
+            logger.debug('getting cert from uri: %s' % cert_uri)
+            res = self._session.get(cert_uri,
                                     verify=self._provider_config
                                     .get_ca_cert_path(),
                                     cookies=cookies)
             res.raise_for_status()
-
             client_cert = res.content
 
             # TODO: check certificate validity
diff --git a/src/leap/services/eip/eipconfig.py b/src/leap/services/eip/eipconfig.py
index 4e74687a..0a7d2b23 100644
--- a/src/leap/services/eip/eipconfig.py
+++ b/src/leap/services/eip/eipconfig.py
@@ -18,8 +18,11 @@
 """
 Provider configuration
 """
-import os
 import logging
+import os
+import re
+
+import ipaddr
 
 from leap.common.check import leap_assert, leap_assert_type
 from leap.common.config.baseconfig import BaseConfig
@@ -33,6 +36,8 @@ class EIPConfig(BaseConfig):
     """
     Provider configuration abstraction class
     """
+    OPENVPN_ALLOWED_KEYS = ("auth", "cipher", "tls-cipher")
+    OPENVPN_CIPHERS_REGEX = re.compile("[A-Z0-9\-]+")
 
     def __init__(self):
         BaseConfig.__init__(self)
@@ -52,7 +57,24 @@ class EIPConfig(BaseConfig):
         return self._safe_get_value("gateways")
 
     def get_openvpn_configuration(self):
-        return self._safe_get_value("openvpn_configuration")
+        """
+        Returns a dictionary containing the openvpn configuration
+        parameters.
+
+        These are sanitized with alphanumeric whitelist.
+
+        @returns: openvpn configuration dict
+        @rtype: C{dict}
+        """
+        ovpncfg = self._safe_get_value("openvpn_configuration")
+        config = {}
+        for key, value in ovpncfg.items():
+            if key in self.OPENVPN_ALLOWED_KEYS and value is not None:
+                sanitized_val = self.OPENVPN_CIPHERS_REGEX.findall(value)
+                if len(sanitized_val) != 0:
+                    _val = sanitized_val[0]
+                    config[str(key)] = str(_val)
+        return config
 
     def get_serial(self):
         return self._safe_get_value("serial")
@@ -61,13 +83,23 @@ class EIPConfig(BaseConfig):
         return self._safe_get_value("version")
 
     def get_gateway_ip(self, index=0):
+        """
+        Returns the ip of the gateway
+        """
         gateways = self.get_gateways()
         leap_assert(len(gateways) > 0, "We don't have any gateway!")
         if index > len(gateways):
             index = 0
             logger.warning("Provided an unknown gateway index %s, " +
                            "defaulting to 0")
-        return gateways[0]["ip_address"]
+        ip_addr_str = gateways[0]["ip_address"]
+
+        try:
+            ipaddr.IPAddress(ip_addr_str)
+            return ip_addr_str
+        except ValueError:
+            logger.error("Invalid ip address in config: %s" % (ip_addr_str,))
+            return None
 
     def get_client_cert_path(self,
                              providerconfig=None,
diff --git a/src/leap/services/eip/tests/__init__.py b/src/leap/services/eip/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/src/leap/services/eip/tests/__init__.py
diff --git a/src/leap/services/eip/tests/test_eipconfig.py b/src/leap/services/eip/tests/test_eipconfig.py
new file mode 100644
index 00000000..ce04c2fc
--- /dev/null
+++ b/src/leap/services/eip/tests/test_eipconfig.py
@@ -0,0 +1,192 @@
+# -*- coding: utf-8 -*-
+# test_eipconfig.py
+# Copyright (C) 2013 LEAP
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+"""
+tests for eipconfig
+"""
+import copy
+import json
+import os
+import unittest
+
+from leap.common.testing.basetest import BaseLeapTest
+from leap.services.eip.eipconfig import EIPConfig
+
+
+sample_config = {
+    "gateways": [
+    {
+        "capabilities": {
+            "adblock": False,
+            "filter_dns": True,
+            "limited": True,
+            "ports": [
+            "1194",
+            "443",
+            "53",
+            "80"
+            ],
+        "protocols": [
+            "tcp",
+            "udp"],
+        "transport": [
+            "openvpn"],
+        "user_ips": False},
+    "host": "host.dev.example.org",
+    "ip_address": "11.22.33.44",
+    "location": "cyberspace"
+    }],
+    "locations": {
+        "ankara": {
+        "country_code": "XX",
+        "hemisphere": "S",
+        "name": "Antarctica",
+        "timezone": "+2"
+        }
+    },
+    "openvpn_configuration": {
+        "auth": "SHA1",
+        "cipher": "AES-128-CBC",
+        "tls-cipher": "DHE-RSA-AES128-SHA"
+    },
+    "serial": 1,
+    "version": 1
+}
+
+
+class EIPConfigTest(BaseLeapTest):
+
+    __name__ = "eip_config_tests"
+    #provider = "testprovider.example.org"
+
+    maxDiff = None
+
+    def setUp(self):
+        pass
+
+    def tearDown(self):
+        pass
+
+    #
+    # helpers
+    #
+
+    def write_config(self, data):
+        self.configfile = os.path.join(
+            self.tempdir, "eipconfig.json")
+        conf = open(self.configfile, "w")
+        conf.write(json.dumps(data))
+        conf.close()
+
+    def test_load_valid_config(self):
+        """
+        load a sample config
+        """
+        self.write_config(sample_config)
+        config = EIPConfig()
+        self.assertRaises(
+            AssertionError,
+            config.get_clusters)
+        self.assertTrue(config.load(self.configfile))
+        self.assertEqual(
+            config.get_openvpn_configuration(),
+            sample_config["openvpn_configuration"])
+        self.assertEqual(
+            config.get_gateway_ip(),
+            "11.22.33.44")
+        self.assertEqual(config.get_version(), 1)
+        self.assertEqual(config.get_serial(), 1)
+        self.assertEqual(config.get_gateways(),
+                         sample_config["gateways"])
+        self.assertEqual(
+            config.get_clusters(), None)
+
+    def test_sanitize_config(self):
+        """
+        check the sanitization of options
+        """
+        # extra parameters
+        data = copy.deepcopy(sample_config)
+        data['openvpn_configuration']["extra_param"] = "FOO"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_openvpn_configuration(),
+            sample_config["openvpn_configuration"])
+
+        # non allowed chars
+        data = copy.deepcopy(sample_config)
+        data['openvpn_configuration']["auth"] = "SHA1;"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_openvpn_configuration(),
+            sample_config["openvpn_configuration"])
+
+        # non allowed chars
+        data = copy.deepcopy(sample_config)
+        data['openvpn_configuration']["auth"] = "SHA1>`&|"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_openvpn_configuration(),
+            sample_config["openvpn_configuration"])
+
+        # lowercase
+        data = copy.deepcopy(sample_config)
+        data['openvpn_configuration']["auth"] = "shaSHA1"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_openvpn_configuration(),
+            sample_config["openvpn_configuration"])
+
+        # all characters invalid -> null value
+        data = copy.deepcopy(sample_config)
+        data['openvpn_configuration']["auth"] = "sha&*!@#;"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_openvpn_configuration(),
+            {'cipher': 'AES-128-CBC',
+             'tls-cipher': 'DHE-RSA-AES128-SHA'})
+
+        # bad_ip
+        data = copy.deepcopy(sample_config)
+        data['gateways'][0]["ip_address"] = "11.22.33.44;"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_gateway_ip(),
+            None)
+
+        data = copy.deepcopy(sample_config)
+        data['gateways'][0]["ip_address"] = "11.22.33.44`"
+        self.write_config(data)
+        config = EIPConfig()
+        config.load(self.configfile)
+        self.assertEqual(
+            config.get_gateway_ip(),
+            None)
+
+if __name__ == "__main__":
+    unittest.main()