From b612f422bf156a3b3927038472ad885b1afa556e Mon Sep 17 00:00:00 2001 From: kali Date: Mon, 3 Sep 2012 02:51:41 +0900 Subject: providercertchecks:check_https_is_working implementing a https server with its own base testcase for convenience. https is delicate, and I think it's better checking against a real implementation than mocking everything here. --- src/leap/eip/checks.py | 62 ++++++++++++++++++- src/leap/eip/tests/test_checks.py | 124 ++++++++++++++++++++++++++++++++++++++ 2 files changed, 184 insertions(+), 2 deletions(-) (limited to 'src/leap/eip') diff --git a/src/leap/eip/checks.py b/src/leap/eip/checks.py index c6a7ca72..4112ef57 100644 --- a/src/leap/eip/checks.py +++ b/src/leap/eip/checks.py @@ -40,12 +40,70 @@ class LeapNetworkChecker(object): class ProviderCertChecker(object): - pass + """ + Several checks needed for getting + client certs and checking tls connection + with provider. + """ + def __init__(self, fetcher=requests): + self.fetcher = fetcher + + def run_all(self, checker=None, skip_download=False): + if not checker: + checker = self + + # For MVS+ + # checker.download_ca_cert() + # checker.download_ca_signature() + # checker.get_ca_signatures() + # checker.is_there_trust_path() + + # For MVS + checker.is_there_provider_ca() + checker.is_https_working() + checker.download_new_client_cert() + + def download_ca_cert(self): + # MVS+ + raise NotImplementedError + + def download_ca_signature(self): + # MVS+ + raise NotImplementedError + + def get_ca_signatures(self): + # MVS+ + raise NotImplementedError + + def is_there_trust_path(self): + # MVS+ + raise NotImplementedError + + def is_there_provider_ca(self): + # XXX fake it till you make it! :P + return True + + # enable this when we have + # a custom "branded" bundle + # certs package. + try: + from leap.custom import certs + certs.ca.pemfile + except ImportError: + raise + + def is_https_working(self, uri=None, cacert=None, verify=True): + assert uri.startswith('https') + self.fetcher.get(uri, verify=verify) + return True + + def download_new_client_cert(self): + return True class EIPConfigChecker(object): """ - Several tests needed + Several checks needed to ensure a EIPConnection can be sucessfully established. use run_all to run all checks. diff --git a/src/leap/eip/tests/test_checks.py b/src/leap/eip/tests/test_checks.py index 1e629203..781fdad5 100644 --- a/src/leap/eip/tests/test_checks.py +++ b/src/leap/eip/tests/test_checks.py @@ -1,3 +1,4 @@ +from BaseHTTPServer import BaseHTTPRequestHandler import copy import json try: @@ -18,6 +19,16 @@ from leap.eip import specs as eipspecs from leap.eip import exceptions as eipexceptions from leap.eip.tests import data as testdata from leap.testing.basetest import BaseLeapTest +from leap.testing.https_server import BaseHTTPSServerTestCase + + +class NoLogRequestHandler: + def log_message(self, *args): + # don't write log msg to stderr + pass + + def read(self, n=None): + return '' class EIPCheckTest(BaseLeapTest): @@ -157,5 +168,118 @@ class EIPCheckTest(BaseLeapTest): sampleconfig = copy.copy(testdata.EIP_SAMPLE_JSON) checker.check_complete_eip_config(config=sampleconfig) + +class ProviderCertCheckerTest(BaseLeapTest): + + __name__ = "provider_cert_checker_tests" + + def setUp(self): + pass + + def tearDown(self): + pass + + # test methods are there, and can be called from run_all + + def test_checker_should_implement_check_methods(self): + checker = eipchecks.ProviderCertChecker() + + # For MVS+ + self.assertTrue(hasattr(checker, "download_ca_cert"), + "missing meth") + self.assertTrue(hasattr(checker, "download_ca_signature"), + "missing meth") + self.assertTrue(hasattr(checker, "get_ca_signatures"), "missing meth") + self.assertTrue(hasattr(checker, "is_there_trust_path"), + "missing meth") + + # For MVS + self.assertTrue(hasattr(checker, "is_there_provider_ca"), + "missing meth") + self.assertTrue(hasattr(checker, "is_https_working"), "missing meth") + self.assertTrue(hasattr(checker, "download_new_client_cert"), + "missing meth") + + def test_checker_should_actually_call_all_tests(self): + checker = eipchecks.ProviderCertChecker() + + mc = Mock() + checker.run_all(checker=mc) + # XXX MVS+ + #self.assertTrue(mc.download_ca_cert.called, "not called") + #self.assertTrue(mc.download_ca_signature.called, "not called") + #self.assertTrue(mc.get_ca_signatures.called, "not called") + #self.assertTrue(mc.is_there_trust_path.called, "not called") + + # For MVS + self.assertTrue(mc.is_there_provider_ca.called, "not called") + self.assertTrue(mc.is_https_working.called, + "not called") + self.assertTrue(mc.download_new_client_cert.called, + "not called") + + # test individual check methods + + def test_is_there_provider_ca(self): + checker = eipchecks.ProviderCertChecker() + self.assertTrue( + checker.is_there_provider_ca()) + + +class ProviderCertCheckerHTTPSTests(BaseHTTPSServerTestCase): + class request_handler(NoLogRequestHandler, BaseHTTPRequestHandler): + def do_GET(self): + #XXX use path to deliver foo stuff + #path = urlparse.urlparse(self.path) + #print path + message = '\n'.join([ + 'OK', + '']) + self.send_response(200) + self.end_headers() + self.wfile.write(message) + + def test_is_https_working(self): + fetcher = requests + uri = "https://%s/" % (self.get_server()) + # bare requests call. this should just pass (if there is + # an https service there). + fetcher.get(uri, verify=False) + checker = eipchecks.ProviderCertChecker(fetcher=fetcher) + self.assertTrue(checker.is_https_working(uri=uri, verify=False)) + + # for local debugs, when in doubt + #self.assertTrue(checker.is_https_working(uri="https://github.com", + #verify=True)) + + # for the two checks below, I know they fail because no ca + # cert is passed to them, and I know that's the error that + # requests return with our implementation. However, I believe + # the right error should be SSL23_READ_BYTES: alert bad certificate + # or something similar. I guess we're receiving this because our + # server is dying prematurely when the handshake is interrupted on the + # client side. In any case I think that requests could handle + # this error more consistently and return a ConnectionError on a + # higher level. + with self.assertRaises(requests.exceptions.SSLError) as exc: + fetcher.get(uri, verify=True) + self.assertTrue( + "SSL23_GET_SERVER_HELLO:unknown protocol" in exc.message) + with self.assertRaises(requests.exceptions.SSLError) as exc: + checker.is_https_working(uri=uri, verify=True) + self.assertTrue( + "SSL23_GET_SERVER_HELLO:unknown protocol" in exc.message) + + # XXX get cacert from testing.https_server + + def test_download_new_client_cert(self): + checker = eipchecks.ProviderCertChecker() + self.assertTrue(checker.download_new_client_cert()) + + #def test_download_bad_client_cert(self): + #checker = eipchecks.ProviderCertChecker() + #self.assertTrue(checker.download_new_client_cert()) + + if __name__ == "__main__": unittest.main() -- cgit v1.2.3