sort_certificate.py

@@ -1,6 +1,6 @@
 #! /usr/bin/env python
 '''
-Sort X509/private key material
+Sort X509/RSA key material
 '''
 
 from __future__ import print_function
@@ -11,6 +11,7 @@ import fileinput
 from argparse import ArgumentParser
 from datetime import datetime
 from OpenSSL import crypto
+from Crypto.Util import asn1
 from cryptography.hazmat.primitives import serialization
 import certifi.core
 
@@ -29,84 +30,15 @@ SHA1 Fingerprint={sha1fingerprint}
 ASN1TIME_FMT = str('%Y%m%d%H%M%SZ'.encode('utf8'))
 OPENSSLTIME_FMT = '%b %e %T %Y GMT'
 
-
+class OnlyRSAKeyException(Exception):
-class PkDecorator(object):
     '''
-    Provide some information on the private key object
+    When we encounter other than RSA crypto material
-    '''
-    pk = None
-
-    def __init__(self, pk):
-        self.pk = pk
-
-    def __str__(self):
-        return "Private key"
-
-
-class PkDecoratorEC(PkDecorator):
-
-    def __str__(self):
-        pk_crypto = self.pk.to_cryptography_key()
-        return "EC Private key curve %s (%d bits)" % (
-                pk_crypto.curve.name, pk_crypto.key_size)
-
-
-class PkDecoratorRSA(PkDecorator):
-
-    def __str__(self):
-        pk_crypto = self.pk.to_cryptography_key()
-        return "RSA Private key %d bits (exponent %d)" % (
-                pk_crypto.key_size,
-                pk_crypto.private_numbers().public_numbers.e)
-
-
-class PkDecoratorDSA(PkDecorator):
-
-    def __str__(self):
-        pk_crypto = self.pk.to_cryptography_key()
-        return "DSA Private key %d bits" % pk_crypto.key_size
-
-
-class PkDecoratorDH(PkDecorator):
-
-    def __str__(self):
-        pk_crypto = self.pk.to_cryptography_key()
-        return "DH Private key %d bits" % pk_crypto.key_size
-
-
-class PkDecoratorFactory(object):
-    '''
-    Provide some information on the private key object
-    '''
-
-    def create(pk):
-        '''
-        Create the appropriate decorater object
-        '''
-        decorators = {
-                crypto.TYPE_DH: PkDecoratorDH,
-                crypto.TYPE_EC: PkDecoratorEC,
-                crypto.TYPE_DSA: PkDecoratorDSA,
-                crypto.TYPE_RSA: PkDecoratorRSA,
-        }
-        if pk.type() in decorators:
-            return decorators[pk.type()](pk)
-        else:
-            raise UnsupportedPkEncryption("Unsupported private key type %d"
-                                          % pk.type())
-
-
-class UnsupportedPkEncryption(Exception):
-    '''
-    When we encounter unsupported encryption algorithms
     '''
     pass
 
-
 class CertificateComponentException(Exception):
     '''
-    When something is not right with the whole cert+intermediates+private key
+    When something is not right with the whole cert+intermediates+private key bundle
-    bundle
     '''
     pass
 
@@ -134,33 +66,45 @@ def load_data(filenames):
     return datas
 
 
-def get_cert_pubkey(cert):
+def get_pub_modulus(cert):
     '''
-    Get the pubkey of a certificate
+    Get the modulus of a certificate
+    '''
+    pub = cert.get_pubkey()
+
+    # Only works for RSA (I think)
+    if pub.type() != crypto.TYPE_RSA:
+        logging.debug('Can only handle RSA crypto:'
+                      '\n\tsubject=%s\n\tissuer%s\n\texpired=%s\n\ttype=%s',
+                      cert.get_subject(),
+                      cert.get_subject(),
+                      cert.has_expired(),
+                      pub.type())
+        raise OnlyRSAKeyException('Can only handle RSA crypto')
+
+    pub_asn1 = crypto.dump_privatekey(crypto.FILETYPE_ASN1, pub)
+    pub_der = asn1.DerSequence()
+    pub_der.decode(pub_asn1)
+    pub_modulus = pub_der[1]
+
+    return pub_modulus
+
+
+def get_priv_modulus(priv):
+    '''
+    Get the modulus of a RSA private key
     '''
 
-    cert_crypto = cert.to_cryptography()
+    # Only works for RSA (I think)
-    pubkey = cert_crypto.public_key()
+    if priv.type() != crypto.TYPE_RSA:
-    pub_bytes = pubkey.public_bytes(
+        raise OnlyRSAKeyException('Can only handle RSA crypto')
-            serialization.Encoding.PEM,
-            serialization.PublicFormat.SubjectPublicKeyInfo)
 
-    return pub_bytes
+    priv_asn1 = crypto.dump_privatekey(crypto.FILETYPE_ASN1, priv)
+    priv_der = asn1.DerSequence()
+    priv_der.decode(priv_asn1)
+    priv_modulus = priv_der[1]
 
-
+    return priv_modulus
-def get_priv_pubkey(priv):
-    '''
-    Get the pubkey of a private key
-    '''
-
-    priv_crypto = priv.to_cryptography_key()
-
-    pubkey = priv_crypto.public_key()
-    pub_bytes = pubkey.public_bytes(
-            serialization.Encoding.PEM,
-            serialization.PublicFormat.SubjectPublicKeyInfo)
-
-    return pub_bytes
 
 
 def match_cert_privkey(cert, priv):
@@ -169,7 +113,7 @@ def match_cert_privkey(cert, priv):
     and reworked
     '''
 
-    return get_cert_pubkey(cert) == get_priv_pubkey(priv)
+    return get_pub_modulus(cert) == get_priv_modulus(priv)
 
 
 def find_root(x509_objects, root_issuers):
@@ -180,7 +124,6 @@ def find_root(x509_objects, root_issuers):
     logging.debug('Retrieved root certificate %s', root_cert.get_subject())
     return root_cert
 
-
 def find_intermediate_root(x509_objects, root_issuers):
     '''
     Find a suitable anchor by finding the intermediate that was signed by root
@@ -188,7 +131,7 @@ def find_intermediate_root(x509_objects, root_issuers):
 
     # Some intermediates have the *same* subject as some root certificates.
     # blacklist them
-    # XXX better use pubkey/hash for that, but can't find the appropriate
+    # XXX better use modulus/hash for that, but can't find the appropriate
     # interface to that at the moment
     excluded_issuers = [str(x.get_subject()) for x in x509_objects
                         if x.get_subject() != x.get_issuer()]
@@ -253,16 +196,14 @@ def order_x509(x509_objects, root_issuers):
             bundle.insert(0, x509_objects.pop(x509_objects.index(sibling[0])))
         else:
             # Lets complain
-            raise CertificateComponentException('Non matching certificates in '
+            raise CertificateComponentException('Non matching certificates in input:'
-                                                'input:'
                                                 ' No sibling found for %s'
                                                 % bundle[0].get_subject())
     return bundle
 
-
 def load_root_issuers():
     '''
-    Return the list of CA roots
+    Return the list of CA roots (RSA only)
     '''
     root_issuers = None
 
@@ -284,13 +225,13 @@ def load_root_issuers():
         for root_cert in root_certs:
             try:
                 logging.debug('subject=%s\n\tissuer%s\n\t'
-                              'expired=%s\n\tpubkey=%s',
+                              'expired=%s\n\tmodulus=%s',
                               root_cert.get_subject(),
                               root_cert.get_issuer(),
                               root_cert.has_expired(),
-                              get_cert_pubkey(root_cert))
+                              get_pub_modulus(root_cert))
-            except UnsupportedPkEncryption as unsupported_crypto_exception:
+            except OnlyRSAKeyException as onlyrsa_exception:
-                logging.debug(unsupported_crypto_exception)
+                logging.debug(onlyrsa_exception)
                 continue
 
         root_issuers = {str(root_cert.get_subject()): root_cert
@@ -302,7 +243,7 @@ def handle_args():
     '''
     Handle tool arguments
     '''
-    parser = ArgumentParser(description='Reorder X509/Private key data for'
+    parser = ArgumentParser(description='Reorder X509/RSA data for'
                                         ' hosting use')
 
     loggrp = parser.add_mutually_exclusive_group()
@@ -318,50 +259,29 @@ def handle_args():
 
     outputgrp = parser.add_mutually_exclusive_group()
 
-    outputgrp.add_argument('-c', '--check',
+    outputgrp.add_argument('--just-certificate', dest='print_cert',
-                           action='store_true',
+                           action='store_true', help='Just print certificate')
-                           help='Only check, output nothing')
+    outputgrp.add_argument('--no-certificate', dest='print_cert',
-    outputgrp.add_argument('--just-certificate',
-                           dest='print_cert',
-                           action='store_true',
-                           help='Just print certificate')
-    outputgrp.add_argument('--no-certificate',
-                           dest='print_cert',
                            action='store_false',
                            help='Omit certificate from output')
     outputgrp.set_defaults(print_cert=True)
 
-    outputgrp.add_argument('--just-chain',
+    outputgrp.add_argument('--just-chain', dest='print_chain',
-                           dest='print_chain',
+                           action='store_true', help='Just print chain')
-                           action='store_true',
+    outputgrp.add_argument('--no-chain', dest='print_chain',
-                           help='Just print chain')
+                           action='store_false', help='Omit chain from output')
-    outputgrp.add_argument('--no-chain',
+    outputgrp.add_argument('--include-root', dest='include_root',
-                           dest='print_chain',
+                           action='store_true', help='Also include the root certificate')
-                           action='store_false',
-                           help='Omit chain from output')
-    outputgrp.add_argument('--include-root',
-                           dest='include_root',
-                           action='store_true',
-                           help='Also include the root certificate')
     outputgrp.set_defaults(print_chain=True)
 
-    outputgrp.add_argument('--key',
+    outputgrp.add_argument('--key', dest='print_key',
-                           dest='print_key',
                            action='store_true', default=True,
                            help='Just print key')
-    outputgrp.add_argument('--no-key',
+    outputgrp.add_argument('--no-key', dest='print_key',
-                           dest='print_key',
+                           action='store_false', help='Omit key from output')
-                           action='store_false',
-                           help='Omit key from output')
     outputgrp.set_defaults(print_key=True)
 
-    parser.add_argument('-i', '--informational',
+    parser.add_argument('x509files', metavar='x509 file', nargs='*',
-                        action='store_true',
-                        help='Show some information about the PEM blocks')
-
-    parser.add_argument('x509files',
-                        metavar='x509 file',
-                        nargs='*',
                         help='x509 fullchain (+ rsa privkey)'
                              ' bundles to be checked')
 
@@ -373,11 +293,12 @@ def main():
     main program start and argument parsing
     '''
 
+
     root_issuers = None
 
     args = handle_args()
 
-    if args.verbose or args.check:
+    if args.verbose:
         logging.basicConfig(level=logging.INFO)
     elif args.debug:
         logging.basicConfig(level=logging.DEBUG)
@@ -423,28 +344,24 @@ def main():
             get_components()
 
         if len(rsa_objects) > 1:
-            raise CertificateComponentException('More than one RSA private key'
+            raise CertificateComponentException('More than one RSA private key found in input.'
-                                                ' found in input.'
                                                 ' Aborting')
         elif rsa_objects:
             if not match_cert_privkey(x509_objects[0], rsa_objects[0]):
                 raise CertificateComponentException('Provided certificate'
-                                                    ' and RSA private key'
+                                                    ' and RSA private key do not match')
-                                                    ' do not match')
             else:
-                logging.info('OK: Public key of provided certificate'
+                logging.info('OK: Modulus of provided certificate'
                              ' and RSA private key match')
         elif len(pk_objects) > 1:
-            raise CertificateComponentException('More than one private key'
+            raise CertificateComponentException('More than one RSA private key found in input.'
-                                                ' found in input.'
                                                 ' Aborting')
         elif pk_objects:
             if not match_cert_privkey(x509_objects[0], pk_objects[0]):
                 raise CertificateComponentException('Provided certificate'
-                                                    ' and private key'
+                                                    ' and private key do not match')
-                                                    ' do not match')
             else:
-                logging.info('OK: Public key of provided certificate'
+                logging.info('OK: Modulus of provided certificate'
                              ' and private key match')
 
         if args.include_root:
@@ -487,7 +404,6 @@ def main():
             logging.info('Subject: %s', x509_subject)
             logging.info('Issuer: %s', x509_issuer)
 
-            if args.informational:
             print(CERTINFO_TEMPLATE.format(
                 subject=x509_subject,
                 issuer=x509_issuer,
@@ -495,17 +411,13 @@ def main():
                 notafter=x509_not_after.strftime(OPENSSLTIME_FMT),
                 sha1fingerprint=x509_object.digest('sha1').decode()))
 
-            if not args.check:
             print(crypto.dump_certificate(crypto.FILETYPE_PEM,
                                           x509_object).decode('ascii'),
                   end='')
 
         if rsa_objects:
-            if not args.check:
             logging.info('Print RSA private keys')
             for rsa_object in rsa_objects:
-                    if args.informational:
-                        print(PkDecoratorFactory.create(rsa_object))
                 print(rsa_object.to_cryptography_key().private_bytes(
                     encoding=serialization.Encoding.PEM,
                     format=serialization.PrivateFormat.TraditionalOpenSSL,
@@ -513,19 +425,12 @@ def main():
                         'ascii'),
                       end='')
         elif pk_objects:
-            if not args.check:
             logging.info('Print private keys')
             for pk_object in pk_objects:
-                    if args.informational:
-                        print(PkDecoratorFactory.create(pk_object))
                 print(crypto.dump_privatekey(crypto.FILETYPE_PEM,
                                              pk_object).decode('ascii'),
                       end='')
 
 
 if __name__ == "__main__":
-    try:
     exit(main())
-    except CertificateComponentException as certcomponent_error:
-        logging.error(certcomponent_error)
-        exit(1)