Changeset 87 for trunk/bin/bletchley-clonecertchain

Timestamp:

03/09/16 17:39:55 (9 years ago)

Author:

tim

Message:

moved general purpose SSL/TLS and certificate functions to a library module
minor improvements to error reporting

File:

• trunk/bin/bletchley-clonecertchain (modified) (5 diffs)

trunk/bin/bletchley-clonecertchain

@@ -24 +24 @@
 
 
-Copyright (C) 2014 Blindspot Security LLC
+Copyright (C) 2014,2016 Blindspot Security LLC
 Author: Timothy D. Morgan
 
@@ -44 +44 @@
 import traceback
 import socket
+from bletchley import ssltls
 try:
     import OpenSSL
@@ -52 +53 @@
     sys.stderr.write('NOTE: pyOpenSSL version 0.14 or later is required!\n')
     sys.exit(2)
-
-
-def createClientContext(method=SSL.SSLv3_METHOD):
-    tlsClientContext = SSL.Context(method)
-    tlsClientContext.set_verify(SSL.VERIFY_NONE, (lambda a,b,c,d,e: True))
-    return tlsClientContext
-
-
-def fetchCertificateChain(host, port):
-    protocols = [SSL.SSLv23_METHOD, SSL.TLSv1_METHOD, 
-                 SSL.TLSv1_1_METHOD, SSL.TLSv1_2_METHOD, 
-                 SSL.SSLv3_METHOD, SSL.SSLv2_METHOD]
-
-    chain = None
-    for p in protocols:
-        serverSock = socket.socket()
-        serverSock.connect((host,port))
-    
-        try:
-            server = SSL.Connection(createClientContext(p), serverSock)
-            server.set_connect_state()
-            server.do_handshake()
-        except Exception as e:
-            sys.stderr.write("Exception during handshake with server: \n")
-            traceback.print_exc(file=sys.stderr)
-            sys.stderr.write("\nThis could happen because the server requires "
-                             "certain SSL/TLS versions or a client certificiate."
-                             "  Have no fear, we'll keep trying...\n\n")
-
-        chain = server.get_peer_cert_chain()
-        if chain:
-            return chain
-
-    return chain
-
-
-def normalizeCertificateName(cert_name):
-    n = cert_name.get_components()
-    n.sort()
-    return tuple(n)
-
-
-def normalizeCertificateChain(chain):
-    # Organize certificates by subject and issuer for quick lookups
-    subject_table = {}
-    issuer_table = {}
-    for c in chain:
-        subject_table[normalizeCertificateName(c.get_subject())] = c
-        issuer_table[normalizeCertificateName(c.get_issuer())] = c
-
-    # Now find root or highest-level intermediary
-    root = None
-    for c in chain:
-        i = normalizeCertificateName(c.get_issuer())
-        s = normalizeCertificateName(c.get_subject())
-        if (i == s) or (i not in subject_table):
-            if root != None:
-                sys.stderr.write("WARN: Multiple root certificates found or broken certificate chain detected.")
-            else:
-                # Go with the first identified "root", since that's more likely to link up with the server cert
-                root = c
-
-    # Finally, build the chain from the top-down in the correct order
-    new_chain = []
-    nxt = root
-    while nxt != None:
-        new_chain = [nxt] + new_chain
-        s = normalizeCertificateName(nxt.get_subject())
-        nxt = issuer_table.get(s)
-    
-    return new_chain
-    
-
-def genFakeKey(certificate):
-    fake_key = OpenSSL.crypto.PKey()
-    old_pubkey = certificate.get_pubkey()
-    fake_key.generate_key(old_pubkey.type(), old_pubkey.bits())
-
-    return fake_key
-
-
-def getDigestAlgorithm(certificate):
-    # XXX: ugly hack because pyopenssl API for this is limited
-    if b'md5' in certificate.get_signature_algorithm():
-        return 'md5'
-    else:
-        return 'sha1'
-
-
-def deleteExtension(certificate, index):
-    import cffi
-    from cffi import FFI
-    ffi = FFI()
-    ffi.cdef('''void* X509_delete_ext(void* x, int loc);''')
-    libssl = ffi.dlopen('libssl.so')
-    ext = libssl.X509_delete_ext(certificate._x509, index)
-    #XXX: supposed to free ext here
-
-
-def removePeskyExtensions(certificate):
-    #for index in range(0,certificate.get_extension_count()):
-    #    e = certificate.get_extension(index)
-    #    print("extension %d: %s\n" % (index, e.get_short_name()), e)
-
-    index = 0
-    while index < certificate.get_extension_count():
-        e = certificate.get_extension(index)
-        if e.get_short_name() in (b'subjectKeyIdentifier', b'authorityKeyIdentifier'):
-            deleteExtension(certificate, index)
-            #XXX: would be nice if each of these extensions were re-added with appropriate values
-            index -= 1
-        index += 1
-    
-    #for index in range(0,certificate.get_extension_count()):
-    #    e = certificate.get_extension(index)
-    #    print("extension %d: %s\n" % (index, e.get_short_name()), e)
-
-
-def genFakeCertificateChain(cert_chain):
-    ret_val = []
-    cert_chain.reverse() # start with highest level authority
-
-    c = cert_chain[0]
-    i = normalizeCertificateName(c.get_issuer())
-    s = normalizeCertificateName(c.get_subject())
-    if s != i:
-        # XXX: consider retrieving root locally and including a forged version instead
-        c.set_issuer(c.get_subject())
-    k = genFakeKey(c)
-    c.set_pubkey(k)
-    removePeskyExtensions(c)
-    c.sign(k, getDigestAlgorithm(c))
-    ret_val.append(c)
-
-    prev = k
-    for c in cert_chain[1:]:
-        k = genFakeKey(c)
-        c.set_pubkey(k)
-        removePeskyExtensions(c)
-        c.sign(prev, getDigestAlgorithm(c))
-        prev = k
-        ret_val.append(c)
-
-    ret_val.reverse()
-    return k,ret_val
 
 
@@ -224 +80 @@
 
 #print("REAL CHAIN:")
-chain = fetchCertificateChain(options.host[0],options.port)
+connection = ssltls.ConnectSSLTLS(options.host[0],options.port)
+chain = ssltls.fetchCertificateChain(connection)
 #for c in chain:
 #    print(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_PEM, c).decode('utf-8'))
@@ -237 +94 @@
     sys.exit(2)
 
-fake_key, fake_chain = genFakeCertificateChain(chain)
+fake_key, fake_chain = ssltls.genFakeCertificateChain(chain)
 print(OpenSSL.crypto.dump_privatekey(OpenSSL.crypto.FILETYPE_PEM, fake_key).decode('utf-8'))
 for c in fake_chain: