diff options
Diffstat (limited to 'src/leap/eip')
| -rw-r--r-- | src/leap/eip/checks.py | 90 | ||||
| -rw-r--r-- | src/leap/eip/exceptions.py | 16 | ||||
| -rw-r--r-- | src/leap/eip/tests/test_checks.py | 61 |
3 files changed, 157 insertions, 10 deletions
diff --git a/src/leap/eip/checks.py b/src/leap/eip/checks.py index 4a2a9599..4dd4a95c 100644 --- a/src/leap/eip/checks.py +++ b/src/leap/eip/checks.py @@ -1,7 +1,10 @@ import logging import ssl +import platform import os +import netifaces +import ping import requests from leap.base import constants as baseconstants @@ -17,8 +20,6 @@ logger = logging.getLogger(name=__name__) """ EIPConfigChecker ---------- -this is the first of 3 consecutive checks that we're implementing. - It is used from the eip conductor (a instance of EIPConnection that is managed from the QtApp), running `run_all` method before trying to call `connect` or any other of the state-changing methods. @@ -29,13 +30,88 @@ into base.tests to be invoked by the base leap init routines. However, I'm testing them alltogether for the sake of having the whole unit reachable and testable as a whole. -Other related checkers - not implemented yet -: -* LeapNetworkChecker +LeapNetworkChecker +------------------ +Network checks. To be moved to base. +docs TBD + +ProviderCertChecker +------------------- +Checks on certificates. +docs TBD """ class LeapNetworkChecker(object): - pass + """ + all network related checks + """ + # XXX to be moved to leap.base.checks + # TODO eventually, use a more portable solution + # like psutil + + def run_all(self, checker=None): + if not checker: + checker = self + self.error = None # ? + + # for MVS + checker.test_internet_connection() + checker.is_internet_up() + checker.ping_gateway() + + def test_internet_connection(self): + # XXX we're not passing the error anywhere. + # XXX we probably should raise an exception here? + # unless we use this as smoke test + try: + requests.get('http://216.172.161.165') + except (requests.HTTPError, requests.RequestException) as e: + self.error = e.message + except requests.ConenctionError as e: + if e.message == "[Errno 113] No route to host": + if not self.is_internet_up(): + self.error = "No valid internet connection found." + else: + self.error = "Provider server appears to be down." + + def is_internet_up(self): + iface, gateway = self.get_default_interface_gateway() + self.ping_gateway(self) + + def get_default_interface_gateway(self): + """only impletemented for linux so far.""" + if not platform.system() == "Linux": + raise NotImplementedError + + f = open("/proc/net/route") + route_table = f.readlines() + f.close() + #toss out header + route_table.pop(0) + + default_iface = None + gateway = None + while route_table: + line = route_table.pop(0) + iface, destination, gateway = line.split('\t')[0:3] + if destination == '00000000': + default_iface = iface + break + + if not default_iface: + raise eipexceptions.NoDefaultInterfaceFoundError + + if default_iface not in netifaces.interfaces(): + raise eipexceptions.InterfaceNotFoundError + + return default_iface, gateway + + def ping_gateway(self, gateway): + #TODO: Discuss how much packet loss (%) is acceptable. + packet_loss = ping.quiet_ping(gateway)[0] + if packet_loss > baseconstants.MAX_ICMP_PACKET_LOSS: + raise eipexceptions.NoConnectionToGateway class ProviderCertChecker(object): @@ -150,6 +226,7 @@ class ProviderCertChecker(object): # XXX TODO # waiting on #507. If we're not using PyOpenSSL or anything alike # we will have to roll our own x509 parsing to extract time info. + # XXX use gnutls def is_valid_pemfile(self, cert_s=None): """ @@ -319,9 +396,6 @@ class EIPConfigChecker(object): # We should WRITE eip config if missing or # incomplete at this point - def ping_gateway(self): - raise NotImplementedError - # # private helpers # diff --git a/src/leap/eip/exceptions.py b/src/leap/eip/exceptions.py index 3c8f6afb..467be7fe 100644 --- a/src/leap/eip/exceptions.py +++ b/src/leap/eip/exceptions.py @@ -108,6 +108,22 @@ class EIPInitBadProviderError(EIPClientError): class EIPConfigurationError(EIPClientError): pass + +class NoDefaultInterfaceFoundError(EIPClientError): + message = "no default interface found" + usermessage = "Looks like your computer is not connected to the internet" + + +class InterfaceNotFoundError(EIPClientError): + # XXX should take iface arg on init maybe? + message = "interface not found" + + +class NoConnectionToGateway(EIPClientError): + message = "no connection to gateway" + usermessage = "Looks like there are problems with your internet connection" + + # # Errors that probably we don't need anymore # chase down for them and check. diff --git a/src/leap/eip/tests/test_checks.py b/src/leap/eip/tests/test_checks.py index 0a87f573..bc7db79c 100644 --- a/src/leap/eip/tests/test_checks.py +++ b/src/leap/eip/tests/test_checks.py @@ -8,8 +8,10 @@ except ImportError: import os import urlparse -from mock import patch, Mock +from StringIO import StringIO +from mock import (patch, Mock) +import ping import requests from leap.base import config as baseconfig @@ -23,6 +25,8 @@ from leap.testing.basetest import BaseLeapTest from leap.testing.https_server import BaseHTTPSServerTestCase from leap.testing.https_server import where as where_cert +_uid = os.getuid() + class NoLogRequestHandler: def log_message(self, *args): @@ -33,6 +37,60 @@ class NoLogRequestHandler: return '' +class LeapNetworkCheckTest(BaseLeapTest): + # XXX to be moved to base.checks + + __name__ = "leap_network_check_tests" + + def setUp(self): + pass + + def tearDown(self): + pass + + def test_checker_should_implement_check_methods(self): + checker = eipchecks.LeapNetworkChecker() + + self.assertTrue(hasattr(checker, "test_internet_connection"), + "missing meth") + self.assertTrue(hasattr(checker, "is_internet_up"), + "missing meth") + self.assertTrue(hasattr(checker, "ping_gateway"), + "missing meth") + + def test_checker_should_actually_call_all_tests(self): + checker = eipchecks.LeapNetworkChecker() + + mc = Mock() + checker.run_all(checker=mc) + self.assertTrue(mc.test_internet_connection.called, "not called") + self.assertTrue(mc.ping_gateway.called, "not called") + self.assertTrue(mc.is_internet_up.called, + "not called") + + def test_get_default_interface_no_interface(self): + checker = eipchecks.LeapNetworkChecker() + with patch('leap.eip.checks.open', create=True) as mock_open: + with self.assertRaises(eipexceptions.NoDefaultInterfaceFoundError): + mock_open.return_value = StringIO( + "Iface\tDestination Gateway\t" + "Flags\tRefCntd\tUse\tMetric\t" + "Mask\tMTU\tWindow\tIRTT") + checker.get_default_interface_gateway() + + def test_ping_gateway_fail(self): + checker = eipchecks.LeapNetworkChecker() + with patch.object(ping, "quiet_ping") as mocked_ping: + with self.assertRaises(eipexceptions.NoConnectionToGateway): + mocked_ping.return_value = [11, "", ""] + checker.ping_gateway("4.2.2.2") + + @unittest.skipUnless(_uid == 0, "root only") + def test_ping_gateway(self): + checker = eipchecks.LeapNetworkChecker() + checker.ping_gateway("4.2.2.2") + + class EIPCheckTest(BaseLeapTest): __name__ = "eip_check_tests" @@ -57,7 +115,6 @@ class EIPCheckTest(BaseLeapTest): "missing meth") self.assertTrue(hasattr(checker, "check_complete_eip_config"), "missing meth") - self.assertTrue(hasattr(checker, "ping_gateway"), "missing meth") def test_checker_should_actually_call_all_tests(self): checker = eipchecks.EIPConfigChecker() |