# 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 = '' for key in ['CN', 'OU', 'O']: if comps.get(key): cert_name = comps[key] break return cert_name