From 914a07aaf8ef52b2eaf88f1bf01fb6f72adcac5a Mon Sep 17 00:00:00 2001 From: kali Date: Sat, 15 Dec 2012 02:25:12 +0900 Subject: use gnutls to parse pemfiles --- src/leap/base/auth.py | 8 +++++-- src/leap/base/tests/test_providers.py | 6 +++-- src/leap/crypto/certs.py | 42 ++++++++++++++++++++++++++++++++--- src/leap/crypto/tests/__init__.py | 0 src/leap/crypto/tests/test_certs.py | 11 +++++++++ 5 files changed, 60 insertions(+), 7 deletions(-) create mode 100644 src/leap/crypto/tests/__init__.py create mode 100644 src/leap/crypto/tests/test_certs.py (limited to 'src') diff --git a/src/leap/base/auth.py b/src/leap/base/auth.py index 56b7cf96..c6bd3518 100644 --- a/src/leap/base/auth.py +++ b/src/leap/base/auth.py @@ -255,6 +255,7 @@ class SRPAuth(requests.auth.AuthBase): try: assert self.srp_usr.authenticated() logger.debug('user is authenticated!') + print 'user is authenticated!' except (AssertionError): raise SRPAuthenticationError( "Auth verification failed.") @@ -355,8 +356,11 @@ if __name__ == "__main__": req.raise_for_status return req - req = test_srp_protected_get('https://localhost:8443/1/cert') - print 'cert :', req.content[:200] + "..." + #req = test_srp_protected_get('https://localhost:8443/1/cert') + req = test_srp_protected_get('%s/1/cert' % SERVER) + import ipdb;ipdb.set_trace() + #print 'cert :', req.content[:200] + "..." + print 'cert :', req.content sys.exit(0) if action == "add": diff --git a/src/leap/base/tests/test_providers.py b/src/leap/base/tests/test_providers.py index d9604fab..8801a3eb 100644 --- a/src/leap/base/tests/test_providers.py +++ b/src/leap/base/tests/test_providers.py @@ -16,10 +16,12 @@ 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"default_language": u"en", + #u"display_name": {u'en': u"Test Provider"}, u"domain": u"testprovider.example.org", + u'name': {u'en': u'Test Provider'}, u"enrollment_policy": u"open", - u"serial": 1, + #u"serial": 1, u"services": [ u"eip" ], diff --git a/src/leap/crypto/certs.py b/src/leap/crypto/certs.py index 8908865d..45d7326d 100644 --- a/src/leap/crypto/certs.py +++ b/src/leap/crypto/certs.py @@ -1,10 +1,14 @@ import ctypes +from StringIO import StringIO +import re import socket import gnutls.connection import gnutls.crypto import gnutls.library +from leap.util.misc import null_check + def get_https_cert_from_domain(domain): """ @@ -20,12 +24,44 @@ def get_https_cert_from_domain(domain): return cert -def get_cert_from_file(filepath): - with open(filepath) as f: - cert = gnutls.crypto.X509Certificate(f.read()) +def get_cert_from_file(_file): + getcert = lambda f: gnutls.crypto.X509Certificate(f.read()) + if isinstance(_file, str): + with open(_file) as f: + cert = getcert(f) + else: + cert = getcert(_file) return cert +def get_pkey_from_file(_file): + getkey = lambda f: gnutls.crypto.X509PrivateKey(f.read()) + if isinstance(_file, str): + with open(_file) as f: + key = getkey(f) + else: + key = getkey(_file) + return key + + +def can_load_cert_and_pkey(string): + try: + f = StringIO(string) + cert = get_cert_from_file(f) + + f = StringIO(string) + key = get_pkey_from_file(f) + + null_check(cert, 'certificate') + null_check(key, 'private key') + except: + # XXX catch GNUTLSError + raise + return False + else: + return True + + def get_cert_fingerprint(domain=None, filepath=None, hash_type="SHA256", sep=":"): """ diff --git a/src/leap/crypto/tests/__init__.py b/src/leap/crypto/tests/__init__.py new file mode 100644 index 00000000..e69de29b diff --git a/src/leap/crypto/tests/test_certs.py b/src/leap/crypto/tests/test_certs.py new file mode 100644 index 00000000..4d167c51 --- /dev/null +++ b/src/leap/crypto/tests/test_certs.py @@ -0,0 +1,11 @@ +import unittest + + +class CertTestCase(unittest.TestCase): + + def test_load_client_and_pkey(self): + self.fail('not implemented') + + +if __name__ == "__main__": + unittest.main() -- cgit v1.2.3