summaryrefslogtreecommitdiff
path: root/src/leap/crypto
diff options
context:
space:
mode:
Diffstat (limited to 'src/leap/crypto')
-rw-r--r--src/leap/crypto/certs.py112
-rw-r--r--src/leap/crypto/certs_gnutls.py112
-rw-r--r--src/leap/crypto/leapkeyring.py2
-rw-r--r--src/leap/crypto/tests/__init__.py0
-rw-r--r--src/leap/crypto/tests/test_certs.py22
5 files changed, 248 insertions, 0 deletions
diff --git a/src/leap/crypto/certs.py b/src/leap/crypto/certs.py
new file mode 100644
index 00000000..cbb5725a
--- /dev/null
+++ b/src/leap/crypto/certs.py
@@ -0,0 +1,112 @@
+import logging
+import os
+from StringIO import StringIO
+import ssl
+import time
+
+from dateutil.parser import parse
+from OpenSSL import crypto
+
+from leap.util.misc import null_check
+
+logger = logging.getLogger(__name__)
+
+
+class BadCertError(Exception):
+ """
+ raised for malformed certs
+ """
+
+
+class NoCertError(Exception):
+ """
+ raised for cert not found in given path
+ """
+
+
+def get_https_cert_from_domain(domain, port=443):
+ """
+ @param domain: a domain name to get a certificate from.
+ """
+ cert = ssl.get_server_certificate((domain, port))
+ x509 = crypto.load_certificate(crypto.FILETYPE_PEM, cert)
+ return x509
+
+
+def get_cert_from_file(_file):
+ null_check(_file, "pem file")
+ if isinstance(_file, (str, unicode)):
+ if not os.path.isfile(_file):
+ raise NoCertError
+ with open(_file) as f:
+ cert = f.read()
+ else:
+ cert = _file.read()
+ x509 = crypto.load_certificate(crypto.FILETYPE_PEM, cert)
+ return x509
+
+
+def get_pkey_from_file(_file):
+ getkey = lambda f: crypto.load_privatekey(
+ crypto.FILETYPE_PEM, 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):
+ """
+ loads certificate and private key from
+ a buffer
+ """
+ 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 Exception as exc:
+ logger.error(type(exc), exc.message)
+ raise BadCertError
+ else:
+ return True
+
+
+def get_cert_fingerprint(domain=None, port=443, filepath=None,
+ hash_type="SHA256", sep=":"):
+ """
+ @param domain: a domain name to get a fingerprint from
+ @type domain: str
+ @param filepath: path to a file containing a PEM file
+ @type filepath: str
+ @param hash_type: the hash function to be used in the fingerprint.
+ must be one of SHA1, SHA224, SHA256, SHA384, SHA512
+ @type hash_type: str
+ @rparam: hex_fpr, a hexadecimal representation of a bytestring
+ containing the fingerprint.
+ @rtype: string
+ """
+ if domain:
+ cert = get_https_cert_from_domain(domain, port=port)
+ if filepath:
+ cert = get_cert_from_file(filepath)
+ hex_fpr = cert.digest(hash_type)
+ return hex_fpr
+
+
+def get_time_boundaries(certfile):
+ cert = get_cert_from_file(certfile)
+ null_check(cert, 'certificate')
+
+ fromts, tots = (cert.get_notBefore(), cert.get_notAfter())
+ from_, to_ = map(
+ lambda ts: time.gmtime(time.mktime(parse(ts).timetuple())),
+ (fromts, tots))
+ return from_, to_
diff --git a/src/leap/crypto/certs_gnutls.py b/src/leap/crypto/certs_gnutls.py
new file mode 100644
index 00000000..20c0e043
--- /dev/null
+++ b/src/leap/crypto/certs_gnutls.py
@@ -0,0 +1,112 @@
+'''
+We're using PyOpenSSL now
+
+import ctypes
+from StringIO import StringIO
+import socket
+
+import gnutls.connection
+import gnutls.crypto
+import gnutls.library
+
+from leap.util.misc import null_check
+
+
+class BadCertError(Exception):
+ """raised for malformed certs"""
+
+
+def get_https_cert_from_domain(domain):
+ """
+ @param domain: a domain name to get a certificate from.
+ """
+ sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+ cred = gnutls.connection.X509Credentials()
+
+ session = gnutls.connection.ClientSession(sock, cred)
+ session.connect((domain, 443))
+ session.handshake()
+ cert = session.peer_certificate
+ return cert
+
+
+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 BadCertError
+ else:
+ return True
+
+def get_cert_fingerprint(domain=None, filepath=None,
+ hash_type="SHA256", sep=":"):
+ """
+ @param domain: a domain name to get a fingerprint from
+ @type domain: str
+ @param filepath: path to a file containing a PEM file
+ @type filepath: str
+ @param hash_type: the hash function to be used in the fingerprint.
+ must be one of SHA1, SHA224, SHA256, SHA384, SHA512
+ @type hash_type: str
+ @rparam: hex_fpr, a hexadecimal representation of a bytestring
+ containing the fingerprint.
+ @rtype: string
+ """
+ if domain:
+ cert = get_https_cert_from_domain(domain)
+ if filepath:
+ cert = get_cert_from_file(filepath)
+
+ _buffer = ctypes.create_string_buffer(64)
+ buffer_length = ctypes.c_size_t(64)
+
+ SUPPORTED_DIGEST_FUN = ("SHA1", "SHA224", "SHA256", "SHA384", "SHA512")
+ if hash_type in SUPPORTED_DIGEST_FUN:
+ digestfunction = getattr(
+ gnutls.library.constants,
+ "GNUTLS_DIG_%s" % hash_type)
+ else:
+ # XXX improperlyconfigured or something
+ raise Exception("digest function not supported")
+
+ gnutls.library.functions.gnutls_x509_crt_get_fingerprint(
+ cert._c_object, digestfunction,
+ ctypes.byref(_buffer), ctypes.byref(buffer_length))
+
+ # deinit
+ #server_cert._X509Certificate__deinit(server_cert._c_object)
+ # needed? is segfaulting
+
+ fpr = ctypes.string_at(_buffer, buffer_length.value)
+ hex_fpr = sep.join(u"%02X" % ord(char) for char in fpr)
+
+ return hex_fpr
+'''
diff --git a/src/leap/crypto/leapkeyring.py b/src/leap/crypto/leapkeyring.py
index bceadc75..c241d0bc 100644
--- a/src/leap/crypto/leapkeyring.py
+++ b/src/leap/crypto/leapkeyring.py
@@ -53,12 +53,14 @@ class LeapCryptedFileKeyring(keyring.backend.CryptedFileKeyring):
def leap_set_password(key, value, seed="xxx"):
+ key, value = map(unicode, (key, value))
keyring.set_keyring(LeapCryptedFileKeyring(seed=seed))
keyring.set_password('leap', key, value)
def leap_get_password(key, seed="xxx"):
keyring.set_keyring(LeapCryptedFileKeyring(seed=seed))
+ #import ipdb;ipdb.set_trace()
return keyring.get_password('leap', key)
diff --git a/src/leap/crypto/tests/__init__.py b/src/leap/crypto/tests/__init__.py
new file mode 100644
index 00000000..e69de29b
--- /dev/null
+++ b/src/leap/crypto/tests/__init__.py
diff --git a/src/leap/crypto/tests/test_certs.py b/src/leap/crypto/tests/test_certs.py
new file mode 100644
index 00000000..e476b630
--- /dev/null
+++ b/src/leap/crypto/tests/test_certs.py
@@ -0,0 +1,22 @@
+import unittest
+
+from leap.testing.https_server import where
+from leap.crypto import certs
+
+
+class CertTestCase(unittest.TestCase):
+
+ def test_can_load_client_and_pkey(self):
+ with open(where('leaptestscert.pem')) as cf:
+ cs = cf.read()
+ with open(where('leaptestskey.pem')) as kf:
+ ks = kf.read()
+ certs.can_load_cert_and_pkey(cs + ks)
+
+ with self.assertRaises(certs.BadCertError):
+ # screw header
+ certs.can_load_cert_and_pkey(cs.replace("BEGIN", "BEGINN") + ks)
+
+
+if __name__ == "__main__":
+ unittest.main()
generated by cgit v1.2.3 (git 2.25.1) at 2024-11-07 02:51:06 +0000