Your Name
689 lines
25 KiB
Python
689 lines
25 KiB
Python
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
|
|
# The term "Broadcom" refers to Broadcom Inc.
|
|
# and/or its subsidiaries.
|
|
|
|
import base64
|
|
import subprocess
|
|
import OpenSSL
|
|
import re
|
|
|
|
from datetime import datetime
|
|
|
|
from lib.exceptions import CommandExecutionError
|
|
from lib.command_runner import CommandRunner
|
|
from lib.console import print_text_error
|
|
from lib.input import MenuInput
|
|
from lib.host_utils import get_file_contents, get_file_type
|
|
from lib.text_utils import TextFilter
|
|
|
|
|
|
OPENSSL_CLI = '/usr/bin/openssl'
|
|
CERTOOL_CLI = '/usr/lib/vmware-vmca/bin/certool'
|
|
|
|
|
|
def get_command_output_with_input(args, input_text):
|
|
"""
|
|
Get command output with the specified input
|
|
|
|
:param args: Command and its arguments
|
|
:param input_text: input text to be supplied to stdin
|
|
:return: command output from stdout
|
|
"""
|
|
cmd = CommandRunner(*args, command_input=input_text, expected_return_code=0)
|
|
_, stdout, _ = cmd.run()
|
|
return stdout
|
|
|
|
|
|
def get_certificate_info(pem_text):
|
|
"""
|
|
Get certificate information from openssl command
|
|
This is a wrapper for 'openssl x509 -text -fingerprint -noout' command
|
|
|
|
:param pem_text: Certificate in PEM format
|
|
:return: Certificate info generated by openssl command
|
|
"""
|
|
args = [OPENSSL_CLI, 'x509', '-noout', '-text', '-fingerprint']
|
|
return get_command_output_with_input(args, pem_text).strip()
|
|
|
|
|
|
def get_subject_hash(pem_cert):
|
|
"""
|
|
Get the Subject hash for a certificate
|
|
This is a wrapper for the 'openssl x509 -noout -subject_hash' command
|
|
|
|
:param pem_cert: Base64 certificate hash
|
|
:return: Subject hash returned by openssl
|
|
"""
|
|
args = [OPENSSL_CLI, 'x509', '-noout', '-subject_hash']
|
|
return get_command_output_with_input(args, pem_cert).strip()
|
|
|
|
|
|
def get_formatted_dn(x509_name, use_slash_sep=False):
|
|
"""
|
|
Get formatted distinguished name for subject and issuer
|
|
|
|
The original vCert depends on 'openssl x509 -subject' format, but the output format
|
|
varies between openssl versions.
|
|
- openssl 1.0 (VC 6.x):
|
|
subject= /CN=CA/DC=vsphere/DC=local/C=US/ST=California/O=sc2-10-184-104-251.eng.vmware.com/OU=VMware Engineering
|
|
- openssl 3.0 (VC 7.x, 8.x):
|
|
subject=CN = CA, DC = vsphere, DC = local, C = US, ST = California, O = sc2-10-184-104-251.eng.vmware.com, OU = VMware Engineering
|
|
- openssl 3.2:
|
|
subject=CN=CA, DC=vsphere, DC=local, C=US, ST=California, O=sc2-10-184-104-251.eng.vmware.com, OU=VMware Engineering
|
|
Unless requested, we will use the last one since it's the most widely used format.
|
|
|
|
:param x509_name: OpenSSL.crypto.X509Name object
|
|
:param use_slash_sep: Use slash separator
|
|
:return: formatted distinguished name
|
|
"""
|
|
comps = []
|
|
comp_map = dict()
|
|
for comp in x509_name.get_components():
|
|
comp_name = str(comp[0], 'utf-8')
|
|
comp_value = str(comp[1], 'utf-8')
|
|
comp_map[comp_name] = comp_value
|
|
comps.append("{}={}".format(comp_name, comp_value))
|
|
if use_slash_sep:
|
|
return "/{}".format('/'.join(comps))
|
|
else:
|
|
return ', '.join(comps)
|
|
|
|
|
|
def get_subject_and_issuer_dn(x509_cert):
|
|
"""
|
|
Get Subject DN and Issuer DN from X509 certificate in a formatted DN
|
|
|
|
:param x509_cert: X509Certificate object
|
|
:return: tuple (formatted subject DN, formatted issuer DN)
|
|
"""
|
|
return get_formatted_dn(x509_cert.get_subject()), get_formatted_dn(x509_cert.get_issuer())
|
|
|
|
|
|
def get_certificate_extensions(x509_cert):
|
|
"""
|
|
Get X509 certificate extensions as a dict object
|
|
|
|
:param x509_cert: X509 certificate
|
|
:return: X509 certificate extensions
|
|
"""
|
|
extensions = dict()
|
|
for idx in range(x509_cert.get_extension_count()):
|
|
ext = x509_cert.get_extension(idx)
|
|
ext_short_name = ext.get_short_name().decode('utf-8')
|
|
try:
|
|
extensions[ext_short_name] = str(ext)
|
|
except OpenSSL.crypto.Error:
|
|
pass
|
|
|
|
return extensions
|
|
|
|
|
|
def get_subject_keyid(x509_cert, remove_colons=False, allow_alternate=False):
|
|
"""
|
|
Get Subject Key Identifier from X509v3 extension
|
|
If the keyid is not available and {allow_alternate} is True, the keyid
|
|
will be generated using serial/subject_dn pair.
|
|
|
|
:param x509_cert: X509 certificate
|
|
:param remove_colons: Remove colon separators
|
|
:param allow_alternate: Allow alternate keyid value using serial/subject_dn
|
|
:return: subject keyid
|
|
"""
|
|
extensions = get_certificate_extensions(x509_cert)
|
|
skid = extensions.get('subjectKeyIdentifier')
|
|
if skid:
|
|
if remove_colons:
|
|
skid = skid.replace(':', '')
|
|
elif allow_alternate:
|
|
serial = get_serial_number(x509_cert)
|
|
subject_dn = get_formatted_dn(x509_cert.get_subject(), use_slash_sep=True)
|
|
skid = "DirName:{},serial:{}".format(subject_dn, serial)
|
|
else:
|
|
skid = ''
|
|
return skid
|
|
|
|
|
|
def get_authority_keyid(x509_cert, remove_colons=False, allow_alternate=False):
|
|
"""
|
|
Get Authority Key Identifier from X509v3 extension
|
|
If the keyid is not available and {allow_alternate} is True, the keyid
|
|
will be generated using serial/DirName pair, if available
|
|
|
|
:param x509_cert: X509 certificate
|
|
:param remove_colons: Remove colon separators
|
|
:param allow_alternate: Allow alternate keyid value using serial/subject_dn
|
|
:return: authority keyid
|
|
"""
|
|
akid = ''
|
|
extensions = get_certificate_extensions(x509_cert)
|
|
akids = extensions.get('authorityKeyIdentifier')
|
|
if akids:
|
|
akid = TextFilter(akids).start_with('keyid:').get_text()
|
|
if not akid and len(akids.splitlines()) == 1 and len(akids.split(':')) >= 7:
|
|
akid = akids
|
|
if akid:
|
|
akid = akid.replace('keyid:', '')
|
|
if remove_colons:
|
|
akid = akid.replace(':', '')
|
|
elif akids and allow_alternate:
|
|
dir_name = TextFilter(akids).start_with('DirName:').get_text()
|
|
serial = TextFilter(akids).start_with('serial:').get_text()
|
|
if dir_name and serial:
|
|
akid = "{},{}".format(dir_name, serial)
|
|
return akid
|
|
|
|
|
|
def get_serial_number(x509_cert):
|
|
"""
|
|
Get certificate serial number in colon separated capitalized text,
|
|
e.g. 88:EC:FC:3B:FB:3F:53:52
|
|
:param x509_cert: X509 certificate
|
|
:return: certificate serial number
|
|
"""
|
|
serial = "{:X}".format(x509_cert.get_serial_number())
|
|
if len(serial) % 2 != 0:
|
|
serial = "0{}".format(serial)
|
|
parts = [serial[idx:idx+2] for idx in range(0, len(serial), 2)]
|
|
return ":".join(parts)
|
|
|
|
|
|
def get_certificate_start_date(x509_cert):
|
|
"""
|
|
Get certificate start date
|
|
|
|
:param x509_cert: X509 certificate object
|
|
:return: Certificate's start date as a datetime object
|
|
"""
|
|
input_format = '%Y%m%d%H%M%SZ'
|
|
return datetime.strptime(x509_cert.get_notBefore().decode('utf-8'), input_format)
|
|
|
|
|
|
def get_certificate_end_date(x509_cert):
|
|
"""
|
|
Get certificate end date
|
|
|
|
:param x509_cert: X509 certificate object
|
|
:return: Certificate's end date as a datetime object
|
|
"""
|
|
input_format = '%Y%m%d%H%M%SZ'
|
|
return datetime.strptime(x509_cert.get_notAfter().decode('utf-8'), input_format)
|
|
|
|
|
|
def get_certificate_fingerprint(x509_cert, algorithm=None, remove_colons=False):
|
|
"""
|
|
Get certificate fingerprint
|
|
|
|
:param x509_cert: X509 certificate object
|
|
:param algorithm: fingerprint algorithm. If it's not specified, use sha1
|
|
:param remove_colons: remove colon delimiters
|
|
:return:
|
|
"""
|
|
if algorithm is None:
|
|
algorithm = 'sha1'
|
|
fingerprint = x509_cert.digest(algorithm).decode('utf-8')
|
|
return fingerprint.replace(':', '') if remove_colons else fingerprint
|
|
|
|
|
|
def get_x509_certificate(pem_cert):
|
|
"""
|
|
Load X509Certificate from PEM text
|
|
|
|
:param pem_cert: Certificate in PEM format
|
|
:return: X509Certificate object
|
|
"""
|
|
return OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem_cert)
|
|
|
|
|
|
def get_certificate_expiry_in_days(x509_cert):
|
|
"""
|
|
Return certificate expiry in days
|
|
|
|
:param x509_cert: X509 certificate object
|
|
:return: certificate's expiry in days
|
|
"""
|
|
return (get_certificate_end_date(x509_cert) - datetime.utcnow()).days
|
|
|
|
|
|
def is_x509_expired(x509_cert):
|
|
"""
|
|
Checks if x509 object is expired
|
|
"""
|
|
days_left = get_certificate_expiry_in_days(x509_cert)
|
|
return True if days_left < 0 else False
|
|
|
|
|
|
def is_cert_expired(pem_cert):
|
|
"""
|
|
Checks if Base64 certificate is expired
|
|
"""
|
|
x509_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem_cert)
|
|
return is_x509_expired(x509_cert)
|
|
|
|
|
|
def is_cert_file_expired(cert_file):
|
|
"""
|
|
Checks if PEM certificate file is expired
|
|
"""
|
|
pem_cert = get_file_contents(cert_file)
|
|
return is_cert_expired(pem_cert)
|
|
|
|
def get_subject_alternative_names(x509_cert):
|
|
"""
|
|
Get SAN entries in the certificate
|
|
:param x509_cert: X590 certificate object
|
|
:return: list of SAN entries (including the prefixes)
|
|
"""
|
|
extensions = get_certificate_extensions(x509_cert)
|
|
return extensions.get('subjectAltName')
|
|
|
|
|
|
def get_cert_info_for_path_building(pem_cert):
|
|
"""
|
|
Obtain the basic certificate information for certificate path
|
|
building. The information includes:
|
|
- subject keyId
|
|
- authority keyId
|
|
- certificate name (common name)
|
|
- whether the certificate is self-signed
|
|
- whether the certificate is trusted
|
|
- whether the certificate is a CA
|
|
- pathlen of the certificate basic constraints
|
|
(or False if basic constraints are not present)
|
|
|
|
:param pem_cert: Certificate in PEM format
|
|
:return: dict object contains the required information
|
|
"""
|
|
x509_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem_cert)
|
|
subject_dn, issuer_dn = get_subject_and_issuer_dn(x509_cert)
|
|
skid = get_subject_keyid(x509_cert, remove_colons=True, allow_alternate=True)
|
|
akid = get_authority_keyid(x509_cert, remove_colons=True, allow_alternate=True)
|
|
basic_constraints = get_certificate_extensions(x509_cert).get('basicConstraints')
|
|
if basic_constraints is not None:
|
|
pathlen_search = re.search('pathlen:[0-9]+', basic_constraints)
|
|
pathlen = pathlen_search.group().split(':')[1] if pathlen_search is not None else False
|
|
else:
|
|
pathlen = False
|
|
if skid and akid:
|
|
is_selfsigned = skid == akid
|
|
else:
|
|
is_selfsigned = subject_dn == issuer_dn
|
|
|
|
cert_name = get_certificate_name(pem_cert)
|
|
|
|
return {
|
|
"subject_keyid": skid,
|
|
"auth_keyid": akid,
|
|
"cert_name": cert_name,
|
|
"is_selfsigned": is_selfsigned,
|
|
"is_trusted": False,
|
|
"is_ca": is_ca_certificate(x509_cert),
|
|
"pathlen": pathlen,
|
|
}
|
|
|
|
|
|
def get_certificate_fetcher_from_list(pem_certs):
|
|
"""
|
|
Method to compose certificate fetcher closure method from certificate list
|
|
in PEM text. The fetcher method will be used by build_certification_path
|
|
method.
|
|
|
|
:param pem_certs: List of certificate in PEM format
|
|
:return: subject keyId list, fetcher method
|
|
"""
|
|
subject_keyids = []
|
|
cert_map = dict()
|
|
for pem_cert in pem_certs:
|
|
cert = get_x509_certificate(pem_cert)
|
|
subject_keyid = get_subject_keyid(cert, remove_colons=True, allow_alternate=True)
|
|
if subject_keyid:
|
|
cert_map[subject_keyid] = pem_cert
|
|
subject_keyids.append(subject_keyid)
|
|
|
|
def fetch_certificate_closure(skid):
|
|
return cert_map.get(skid)
|
|
|
|
return subject_keyids, fetch_certificate_closure
|
|
|
|
|
|
def build_certification_path(pem_cert, trusted_subject_keyids, certificate_fetcher):
|
|
"""
|
|
Build certification path using a strict method of subject keyId and authority keyId
|
|
matching.
|
|
|
|
:param pem_cert: Certificate in PEM format. It may contain intermediate CA certificate
|
|
:param trusted_subject_keyids: list of subject keyId of trusted root CA certificates
|
|
:param certificate_fetcher: method to fetch certificate from subject keyId
|
|
:return: list of certificate info, from leaf certificate to root CA certificate
|
|
"""
|
|
cert_map = dict()
|
|
machine_certs = split_certificates_from_pem(pem_cert)
|
|
cur_cert = None
|
|
for idx, machine_cert in enumerate(machine_certs):
|
|
cert_info = get_cert_info_for_path_building(machine_cert)
|
|
subject_keyid = cert_info['subject_keyid']
|
|
cert_info['is_trusted'] = subject_keyid in trusted_subject_keyids
|
|
if idx == 0:
|
|
cur_cert = cert_info
|
|
else:
|
|
cert_map[subject_keyid] = cert_info
|
|
|
|
cert_path = [cur_cert]
|
|
while not cur_cert['is_selfsigned'] and cur_cert['auth_keyid'] is not None:
|
|
auth_keyid = cur_cert['auth_keyid']
|
|
cert_info = cert_map.get(auth_keyid)
|
|
if not cert_info:
|
|
next_pem_cert = certificate_fetcher(auth_keyid)
|
|
if not next_pem_cert:
|
|
break
|
|
cert_info = get_cert_info_for_path_building(next_pem_cert)
|
|
if cert_info['subject_keyid'] != auth_keyid:
|
|
break
|
|
cert_info['is_trusted'] = True
|
|
cert_map[cert_info["subject_keyid"]] = cert_info
|
|
cert_path.insert(0, cert_info)
|
|
cur_cert = cert_info
|
|
|
|
return cert_path
|
|
|
|
|
|
def split_certificates_from_pem(pem):
|
|
"""
|
|
Get list of individual certificates from PEM text
|
|
|
|
:param pem: Certificates in PEM format
|
|
:return: list of certificates in PEM format
|
|
"""
|
|
return TextFilter(pem).match_block('-----BEGIN CERTIFICATE-----',
|
|
'-----END CERTIFICATE-----', concatenate=True).get_lines()
|
|
|
|
|
|
def get_certificate_from_host(hostname, port=443):
|
|
"""
|
|
Get server TLS certificate from host via SSL handshake
|
|
:param hostname: hostname to connect
|
|
:param port: port
|
|
:return: Server certificate in PEM format
|
|
"""
|
|
remote_host = "{}:{}".format(hostname, port)
|
|
output = CommandRunner(OPENSSL_CLI, 's_client', '-connect', remote_host, '-showcerts',
|
|
command_input=subprocess.DEVNULL).run_and_get_output()
|
|
pem_cert = TextFilter(output).match_block('^-----BEGIN CERTIFICATE-----$',
|
|
'^-----END CERTIFICATE-----$',
|
|
concatenate=True).get_text()
|
|
return pem_cert
|
|
|
|
|
|
def build_pem_certificate(cert_text):
|
|
"""
|
|
Build PEM certificate based on certificate text input
|
|
This is required since certificates are often stored in DB
|
|
using a single line text.
|
|
|
|
:param cert_text: certificate text to be processed
|
|
:return: return certificate in a proper PEM format
|
|
"""
|
|
if type(cert_text) is bytes:
|
|
if cert_text.startswith(b'-----BEGIN CERTIFICATE-----'):
|
|
pem_cert = cert_text.replace(b'\\n', b'\n').decode('utf-8').strip()
|
|
elif cert_text.startswith(b'LS0tLS1CRUdJTiBDRVJUSUZJQ0FURS0tLS0t'):
|
|
pem_cert = base64.decodebytes(cert_text).decode('utf-8').strip()
|
|
else:
|
|
pem_cert = '\n'.join(['-----BEGIN CERTIFICATE-----',
|
|
base64.encodebytes(cert_text).decode('utf-8').strip(),
|
|
'-----END CERTIFICATE-----'])
|
|
else:
|
|
if cert_text.startswith('-----BEGIN CERTIFICATE-----'):
|
|
pem_cert = cert_text.replace('\\n', '\n')
|
|
else:
|
|
pem_cert = '\n'.join(['-----BEGIN CERTIFICATE-----', cert_text.strip(),
|
|
'-----END CERTIFICATE-----'])
|
|
|
|
# reformat to ensure that each line less than or equal to 72 characters
|
|
lines = pem_cert.splitlines()
|
|
contents = ''.join(lines[1:-1])
|
|
result = [lines[0]]
|
|
for line in contents.splitlines():
|
|
while len(line) > 72:
|
|
result.append(line[:72])
|
|
line = line[72:]
|
|
if line:
|
|
result.append(line)
|
|
result.append(lines[-1])
|
|
return '\n'.join(result)
|
|
|
|
|
|
def is_ca_certificate(x509_cert):
|
|
"""
|
|
Check if certificate is a CA certificate
|
|
:param x509_cert: certificate
|
|
:return: True of a CA certificate, otherwise False
|
|
"""
|
|
basic_constraints = get_certificate_extensions(x509_cert).get('basicConstraints')
|
|
key_usage = get_certificate_extensions(x509_cert).get('keyUsage')
|
|
is_ca = basic_constraints is not None and 'CA:TRUE' in basic_constraints
|
|
can_sign_certs = key_usage is not None and 'Certificate Sign' in key_usage
|
|
return is_ca and can_sign_certs
|
|
|
|
|
|
def is_self_signed_certificate(x509_cert):
|
|
"""
|
|
Check if the certificate is a self-signed certificate
|
|
:param x509_cert: X509 certificate
|
|
:return: True if self-signed, otherwise False
|
|
"""
|
|
subject_kid = get_subject_keyid(x509_cert, allow_alternate=True)
|
|
auth_keyid = get_authority_keyid(x509_cert, allow_alternate=True)
|
|
if subject_kid and auth_keyid:
|
|
return subject_kid == auth_keyid
|
|
else:
|
|
subject_dn, issuer_dn = get_subject_and_issuer_dn(x509_cert)
|
|
return subject_dn == issuer_dn
|
|
|
|
|
|
def generate_vmca_signed_certificate(config, output_dir, filename):
|
|
"""
|
|
Generate VMCA signed certificate using certool utility
|
|
In the {output_dir} directory, the following will be created:
|
|
- {filename}.key : the private key
|
|
- {filename}.pub : the public key
|
|
- {filename}.crt : the certificate file
|
|
|
|
:param config: certool config file
|
|
:param output_dir: The output directory
|
|
:param filename: The filename for the output (without extension)
|
|
"""
|
|
privkey_arg = "--privkey={}/{}.key".format(output_dir, filename)
|
|
pubkey_arg = "--pubkey={}/{}.pub".format(output_dir, filename)
|
|
CommandRunner(CERTOOL_CLI, '--genkey', privkey_arg, pubkey_arg, expected_return_code=0).run()
|
|
|
|
config_arg = "--config={}".format(config)
|
|
cert_arg = "--cert={}/{}.crt".format(output_dir, filename)
|
|
CommandRunner(CERTOOL_CLI, privkey_arg, '--gencert', cert_arg, config_arg, expected_return_code=0).run()
|
|
|
|
|
|
def detect_and_convert_to_pem(input_cert, password=None, allow_input=False):
|
|
"""
|
|
Detect certificate file format and convert it to PEM certificate as necessary
|
|
:param input_cert: The certificate file
|
|
:param password: In case of PKCS#12, the store password
|
|
:param allow_input: In case of PKCS#12, allow the method to ask for password
|
|
:return: tuple of ( certificate in PEM, private key in PEM of None )
|
|
"""
|
|
file_type = get_file_type(input_cert)
|
|
if 'PEM' in file_type or 'ASCII' in file_type:
|
|
contents = get_file_contents(input_cert)
|
|
cert_pem, key_pem = get_certificate_and_private_key(contents)
|
|
if cert_pem or key_pem:
|
|
return cert_pem, key_pem
|
|
|
|
# need conversion, try PKCS#12 format
|
|
if is_pkcs12_file(input_cert):
|
|
result = convert_pkcs12_to_pem(input_cert)
|
|
if password is not None:
|
|
result = convert_pkcs12_to_pem(input_cert, password)
|
|
if result is None:
|
|
result = convert_pkcs12_to_pem(input_cert, 'Antidisestablishmentarianism123581321',
|
|
allow_input=allow_input)
|
|
if result is None:
|
|
raise CommandExecutionError('Unable to validate keystore password')
|
|
return result
|
|
|
|
# DER format
|
|
cert_pem = CommandRunner(OPENSSL_CLI, 'x509', '-inform', 'der', '-in', input_cert,
|
|
'-outform', 'pem').run_and_get_output().strip()
|
|
key_pem = CommandRunner(OPENSSL_CLI, 'rsa', '-inform', 'der', '-in', input_cert,
|
|
'-outform', 'pem').run_and_get_output().strip()
|
|
output = "{}\n{}".format(cert_pem, key_pem) if cert_pem or key_pem else ''
|
|
|
|
if not output:
|
|
# PKCS#7, PEM
|
|
output = CommandRunner(OPENSSL_CLI, 'pkcs7', '-print_certs', '-inform', 'pem', '-in',
|
|
input_cert, '-outform', 'pem').run_and_get_output().strip()
|
|
if not output:
|
|
# PKCS#7, DER
|
|
output = CommandRunner(OPENSSL_CLI, 'pkcs7', '-print_certs', '-inform', 'der', '-in',
|
|
input_cert, '-outform', 'pem').run_and_get_output().strip()
|
|
if output:
|
|
return get_certificate_and_private_key(output)
|
|
else:
|
|
raise CommandExecutionError('Unsupported file format')
|
|
|
|
|
|
def is_pkcs12_file(filename):
|
|
ret_code, _, stderr = CommandRunner(OPENSSL_CLI, 'pkcs12', '-noout', '-in', filename,
|
|
'-passin', 'pass:').run()
|
|
if ret_code != 0 and 'asn1 error' in stderr:
|
|
return False
|
|
else:
|
|
return True
|
|
|
|
|
|
def convert_pkcs12_to_pem(cert_file, password=None, allow_input=False):
|
|
"""
|
|
Convert PKCS#12 format to PEM
|
|
:param cert_file:
|
|
:param password:
|
|
:param allow_input:
|
|
:return:
|
|
"""
|
|
password_arg = "pass:{}".format(password if password is not None else '')
|
|
ret_code, stdout, stderr = CommandRunner(OPENSSL_CLI, 'pkcs12', '-nokeys', '-info',
|
|
'-in', cert_file, '-passin', password_arg, '-nomacver').run()
|
|
if ret_code != 0 and allow_input and 'pkcs12 cipherfinal error' in stderr:
|
|
retry = 0
|
|
password = ''
|
|
while retry < 3 and not password:
|
|
password = MenuInput('Enter password for PKCS#12 keystore: ',
|
|
case_insensitive=False, masked=True).get_input()
|
|
password_arg = "pass:{}".format(password)
|
|
ret_code, stdout, _ = CommandRunner(OPENSSL_CLI, 'pkcs12', '-nokeys', '-info',
|
|
'-in', cert_file, '-passin', password_arg, '-nomacver').run()
|
|
if ret_code != 0:
|
|
print_text_error('Invalid password.')
|
|
password = ''
|
|
retry += 1
|
|
|
|
if ret_code == 0:
|
|
cert_pem, _ = get_certificate_and_private_key(stdout)
|
|
ret_code, stdout, _ = CommandRunner(OPENSSL_CLI, 'pkcs12', '-nocerts', '-nodes',
|
|
'-info', '-in', cert_file, '-passin', password_arg, '-nomacver').run()
|
|
_, key_pem = get_certificate_and_private_key(stdout)
|
|
return cert_pem, key_pem
|
|
|
|
return None
|
|
|
|
|
|
def get_certificate_and_private_key(text):
|
|
"""
|
|
Separate certificate and private key parts from PEM text
|
|
:param text: PEM text contains certificate and private key
|
|
"""
|
|
cert_pem = TextFilter(text)\
|
|
.match_block('^-----BEGIN CERTIFICATE-----',
|
|
'^-----END CERTIFICATE-----').get_text()
|
|
key_pem = TextFilter(text) \
|
|
.match_block('^-----BEGIN .*PRIVATE KEY-----',
|
|
'^-----END .*PRIVATE KEY-----') \
|
|
.get_text()
|
|
|
|
return cert_pem, key_pem
|
|
|
|
|
|
def get_key_modulus_from_pem_text(cert_or_key_text):
|
|
"""
|
|
Get the key modulus from certificate or private key text (PEM)
|
|
:param cert_or_key_text: certificate or key file
|
|
:return: The key modulus
|
|
"""
|
|
if TextFilter(cert_or_key_text).match('^-----BEGIN CERTIFICATE-----').get_lines():
|
|
modulus = CommandRunner(OPENSSL_CLI, 'x509', '-noout', '-modulus',
|
|
command_input=cert_or_key_text).run_and_get_output()
|
|
|
|
elif TextFilter(cert_or_key_text).match('^-----BEGIN .*PRIVATE KEY-----').get_text():
|
|
modulus = CommandRunner(OPENSSL_CLI, 'rsa', '-noout', '-modulus',
|
|
command_input=cert_or_key_text).run_and_get_output()
|
|
else:
|
|
return None
|
|
|
|
return TextFilter(modulus).cut('=', [1]).get_text().strip()
|
|
|
|
|
|
def generate_csr(openssl_config, csr_file, key_file):
|
|
"""
|
|
Generate CSR file using config {openssl_config}
|
|
:param openssl_config: OpenSSL config file
|
|
:param csr_file: The output CSR file
|
|
:param key_file: The output key file
|
|
"""
|
|
command = CommandRunner(OPENSSL_CLI, 'req', '-new', '-newkey', 'rsa:3072', '-nodes', '-config',
|
|
openssl_config, '-out', csr_file, '-keyout', key_file, expected_return_code=0)
|
|
command.run()
|
|
|
|
|
|
def load_pem_certificate_file_in_der(cert_file, leaf_only=False):
|
|
"""
|
|
Load PEM certificates and return them in DER format
|
|
"""
|
|
certs_pem = split_certificates_from_pem(get_file_contents(cert_file))
|
|
certs_der = []
|
|
for cert_pem in certs_pem:
|
|
cert_x509 = get_x509_certificate(cert_pem)
|
|
certs_der.append(OpenSSL.crypto.dump_certificate(OpenSSL.crypto.FILETYPE_ASN1, cert_x509))
|
|
if leaf_only:
|
|
return certs_der[:1]
|
|
return certs_der
|
|
|
|
|
|
def load_pem_key_file_in_pkcs8_der(key_file):
|
|
"""
|
|
Load PEM key file and return them in PKCS#8 DER format
|
|
"""
|
|
key_der = CommandRunner(OPENSSL_CLI, 'pkcs8', '-topk8', '-inform', 'pem', '-outform',
|
|
'der', '-in', key_file, '-out', '/dev/stdout', '-nocrypt',
|
|
command_input=subprocess.DEVNULL, binary_output=True).run_and_get_output()
|
|
return key_der
|
|
|
|
|
|
def get_certificate_name(pem_cert, target_dn='subject') -> str:
|
|
"""
|
|
Get the name of the entity in an x509 certificate, taken from the
|
|
Subject attributes CommonName, OrgUnit, and Organization (in that order of preference)
|
|
:param pem_cert: Base64 hash of a certificate
|
|
:return: string of the computed certificate name
|
|
"""
|
|
x509_cert = OpenSSL.crypto.load_certificate(OpenSSL.crypto.FILETYPE_PEM, pem_cert)
|
|
# get certificate name, typically will return the common name
|
|
comps = dict()
|
|
dn = x509_cert.get_subject() if target_dn == 'subject' else x509_cert.get_issuer()
|
|
for comp in dn.get_components():
|
|
comp_name = str(comp[0], 'utf-8')
|
|
comp_value = str(comp[1], 'utf-8')
|
|
comps[comp_name] = comp_value
|
|
|
|
cert_name = '<unknown>'
|
|
for key in ['CN', 'OU', 'O']:
|
|
if comps.get(key):
|
|
cert_name = comps[key]
|
|
break
|
|
return cert_name
|