Compare commits
11 Commits
2b4bfd70ef
...
ebe467260a
Author | SHA1 | Date | |
---|---|---|---|
ebe467260a | |||
b0562e1497 | |||
4cffe77fa0 | |||
66f45feea5 | |||
fe5286b8e5 | |||
e4ca594165 | |||
c866d219cb | |||
e22223ba66 | |||
1e5aa31eb8 | |||
02a83a72a2 | |||
|
fc3e220d61 |
@ -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)
|
||||||
|
Loading…
x
Reference in New Issue
Block a user