Compare commits

...

11 Commits

View File

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