workaround for srp server timing out on consecutive runs

3 files changed, 86 insertions, 159 deletions

diff --git a/src/leap/crypto/tests/fake_provider.py b/src/leap/crypto/tests/fake_provider.py
index 4b05bbff..d3e05812 100755
--- a/src/leap/crypto/tests/fake_provider.py
+++ b/src/leap/crypto/tests/fake_provider.py
@@ -329,5 +329,3 @@ if __name__ == "__main__":
     reactor.listenTCP(8000, factory)
     reactor.listenSSL(8443, factory, OpenSSLServerContextFactory())
     reactor.run()
-
-
diff --git a/src/leap/crypto/tests/test_srpauth.py b/src/leap/crypto/tests/test_srpauth.py
deleted file mode 100644
index ce9403c7..00000000
--- a/src/leap/crypto/tests/test_srpauth.py
+++ /dev/null
@@ -1,136 +0,0 @@
-# -*- coding: utf-8 -*-
-# test_srpauth.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 leap/crypto/srpauth.py
-"""
-try:
-    import unittest
-except ImportError:
-    import unittest
-import os
-import sys
-
-from mock import MagicMock
-from nose.twistedtools import reactor, threaded_reactor, stop_reactor
-from twisted.python import log
-
-from leap.common.testing.https_server import where
-from leap.config.providerconfig import ProviderConfig
-from leap.crypto import srpauth
-from leap.crypto import srpregister
-from leap.crypto.tests import fake_provider
-
-log.startLogging(sys.stdout)
-
-
-def _get_capath():
-    return where("cacert.pem")
-
-_here = os.path.split(__file__)[0]
-
-
-class ImproperlyConfiguredError(Exception):
-    """
-    Raised if the test provider is missing configuration
-    """
-
-
-class SRPRegisterTestCase(unittest.TestCase):
-    """
-    Tests for the SRP Authentication class
-    """
-    __name__ = "SRPAuth tests"
-
-    @classmethod
-    def setUpClass(cls):
-        """
-        Sets up this TestCase with a simple and faked provider instance:
-
-        * runs a threaded reactor
-        * loads a mocked ProviderConfig that points to the certs in the
-          leap.common.testing module.
-        """
-        factory = fake_provider.get_provider_factory()
-        reactor.listenTCP(8000, factory)
-        reactor.listenSSL(
-            8443, factory,
-            fake_provider.OpenSSLServerContextFactory())
-        threaded_reactor()
-
-        provider = ProviderConfig()
-        provider.get_ca_cert_path = MagicMock()
-        provider.get_ca_cert_path.return_value = _get_capath()
-        loaded = provider.load(path=os.path.join(
-            _here, "test_provider.json"))
-        if not loaded:
-            raise ImproperlyConfiguredError(
-                "Could not load test provider config")
-        cls.provider = provider
-        cls.register = srpregister.SRPRegister(provider_config=provider)
-        cls.auth = srpauth.SRPAuth(provider)
-        cls._auth_instance = cls.auth.__dict__['_SRPAuth__instance']
-        cls.authenticate = cls._auth_instance.authenticate
-        cls.logout = cls._auth_instance.logout
-
-    @classmethod
-    def tearDownClass(cls):
-        """
-        Stops reactor when tearing down the class
-        """
-        stop_reactor()
-
-    def test_auth(self):
-        """
-        Checks whether a pair of valid credentials is able to be authenticated.
-        """
-        TEST_USER = "register_test_auth"
-        TEST_PASS = "pass"
-
-        # pristine registration, should go well
-        ok = self.register.register_user(TEST_USER, TEST_PASS)
-        self.assertTrue(ok)
-
-        self.authenticate(TEST_USER, TEST_PASS)
-        with self.assertRaises(AssertionError):
-            # AssertionError: already logged in
-            # We probably could take this as its own exception
-            self.authenticate(TEST_USER, TEST_PASS)
-
-        self.logout()
-
-        # cannot log out two times in a row (there's no session)
-        with self.assertRaises(AssertionError):
-            self.logout()
-
-    def test_auth_with_bad_credentials(self):
-        """
-        Checks that auth does not succeed with bad credentials.
-        """
-        TEST_USER = "register_test_auth"
-        TEST_PASS = "pass"
-
-        # non-existent credentials, should fail
-        with self.assertRaises(srpauth.SRPAuthenticationError):
-            self.authenticate("baduser_1", "passwrong")
-
-        # good user, bad password, should fail
-        with self.assertRaises(srpauth.SRPAuthenticationError):
-            self.authenticate(TEST_USER, "passwrong")
-
-        # bad user, good password, should fail too :)
-        with self.assertRaises(srpauth.SRPAuthenticationError):
-            self.authenticate("myunclejoe", TEST_PASS)
diff --git a/src/leap/crypto/tests/test_srpregister.py b/src/leap/crypto/tests/test_srpregister.py
index b065958d..a59f71cb 100644
--- a/src/leap/crypto/tests/test_srpregister.py
+++ b/src/leap/crypto/tests/test_srpregister.py
@@ -15,7 +15,9 @@
 # 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 leap/crypto/srpregister.py
+Tests for:
+    * leap/crypto/srpregister.py
+    * leap/crypto/srpauth.py
 """
 try:
     import unittest
@@ -30,7 +32,7 @@ from twisted.python import log
 
 from leap.common.testing.https_server import where
 from leap.config.providerconfig import ProviderConfig
-from leap.crypto import srpregister
+from leap.crypto import srpregister, srpauth
 from leap.crypto.tests import fake_provider
 
 log.startLogging(sys.stdout)
@@ -48,11 +50,11 @@ class ImproperlyConfiguredError(Exception):
     """
 
 
-class SRPRegisterTestCase(unittest.TestCase):
+class SRPTestCase(unittest.TestCase):
     """
-    Tests for the SRP Register class
+    Tests for the SRP Register and Auth classes
     """
-    __name__ = "SRPRegister tests"
+    __name__ = "SRPRegister and SRPAuth tests"
 
     @classmethod
     def setUpClass(cls):
@@ -60,30 +62,39 @@ class SRPRegisterTestCase(unittest.TestCase):
         Sets up this TestCase with a simple and faked provider instance:
 
         * runs a threaded reactor
+        * loads a mocked ProviderConfig that points to the certs in the
+          leap.common.testing module.
         """
         factory = fake_provider.get_provider_factory()
-        reactor.listenTCP(8000, factory)
-        reactor.listenSSL(
-            8443, factory,
+        http = reactor.listenTCP(8001, factory)
+        https = reactor.listenSSL(
+            0, factory,
             fake_provider.OpenSSLServerContextFactory())
-        threaded_reactor()
-
-    def setUp(self):
-        """
-        Sets up common parameters for each test:
+        get_port = lambda p: p.getHost().port
+        cls.http_port = get_port(http)
+        cls.https_port = get_port(https)
 
-        * loads a mocked ProviderConfig that points to the certs in the
-          leap.common.testing module.
-        """
         provider = ProviderConfig()
         provider.get_ca_cert_path = MagicMock()
         provider.get_ca_cert_path.return_value = _get_capath()
+
+        provider.get_api_uri = MagicMock()
+        provider.get_api_uri.return_value = cls._get_https_uri()
+
         loaded = provider.load(path=os.path.join(
             _here, "test_provider.json"))
         if not loaded:
             raise ImproperlyConfiguredError(
                 "Could not load test provider config")
-        self.register = srpregister.SRPRegister(provider_config=provider)
+        cls.register = srpregister.SRPRegister(provider_config=provider)
+
+        cls.auth = srpauth.SRPAuth(provider)
+        cls._auth_instance = cls.auth.__dict__['_SRPAuth__instance']
+        cls.authenticate = cls._auth_instance.authenticate
+        cls.logout = cls._auth_instance.logout
+
+        # run!
+        threaded_reactor()
 
     @classmethod
     def tearDownClass(cls):
@@ -92,6 +103,17 @@ class SRPRegisterTestCase(unittest.TestCase):
         """
         stop_reactor()
 
+    # helper methods
+
+    @classmethod
+    def _get_https_uri(cls):
+        """
+        Returns a https uri with the right https port initialized
+        """
+        return "https://localhost:%s" % (cls.https_port,)
+
+    # Register tests
+
     def test_register_user(self):
         """
         Checks if the registration of an unused name works as expected when
@@ -109,15 +131,13 @@ class SRPRegisterTestCase(unittest.TestCase):
         # FIXME currently we are catching this in an upper layer,
         # we could bring the error validation to the SRPRegister class
         ok = self.register.register_user("foouser_second", "barpass")
-        # XXX
-        #self.assertFalse(ok)
 
     def test_correct_http_uri(self):
         """
         Checks that registration autocorrect http uris to https ones.
         """
-        HTTP_URI = "http://localhost:8443"
-        HTTPS_URI = "https://localhost:8443/1/users"
+        HTTP_URI = "http://localhost:%s" % (self.https_port, )
+        HTTPS_URI = "https://localhost:%s/1/users" % (self.https_port, )
         provider = ProviderConfig()
         provider.get_ca_cert_path = MagicMock()
         provider.get_ca_cert_path.return_value = _get_capath()
@@ -130,6 +150,7 @@ class SRPRegisterTestCase(unittest.TestCase):
         if not loaded:
             raise ImproperlyConfiguredError(
                 "Could not load test provider config")
+
         self.register = srpregister.SRPRegister(provider_config=provider)
 
         # ... and we check that we're correctly taking the HTTPS protocol
@@ -140,3 +161,47 @@ class SRPRegisterTestCase(unittest.TestCase):
         self.assertTrue(ok)
 
         # XXX need to assert that _get_registration_uri was called too
+
+    # Auth tests
+
+    def test_auth(self):
+        """
+        Checks whether a pair of valid credentials is able to be authenticated.
+        """
+        TEST_USER = "register_test_auth"
+        TEST_PASS = "pass"
+
+        # pristine registration, should go well
+        ok = self.register.register_user(TEST_USER, TEST_PASS)
+        self.assertTrue(ok)
+
+        self.authenticate(TEST_USER, TEST_PASS)
+        with self.assertRaises(AssertionError):
+            # AssertionError: already logged in
+            # We probably could take this as its own exception
+            self.authenticate(TEST_USER, TEST_PASS)
+
+        self.logout()
+
+        # cannot log out two times in a row (there's no session)
+        with self.assertRaises(AssertionError):
+            self.logout()
+
+    def test_auth_with_bad_credentials(self):
+        """
+        Checks that auth does not succeed with bad credentials.
+        """
+        TEST_USER = "register_test_auth"
+        TEST_PASS = "pass"
+
+        # non-existent credentials, should fail
+        with self.assertRaises(srpauth.SRPAuthenticationError):
+            self.authenticate("baduser_1", "passwrong")
+
+        # good user, bad password, should fail
+        with self.assertRaises(srpauth.SRPAuthenticationError):
+            self.authenticate(TEST_USER, "passwrong")
+
+        # bad user, good password, should fail too :)
+        with self.assertRaises(srpauth.SRPAuthenticationError):
+            self.authenticate("myunclejoe", TEST_PASS)