add vmware-tools

This commit is contained in:
Your Name
2026-06-23 11:27:54 +08:00
parent 2a794fa8d3
commit 4f39d384e2
198 changed files with 14170 additions and 0 deletions
@@ -0,0 +1,6 @@
# Copyright (c) 2024 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
from . import view_certificate
from . import check_certificate
@@ -0,0 +1,981 @@
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import logging
import re
import OpenSSL
import glob
from enum import Enum
from lib import vcdb
from lib import vecs
from lib import vmdir
from lib.certificate_utils import (
get_x509_certificate, get_certificate_expiry_in_days, get_certificate_extensions,
get_certificate_fingerprint, build_certification_path, split_certificates_from_pem,
get_subject_keyid, get_certificate_from_host, build_pem_certificate,
is_ca_certificate, get_subject_hash, get_authority_keyid, get_subject_and_issuer_dn,
get_certificate_name
)
from lib.console import (
print_task, print_task_status, print_text_error, print_task_status_error,
print_task_status_warning, print_header, ColorKey, print_text
)
from lib.constants import (VMCA_CERT_FILE_PATH, RBD_CERT_FILE_PATH, READ, WRITE, VMWARE_DEPOT_HOST, VMWARE_DEPOT_CERT_ISSUER)
from lib.environment import Environment
from lib.exceptions import MenuExitException
from lib.host_utils import (
is_file_exists, get_file_contents, get_vc_version, VcVersion,
get_ip_address
)
from lib.menu import Menu
from lib.services import is_service_running
from operation.common import get_vcenter_extensions, get_vcenter_extension_expected_thumbprints
logger = logging.getLogger(__name__)
class CertificateStatus(Enum):
CERT_STATUS_EXPIRES_SOON = 'One or more certificates are expiring within 30 days'
CERT_STATUS_MISSING_PNID = 'One or more certificates are missing the PNID >>DETAILS<< from the SAN entry'
CERT_STATUS_MISSING_SAN = 'One or more certificates do not have any Subject Alternative Name values'
CERT_STATUS_KEY_USAGE = 'One or more certificates do not have the recommended\nKey Usage values'
CERT_STATUS_EXPIRED = 'One or more certificates are expired'
CERT_STATUS_NON_CA = 'One or more certificates are not CA certificates'
CERT_STATUS_BAD_ALIAS = \
'One or more entries in the TRUSTED_ROOTS store have an alias that is not the SHA1 thumbprint'
CERT_STATUS_SHA1_SIGNING = 'One or more certificates are signed using the SHA-1 algorithm'
CERT_STATUS_MISSING = 'One or more certificates are missing'
CERT_STATUS_MISSING_VMDIR = \
'One or more CA certificates are missing from VMware Directory'
CERT_STATUS_MISMATCH_SERVICE_PRINCIPAL = \
'One or more Solution User certificates does not match\nthe Service Principal certificate in VMware Directory'
CERT_STATUS_TOO_MANY_CRLS = 'The number of CRLs in VECS may be preventing some services from starting'
CERT_STATUS_MISSING_CA = \
'One or more certificates do not have all of the CA\n' \
'certificates in its signing chain in VMware Directory'
CERT_STATUS_EXPIRED_EMBEDDED_CA = \
'One or more certificates has a CA certificate embedded\n' \
'in its chain that is expired'
CERT_STATUS_STORE_MISSING = 'One or more VECS stores are missing'
CERT_STATUS_STORE_PERMISSIONS = 'One or more VECS stores are missing permissions'
CERT_STATUS_SERVICE_PRINCIPAL_MISSING = \
'One or more Service Principal entries are missing\nfrom VMware Directory'
CERT_STATUS_VMCA_EMPTY_CONFIG = \
'There are one or more vpxd.certmgmt.certs.cn.* settings with empty values\n' \
'This can cause issues pushing VMCA-signed certificates to ESXi hosts'
CERT_STATUS_VMCA_MODE = \
"The certificate management mode is set to 'thumbprint'\n" \
"This is not recommended, and should be set to 'vmca' or 'custom'"
CERT_STATUS_CLIENT_CA_LIST_FILE_MISSING = \
'The Smart Card issuing CA filter file does not exist at the following location:\n>>DETAILS<<'
CERT_STATUS_CLIENT_CA_LIST_FILE_EMPTY = \
'The Smart Card issuing CA filter file at the following location is empty:\n>>DETAILS<<'
CERT_STATUS_STS_VECS_CONFIG = \
'The STS server is configured to use a VECS store other than\nthe MACHINE_SSL_CERT store'
CERT_STATUS_STS_CONNECTION_STRINGS_NUMBER = \
'There are multiple STS ConnectionStrings values found in VMware Directory'
CERT_STATUS_STS_CONNECTION_STRINGS_HOSTNAME = \
'The STS ConnectionStrings value is not set properly for an SSO\ndomain with multiple Domain Controllers'
CERT_STATUS_UNSUPPORTED_SIGNATURE_ALGORITHM = \
'One or more certificates is using an unsupported\nsignature algorithm'
CERT_STATUS_CA_MISSING_SKID = 'One or more CA certificates is missing the Subject Key ID extension'
CERT_STATUS_ROGUE_CA = \
'One or more certificates are invalid because it or a signing CA\n' \
'extends beyond the pathlen restrictions of a parent CA'
CERT_STATUS_DUPLICATE_CA = \
'Two or more CA certificates in VMWare Directory or VECS have the\n' \
'same Subject string, which can cause issues with certificate\n' \
'validation'
TRUST_ANCHORS_MISMATCH = "One or more vCenter/PSC nodes have mismatched SSL trust anchors"
TRUST_ANCHORS_UNKNOWN = \
'The Machine SSL certificate could not be obtained from\n' \
'the following nodes to check SSL trust anchors:'
TRUST_ANCHORS_CHECK_URI_MISMATCH = \
'One or more vCenter/PSC nodes have mismatched SSL trust anchors and\n' \
'have Lookup Service registrations using the IP address instead\n' \
'of the PNID in the endpoint URIs. These can be fixed with the\n' \
'lsdoctor utility: https://knowledge.broadcom.com/external/article/320837/using-the-lsdoctor-tool.html'
TRUST_ANCHORS_CHECK_URI_IP = \
'One or more vCenter/PSC nodes have Lookup Service registrations\n' \
'using the IP address instead of the PNID in the endpoint URIs.\n' \
'These can be fixed with the lsdoctor utility:\n' \
'https://knowledge.broadcom.com/external/article/320837/using-the-lsdoctor-tool.html'
TRUST_ANCHORS_CHECK_URI_OTHER = \
'One or more vCenters have no Lookup Service registration endpoints\n' \
'using the current hostname or IP address. There could be registrations\n' \
'for these vCenters using a different hostname or IP address.\n' \
'These can be fixed with the lsdoctor utility: \n' \
'https://knowledge.broadcom.com/external/article/320837/using-the-lsdoctor-tool.html'
TRUST_ANCHORS_CHECK_PENDING = \
'The current Machine SSL certificate is not being served on port\n' \
'443 due to a pending service restart\n'
def check_certificate_dummy(**_):
print_text_error('=== Unsupported check certificate operation! ===')
def get_check_options(store, is_solution_user):
"""
Get check options based on VECS store name or solution user
:param store: VECS store name
:param is_solution_user: whether the certificate is solution user certificate
:return: list of check option
"""
if store == 'MACHINE_SSL_CERT':
options = ['CHECK_PNID', 'CHECK_KU', 'CHECK_SAN', 'CHECK_CA_CHAIN', 'CHECK_EMBEDDED_CHAIN', 'CHECK_ROGUE_CA']
elif store == 'SMS':
options = []
else:
options = ['CHECK_KU', 'CHECK_SAN', 'CHECK_CA_CHAIN', 'CHECK_EMBEDDED_CHAIN', 'CHECK_ROGUE_CA']
if is_solution_user:
options.append('CHECK_SERVICE_PRINCIPAL')
if store in ['wcp', 'wcpsvc']:
options.remove('CHECK_SAN')
return options
def check_vecs_certificate(store, alias, is_solution_user=False):
"""
Check specific certificate in VECS
:param store: VECS store name
:param alias: certificate alias
:param is_solution_user: whether the certificate is solution user certificate
"""
logger.info("Checking VECS certificate: store {}, alias {}".format(store, alias))
options = get_check_options(store, is_solution_user)
logger.info("Check options: {}".format(options))
aliases = vecs.get_certificate_aliases(store)
if alias not in aliases:
add_certificate_status(CertificateStatus.CERT_STATUS_MISSING)
print_task_status_error('NOT FOUND')
logger.error("Certificate for alias {} was not found in store {}".format(alias, store))
return
pem_cert = vecs.get_certificate(store, alias)
if not pem_cert:
print_task_status_error('PROBLEM')
logger.error("Failed to obtain certificate for alias {} in store {}".format(alias, store))
return
cert = get_x509_certificate(pem_cert)
days_left = get_certificate_expiry_in_days(cert)
cert_desc = "Certificate for alias {} in store {}".format(alias, store)
if days_left < 0:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRED)
print_task_status_warning('EXPIRED')
logger.warning("{} is expired".format(cert_desc))
return
elif days_left < 30:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRES_SOON)
print_task_status_warning("{} DAYS".format(days_left))
logger.info("{} will expire in {} days".format(cert_desc, days_left))
return
extensions = get_certificate_extensions(cert)
env = Environment.get_environment()
if 'CHECK_PNID' in options:
pnid = env.get_value('PNID')
san = extensions.get('subjectAltName')
if san is None or pnid not in san:
add_certificate_status(CertificateStatus.CERT_STATUS_MISSING_PNID, pnid)
print_task_status_warning('NO PNID')
logger.warning("{} does not have the PNID {} in the Subject Alternative Name field".format(cert_desc, pnid))
return
if 'CHECK_KU' in options:
if not check_key_usage(cert, "{}:{}".format(store, alias)):
add_certificate_status(CertificateStatus.CERT_STATUS_KEY_USAGE)
print_task_status_warning('KEY USAGE')
logger.warning("{} does not have the expected Key Usage values".format(cert_desc))
return
if 'CHECK_SAN' in options:
san = extensions.get('subjectAltName')
if not san:
add_certificate_status(CertificateStatus.CERT_STATUS_MISSING_SAN)
print_task_status_warning('NO SAN')
logger.warning("{} has no values in Subject Alternative Name field".format(cert_desc))
return
if 'CHECK_SERVICE_PRINCIPAL' in options:
fingerprint1 = get_certificate_fingerprint(cert, 'sha1')
solution_user_cert = vmdir.get_solution_user_certificate(store)
fingerprint2 = get_certificate_fingerprint(get_x509_certificate(solution_user_cert), 'sha1')
if fingerprint1 != fingerprint2:
add_certificate_status(CertificateStatus.CERT_STATUS_MISMATCH_SERVICE_PRINCIPAL)
print_task_status_warning('MISMATCH')
logger.warning("{} does not match the certificate for the corresponding Service Principal "
"in VMware Directory".format(cert_desc))
return
if 'CHECK_CA_CHAIN' in options:
if check_missing_ca(pem_cert):
add_certificate_status(CertificateStatus.CERT_STATUS_MISSING_CA)
print_task_status_warning('MISSING CA')
logger.warning("{} is missing one of the CA certificates in the certificate chain".format(cert_desc))
return
if 'CHECK_EMBEDDED_CHAIN' in options:
pem_certs = split_certificates_from_pem(pem_cert)
pem_certs = pem_certs[1:]
for pem in pem_certs:
x509_cert = get_x509_certificate(pem)
sha1_fingerprint = get_certificate_fingerprint(x509_cert)
logger.info("Checking embedded CA certificate with SHA1 fingerprint {}".format(sha1_fingerprint))
days_left = get_certificate_expiry_in_days(x509_cert)
if days_left < 0:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRED_EMBEDDED_CA)
print_task_status_warning('EMBEDDED CA')
logger.warning("The embedded CA certificate is expired:\n{}".format(pem))
return
if 'CHECK_ROGUE_CA' in options:
is_rogue_cert, _ = check_rogue_ca(pem_cert)
if is_rogue_cert:
add_certificate_status(CertificateStatus.CERT_STATUS_ROGUE_CA)
print_task_status_warning('ROGUE')
logger.warning("The certificate is invalid because of a CA that extends beyond a parent CA path length restriction:\n")
return
if not check_signature_algorithm(cert):
add_certificate_status(CertificateStatus.CERT_STATUS_UNSUPPORTED_SIGNATURE_ALGORITHM)
print_task_status_warning('ALGORITHM')
logger.warning("{} is signed with unsupported signature algorithm".format(cert_desc))
else:
print_task_status('VALID')
def check_signature_algorithm(x509_cert):
"""
Check if the signature algorithm is supported
:param x509_cert: X509Certificate object
:return: True if the signature algorithm is supported
"""
unsupported_signature_algs = ['md2WithRSAEncryption', 'md5WithRSAEncryption', 'RSASSA-PSS', 'dsaWithSHA1',
'ecdsa_with_SHA1', 'sha1WithRSAEncryption']
sign_alg = x509_cert.get_signature_algorithm().decode('utf-8')
logger.info("Checking certificate signature algorithm {} against unsupported signature algorithms {}"
.format(sign_alg, unsupported_signature_algs))
return False if sign_alg in unsupported_signature_algs else True
def check_key_usage(x509_cert, cert_desc):
"""
Check certificate keyUsage extension and validate that all keyUsage are in the supported list
:param x509_cert: X509Certificate object
:param cert_desc: Certificate description (for logging)
:return: True if all keyUsages are supported
"""
supported_kus = ['Digital Signature', 'Key Encipherment', 'Key Agreement', 'Data Encipherment',
'Non Repudiation']
logger.info("Checking Key Usage for cert {} among supported values of: {}".format(cert_desc, supported_kus))
extensions = get_certificate_extensions(x509_cert)
kus = extensions.get('keyUsage', '').split(', ')
for ku in kus:
if ku not in supported_kus:
logger.warning("Found unsupported Key Usage value: {}".format(ku))
return False
return True
def check_certificate_basic(cert):
"""
Some basic checks on certificate
:param cert: X509Certificate
"""
x509_cert = get_x509_certificate(cert)
days_left = get_certificate_expiry_in_days(x509_cert)
if days_left < 0:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRED)
print_task_status_warning('EXPIRED')
logger.warning("Certificate is expired")
return
elif days_left < 30:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRES_SOON)
print_task_status_warning("{} DAYS".format(days_left))
logger.warning("Certificate expires in {} days".format(days_left))
return
if not check_signature_algorithm(x509_cert):
add_certificate_status(CertificateStatus.CERT_STATUS_UNSUPPORTED_SIGNATURE_ALGORITHM)
print_task_status_warning('ALGORITHM')
logger.warning("Certificate is signed with unsupported signature algorithm")
return
if is_ca_certificate(x509_cert):
skid = get_subject_keyid(x509_cert)
if not skid:
add_certificate_status(CertificateStatus.CERT_STATUS_CA_MISSING_SKID)
print_task_status_warning('NO SKID')
return
print_task_status('VALID')
def check_file_system_certificate(file):
"""
Check certificate file stored on file system
:param file: Path of the certificate file
:return:
"""
if not is_file_exists(file):
add_certificate_status(CertificateStatus.CERT_STATUS_MISSING)
print_task_status_warning('NOT FOUND')
logger.error("Certificate at {} could not be found".format(file))
return
logger.info("Checking certificate at {}".format(file))
pem_cert = get_file_contents(file)
check_certificate_basic(pem_cert)
def check_certificate_status():
"""
Entry point for check certificate status operation
"""
env = Environment.get_environment()
print_task('Checking Machine SSL certificate')
check_vecs_certificate('MACHINE_SSL_CERT', '__MACHINE_CERT')
aliases = vecs.get_certificate_aliases('MACHINE_SSL_CERT')
if '__MACHINE_CSR' in aliases:
print_task('Checking Machine SSL CSR')
check_vecs_certificate('MACHINE_SSL_CERT', '__MACHINE_CSR')
print('Checking Solution User certificates:')
solution_users = env.get_value('SOLUTION_USERS')
for sol_user in solution_users:
print_task(" {}".format(sol_user))
check_vecs_certificate(sol_user, sol_user, is_solution_user=True)
vc_version = get_vc_version()
print_task('Checking SMS self-signed certificate')
check_vecs_certificate('SMS', 'sms_self_signed')
if vc_version >= VcVersion.V8:
print_task('Checking SMS VMCA-signed certificate')
check_vecs_certificate('SMS', 'sps-extension')
print_task('Checking data-encipherment certificate')
check_vecs_certificate('data-encipherment', 'data-encipherment')
print_task('Checking Authentication Proxy certificate')
check_file_system_certificate(VMCA_CERT_FILE_PATH)
print_task('Checking Auto Deploy CA certificate')
check_file_system_certificate(RBD_CERT_FILE_PATH)
cert_file = '/usr/lib/vmware-vmdir/share/config/vmdircert.pem'
if is_file_exists(cert_file):
print_task('Checking VMDir certificate')
check_file_system_certificate(cert_file)
store_list = vecs.get_store_list()
if 'BACKUP_STORE' in store_list:
print('Checking BACKUP_STORE entries:')
for alias in vecs.get_certificate_aliases('BACKUP_STORE'):
print_task(" {}".format(alias))
check_vecs_certificate('BACKUP_STORE', alias)
if 'BACKUP_STORE_H5C' in store_list:
print_task('Checking BACKUP_STORE_H5C entries:')
for alias in vecs.get_certificate_aliases('BACKUP_STORE_H5C'):
print_task(" {}".format(alias))
check_vecs_certificate('BACKUP_STORE_H5C', alias)
if 'STS_INTERNAL_SSL_CERT' in store_list:
print_task('Checking legacy Lookup Service certificate')
check_vecs_certificate('STS_INTERNAL_SSL_CERT', '__MACHINE_CERT')
print_task('Checking VMCA certificate')
check_file_system_certificate(VMCA_CERT_FILE_PATH)
def check_sts_tenant_certificates():
"""
Entry point for STS Tenant certificates check
"""
certs_map = vmdir.get_sts_tenant_certificates()
for tenant in certs_map.keys():
print("Checking {}:".format(tenant))
for pem_cert in certs_map[tenant]:
x509_cert = get_x509_certificate(pem_cert)
check_sts_tenant_certificate(x509_cert, tenant)
def check_sts_tenant_certificate(x509_cert, tenant):
"""
Check specific STS tenant certificate
:param x509_cert: X509Certificate object
:param tenant: Tenant name (for output message)
"""
is_ca = is_ca_certificate(x509_cert)
print_task(" {} {} certificate".format(tenant, 'CA' if is_ca else 'signing'))
cert_desc = "STS tenant certificate {}".format(tenant)
days_left = get_certificate_expiry_in_days(x509_cert)
if days_left < 0:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRED)
print_task_status_warning('EXPIRED')
logger.warning("{} is expired".format(cert_desc))
return
if is_ca:
skid = get_subject_keyid(x509_cert, remove_colons=True)
if not vmdir.get_ca_certificate(skid):
add_certificate_status(CertificateStatus.CERT_STATUS_MISSING_VMDIR)
print_task_status_warning('MISSING')
logger.warning("{} is missing from VMDir".format(cert_desc))
return
elif not check_key_usage(x509_cert, 'STS Tenant'):
add_certificate_status(CertificateStatus.CERT_STATUS_KEY_USAGE)
print_task_status_warning('KEY USAGE')
return
if not check_signature_algorithm(x509_cert):
add_certificate_status(CertificateStatus.CERT_STATUS_UNSUPPORTED_SIGNATURE_ALGORITHM)
print_task_status_warning('ALGORITHM')
logger.warning("{}is signed with unsupported signature algorithm".format(cert_desc))
return
if days_left < 30:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRES_SOON)
print_task_status_warning("{} DAYS".format(days_left))
logger.warning("{} expires in {} days".format(cert_desc, days_left))
else:
print_task_status('VALID')
def check_ca_certificates_in_vmdir():
"""
Entry point for CA certificates check in VMDir
"""
logger.info('Checking CA certificates in VMDir')
subject_keyids = vmdir.get_all_ca_subject_keyids()
for subject_keyid in subject_keyids:
logger.info("Checking certificate with CN(id) {}".format(subject_keyid))
pem_cert = vmdir.get_ca_certificate(subject_keyid)
print_task(subject_keyid)
cert = get_x509_certificate(pem_cert)
extensions = get_certificate_extensions(cert)
days_left = get_certificate_expiry_in_days(cert)
cert_subject_keyid = get_subject_keyid(cert)
basic_constraints = extensions.get('basicConstraints')
cert_desc = "Certificate with CN(id) {}".format(subject_keyid)
is_rogue_ca, _ = check_rogue_ca(pem_cert)
if days_left < 0:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRED)
print_task_status_warning('EXPIRED')
logger.warning("{} is expired".format(cert_desc))
elif basic_constraints is None or 'CA:TRUE' not in basic_constraints:
add_certificate_status(CertificateStatus.CERT_STATUS_NON_CA)
print_task_status_warning('NON-CA')
logger.warning("{} is not a CA certificate".format(cert_desc))
elif is_rogue_ca:
add_certificate_status(CertificateStatus.CERT_STATUS_ROGUE_CA)
print_task_status_warning('ROGUE')
logger.warning("{} violates the path length restriction of a parent CA certificate".format(cert_desc))
elif check_duplicate_ca(pem_cert):
add_certificate_status(CertificateStatus.CERT_STATUS_DUPLICATE_CA)
print_task_status_warning('DUPLICATE')
logger.warning("{} has a duplicate Subject string with one or more CA certificates".format(cert_desc))
elif not cert_subject_keyid:
add_certificate_status(CertificateStatus.CERT_STATUS_CA_MISSING_SKID)
print_task_status_warning('NO SKID')
logger.warning("{} does not have Subject Key Id".format(cert_desc))
elif days_left < 30:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRES_SOON)
print_task_status_warning("{} DAYS".format(days_left))
logger.warning("{} expires in {} days".format(cert_desc, days_left))
else:
print_task_status('VALID')
def check_ca_certificates_in_vecs():
"""
Entry point for CA certificates check in VECS
"""
logger.info('Checking CA certificates in VECS')
pem_certs, aliases = vecs.get_all_ca_certificates()
for pem_cert, alias in zip(pem_certs, aliases):
logger.info("Checking certificate with alias {}".format(alias))
print_task(alias)
cert = get_x509_certificate(pem_cert)
basic_constraints = get_certificate_extensions(cert).get('basicConstraints')
days_left = get_certificate_expiry_in_days(cert)
cert_desc = "Certificate with alias {}".format(alias)
is_rogue_ca, _ = check_rogue_ca(pem_cert)
if days_left < 0:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRED)
print_task_status_warning('EXPIRED')
logger.warning("{} is expired".format(cert_desc))
elif basic_constraints is None or 'CA:TRUE' not in basic_constraints:
add_certificate_status(CertificateStatus.CERT_STATUS_NON_CA)
print_task_status_warning('NON-CA')
logger.warning("{} is not a CA certificate".format(cert_desc))
elif is_rogue_ca:
add_certificate_status(CertificateStatus.CERT_STATUS_ROGUE_CA)
print_task_status_warning('ROGUE')
logger.warning("{} violates the path length restriction of a parent CA certificate".format(cert_desc))
elif check_duplicate_ca(pem_cert):
add_certificate_status(CertificateStatus.CERT_STATUS_DUPLICATE_CA)
print_task_status_warning('DUPLICATE')
logger.warning("{} has a duplicate Subject string with one or more CA certificates".format(cert_desc))
elif alias != get_certificate_fingerprint(cert, remove_colons=True).lower():
add_certificate_status(CertificateStatus.CERT_STATUS_BAD_ALIAS)
print_task_status_warning('BAD ALIAS')
logger.warning("{} is registered using a bad alias".format(cert_desc))
elif days_left < 30:
add_certificate_status(CertificateStatus.CERT_STATUS_EXPIRES_SOON)
print_task_status_warning("{} DAYS".format(days_left))
logger.warning("{} expires in {} days".format(cert_desc, days_left))
else:
print_task_status('VALID')
def check_service_principals():
"""
Entry point for service principals check
"""
logger.info('Checking service principals in VMware Directory')
service_principals = vmdir.get_service_principals()
if not service_principals:
print_task('Listing SSO Service Principals')
print_task_status_warning('FAILED')
logger.error('Could not get list of Service Principal entries from VMware Directory')
return
env = Environment.get_environment()
machine_id = env.get_value('MACHINE_ID')
solution_users = env.get_value('SOLUTION_USERS')
print("Node {}:".format(machine_id))
for solution_user in solution_users:
print_task(" {}".format(solution_user))
if "{}-{}".format(solution_user, machine_id) in service_principals:
print_task_status('PRESENT')
else:
print_task_status_warning('MISSING')
add_certificate_status(CertificateStatus.CERT_STATUS_SERVICE_PRINCIPAL_MISSING)
logger.warning("Missing service principal {} in VMware Directory".format(solution_user))
def check_crls():
"""
Check Certificate Revocation List in VECS
"""
logger.info("Checking the number of CRLS in VECS")
num_entries = len(vecs.get_certificate_aliases('TRUSTED_ROOT_CRLS'))
print_task('Number of CRLs in VECS')
if num_entries < 30:
print_task_status(num_entries)
elif num_entries < 100:
print_task_status_warning(num_entries)
else:
print_task_status_error(num_entries)
add_certificate_status(CertificateStatus.CERT_STATUS_TOO_MANY_CRLS)
logger.info("Number of CRLs in VECS: {}".format(num_entries))
def get_ids_domain_and_certificates(identity_sources, source_type):
result = []
for source in identity_sources:
if source['type'] == source_type:
result.append((source['domain_name'], source['certificates']))
return result
def get_ids_domain_and_certificates_by_domain(identity_sources, source_type, source_domain):
result = []
for source in identity_sources:
if source['type'] == source_type and source['domain_name'] == source_domain:
result.append((source['domain_name'], source['certificates']))
return result
def check_identity_source_certificates():
"""
Entry point for checking identity source certificates
"""
logger.info('Checking identity source certificates')
env = Environment.get_environment()
is_cac_configured = env.get_value('CAC_CONFIGURED')
if is_cac_configured:
check_smart_card_filter_file_certs()
check_smart_card_vmdir_certs()
identity_sources = vmdir.get_identity_sources()
# OpenLDAP
domain_and_certs = get_ids_domain_and_certificates(identity_sources, 'OpenLDAP')
if domain_and_certs:
print_header('Checking OpenLDAP LDAPS certificates')
for domain_name, certificates in domain_and_certs:
print("Domain: {}".format(domain_name))
for index, cert in enumerate(certificates, 1):
print_task(" Certificate {}".format(index))
check_certificate_basic(cert)
# AD over LDAP
domain_and_certs = get_ids_domain_and_certificates(identity_sources, 'AD over LDAP')
if domain_and_certs:
print_header('Checking AD over LDAPS certificates')
for domain_name, certificates in domain_and_certs:
print("Domain: {}".format(domain_name))
for index, cert in enumerate(certificates, 1):
print_task(" Certificate {}".format(index))
check_certificate_basic(cert)
# ADFS
domain_and_certs = get_ids_domain_and_certificates(identity_sources, 'ADFS')
if domain_and_certs:
print_header('Checking ADFS certificates')
for _, certificates in domain_and_certs:
for index, cert in enumerate(certificates, 1):
print_task("Certificate {}".format(index))
check_certificate_basic(cert)
def check_smart_card_filter_file_certs():
print_header('Check Smart Card Issuing CA Filter File')
print_task('Check CA Filter File')
env = Environment.get_environment()
cac_filter_file = env.get_value('SMART_CARD_FILTER_FILE')
if not is_file_exists(cac_filter_file):
print_task_status_warning('MISSING')
add_certificate_status(CertificateStatus.CERT_STATUS_CLIENT_CA_LIST_FILE_MISSING, cac_filter_file)
else:
pem = get_file_contents(cac_filter_file)
filter_certs = split_certificates_from_pem(pem)
if not filter_certs:
print_task_status_warning('EMPTY')
add_certificate_status(CertificateStatus.CERT_STATUS_CLIENT_CA_LIST_FILE_EMPTY, cac_filter_file)
return
print_task_status('OK')
for index, cert in enumerate(filter_certs, 1):
print_task("Certificate {}".format(index))
check_certificate_basic(cert)
def check_smart_card_vmdir_certs():
print_header('Check VMDir Smart Card Issuing CA Certificates')
cac_ca_certs = vmdir.get_smart_card_issuing_ca_certs()
for index, cert in enumerate(cac_ca_certs, 1):
print_task("Certificate {}".format(index))
check_certificate_basic(cert)
def check_ssl_trust_anchors():
"""
Check if trust anchor certificates are match to server SSL certificate by
calculating the certificate thumbprints
Note: This check seems not correct. The check should be a certificate chain
validation using the trust anchor, not just leaf certificate thumbprint
comparison.
"""
logger.info("Checking SSL trust anchors")
env = Environment.get_environment()
hostname = env.get_value('HOSTNAME')
sso_domain_nodes = vmdir.get_sso_domain_nodes()
is_mismatch = False
is_using_ip_address = False
is_pending = False
for node in sso_domain_nodes:
print_task(node)
try:
pem_cert = get_certificate_from_host(node, 443)
x509_cert = get_x509_certificate(pem_cert)
node_thumbprint = get_certificate_fingerprint(x509_cert)
except OpenSSL.crypto.Error:
add_certificate_status(CertificateStatus.TRUST_ANCHORS_UNKNOWN)
unknown_nodes = env.get_value('TRUST_ANCHORS_UNKNOWN_NODES')
if unknown_nodes is None:
unknown_nodes = []
env.set_value('TRUST_ANCHORS_UNKNOWN_NODES', unknown_nodes)
unknown_nodes.append(node)
print_task_status_warning('UNKNOWN')
continue
ip_address = get_ip_address(node)
trust_anchors = vmdir.get_node_trust_anchors(node)
if not trust_anchors:
is_using_ip_address = True
trust_anchors = vmdir.get_node_trust_anchors(ip_address)
if not trust_anchors:
add_certificate_status(CertificateStatus.TRUST_ANCHORS_CHECK_URI_OTHER)
print_task_status_warning('MISSING')
continue
for cert_entry in trust_anchors:
pem_cert = build_pem_certificate(cert_entry)
fingerprint = get_certificate_fingerprint(get_x509_certificate(pem_cert))
logger.info("Checking node thumbprint {} against trust anchor thumbprint {}"
.format(node_thumbprint, fingerprint))
if fingerprint != node_thumbprint:
if node == hostname or node == ip_address:
machine_ssl_cert = vecs.get_certificate('MACHINE_SSL_CERT', '__MACHINE_CERT')
x509_machine_ssl_cert = get_x509_certificate(machine_ssl_cert)
machine_ssl_thumbprint = get_certificate_fingerprint(x509_machine_ssl_cert)
if fingerprint == machine_ssl_thumbprint:
is_pending = True
else:
is_mismatch = True
logger.info("Searching for ghost trust anchors")
vcs = vmdir.get_registered_vcenters()
for deployment_id, dn in vcs:
uris = '\n'.join(vmdir.get_endpoint_registrations(dn))
pattern = ".*https://{0}/.*|.*https://{0}:.*|.*https://{1}/.*|.*https://{1}:.*".format(node, ip_address)
if not re.match(pattern, uris):
continue
logger.debug("Found vCenter registration for {}: {}".format(node, deployment_id))
ghost_vmonapi_dn = get_ghost_vmonapi_dn(deployment_id)
if not ghost_vmonapi_dn:
continue
logger.debug("cis.vmonapi registration DN: {}".format(ghost_vmonapi_dn))
trust_anchors = get_ghost_vmonapi_trust_anchors(ghost_vmonapi_dn)
for cert in trust_anchors:
pem_cert = build_pem_certificate(cert)
x509_cert = get_x509_certificate(pem_cert)
fingerprint = get_certificate_fingerprint(x509_cert)
logger.info("Checking node thumbprint {} against trust anchor thumbprint {}"
.format(node_thumbprint, fingerprint))
if node_thumbprint != fingerprint:
if node == hostname or node == ip_address:
machine_ssl_cert = vecs.get_certificate('MACHINE_SSL_CERT', '__MACHINE_CERT')
x509_machine_ssl_cert = get_x509_certificate(machine_ssl_cert)
machine_ssl_thumbprint = get_certificate_fingerprint(x509_machine_ssl_cert)
if fingerprint == machine_ssl_thumbprint:
is_pending = True
else:
is_mismatch = True
if is_mismatch:
add_certificate_status(CertificateStatus.TRUST_ANCHORS_MISMATCH)
if is_using_ip_address:
add_certificate_status(CertificateStatus.TRUST_ANCHORS_CHECK_URI_MISMATCH)
print_task_status_warning('MISMATCH*')
else:
print_task_status_warning('MISMATCH')
elif is_pending:
print_task_status_warning('PENDING')
add_certificate_status(CertificateStatus.TRUST_ANCHORS_CHECK_PENDING)
else:
if is_using_ip_address:
add_certificate_status(CertificateStatus.TRUST_ANCHORS_CHECK_URI_OTHER)
print_task_status_warning('CHECK URI')
else:
print_task_status('VALID')
def get_ghost_vmonapi_dn(deployment_id):
domain_dn = Environment.get_environment().get_value("SSO_DOMAIN_DN")
search_base = "cn=Sites,cn=Configuration,{}".format(domain_dn)
search_filter = "(&(vmwLKUPType=cis.vmonapi)(vmwLKUPDeploymentNodeId={}))".format(deployment_id)
search_attributes = ['dn']
results = vmdir.perform_ldap_search(search_base, search_filter, search_attributes)
return results[0]['dn'] if results else ''
def get_ghost_vmonapi_trust_anchors(vmonapi_dn):
search_filter = '(vmwLKUPURI=http://localhost*)'
search_attributes = ['vmwLKUPEndpointSslTrust']
results = vmdir.perform_ldap_search(vmonapi_dn, search_filter, search_attributes)
trust_anchors = []
for entry in results:
trust_anchors.extend(entry['vmwLKUPEndpointSslTrust'])
return trust_anchors
def check_vcenter_extension_thumbprints():
"""
Entry point for certificate check on vcenter extension thumbprints
"""
if not is_service_running('vmware-vpostgres'):
print_text_error('The vPostgres service is stopped!\n'''
'Please ensure this service is running before updating vCenter extension thumbprints.\n'
'Hint: Check the number of CRL entries in VECS')
return None
vcenter_extensions = get_vcenter_extensions()
extension_thumbprints = vcdb.get_extension_thumbprints(vcenter_extensions)
expected_thumbprints = get_vcenter_extension_expected_thumbprints(vcenter_extensions)
check_result = True
for extension, (thumbprint, db_cert_pem) in extension_thumbprints.items():
expected_thumbprint, expected_cert_type, expected_cert_pem = expected_thumbprints[extension]
print_task("{} ({})".format(extension, expected_cert_type))
logger.info("Comparing {} thumbprint of {} to {}".format(extension, thumbprint, expected_thumbprint))
if get_vc_version() < VcVersion.V9:
if thumbprint == expected_thumbprint:
print_task_status('MATCHES')
else:
print_task_status_warning('MISMATCH')
check_result = False
else:
if thumbprint == expected_thumbprint and db_cert_pem == expected_cert_pem:
print_task_status('MATCHES')
else:
print_task_status_warning('MISMATCH')
check_result = False
return check_result
def add_certificate_status(cert_status, detail=None):
env = Environment.get_environment()
results = env.get_value('CERTIFICATE_CHECK_RESULT')
if results is None:
results = []
env.set_value('CERTIFICATE_CHECK_RESULT', results)
if cert_status not in [r[0] for r in results]:
results.append([cert_status, detail])
def reset_certificate_status():
env = Environment.get_environment()
env.set_value('CERTIFICATE_CHECK_RESULT', None)
env.set_value('TRUST_ANCHORS_UNKNOWN_NODES', None)
def show_check_result_summary():
env = Environment.get_environment()
results = env.get_value('CERTIFICATE_CHECK_RESULT')
if not results:
return
print_text_error('\n------------------------!!! Attention !!!------------------------')
for status, detail in results:
lines = status.value.format_map(env.get_map()).splitlines()
for idx, line in enumerate(lines):
print_text_error(' - ' if idx == 0 else ' ', end='')
if status == CertificateStatus.TRUST_ANCHORS_UNKNOWN:
for unknown_node in env.get_value('TRUST_ANCHORS_UNKNOWN_NODES'):
print_text_error(" {}".format(unknown_node))
print_text_error(line if not detail else line.replace('>>DETAILS<<', detail))
def check_rogue_ca(pem_cert):
"""
Check if a certificate is invalid due to a CA extending beyond the
path length restriction of a parent CA
:param pem_cert: Base64 certificate hash
:return cert_is_rogue: True if certificate is invalid due to a rogue CA, False otherwise,
rogue_ca: name of CA in the signing chain that violates the path length restriction
of a parent CA
"""
logger.info('Entering the function to check for Rogue CA')
cert_is_rogue = False
rogue_ca = ''
subject_keyids = vmdir.get_all_ca_subject_keyids()
cert_path = build_certification_path(pem_cert, subject_keyids, vmdir.get_ca_certificate)
ca_certs_in_path = len(cert_path) if cert_path[-1]['is_ca'] is True else len(cert_path)-1
for index, cert in enumerate(cert_path):
logger.info("Checking for path length restrictions on certificate {}".format(cert_path[index]['cert_name']))
if cert['pathlen'] is not False:
logger.info("Path length restriction found on certificate {}: {}".format(cert_path[index]['cert_name'], cert_path[index]['pathlen']))
max_allowed_ca = index + int(cert['pathlen']) + 1
logger.info("Maximum allowed CAs in the signing chain is: {}".format(max_allowed_ca))
if ca_certs_in_path > max_allowed_ca:
cert_is_rogue = True
rogue_ca = cert_path[max_allowed_ca]['cert_name']
logger.info("Path length violation found on certificate {}".format(cert_path[max_allowed_ca]['cert_name']))
break
return cert_is_rogue, rogue_ca
def check_duplicate_ca(pem_cert):
"""
Check if there are multiple CA certificates with the same Subject string (but different private key).
This can cause certificate validation issues.
:param pem_cert: Base64 certificate hash
:return True if there are found to be duplicate certificates with the same Subject string, False otherwise
"""
subject_hash = get_subject_hash(pem_cert)
return True if len(glob.glob("/etc/ssl/certs/{}.[0-9]".format(subject_hash))) > 1 else False
def check_missing_ca(pem_cert, provided_ca_chain=False):
"""
Check to see if any CA certificates in the signing chain of a cert are missing from VMware Directory
:param pem_cert: Base64 certificate hash
:return: True if there is a CA certificate in the signing chain missing from VMware Directory, False otherwise
"""
ca_is_missing = False
x509_cert = get_x509_certificate(pem_cert)
if not provided_ca_chain:
subject_keyids = vmdir.get_all_ca_subject_keyids()
else:
subject_keyids = []
ca_certs = split_certificates_from_pem(get_file_contents(provided_ca_chain))
for cert in ca_certs:
ca_x509 = get_x509_certificate(cert)
subject_keyids.append(get_authority_keyid(ca_x509, True))
current_keyid = get_authority_keyid(x509_cert, True)
while True:
logger.info("Looking for AKID {} in {}".format(current_keyid, subject_keyids))
if current_keyid not in subject_keyids:
logger.info("AKID {} was NOT found in {}".format(current_keyid, subject_keyids))
ca_is_missing = True
break
else:
ca_x509 = get_x509_certificate(vmdir.get_ca_certificate(current_keyid))
subject_dn, issuer_dn = get_subject_and_issuer_dn(ca_x509)
# if we've reached the Root CA cert, we're done
if subject_dn == issuer_dn:
break
current_keyid = get_authority_keyid(ca_x509, True)
return ca_is_missing
def check_ssl_interception():
"""
Checks to see if SSL interception is occurring between vCenter and one of the online VMware repositories
"""
print_task("Checking {}".format(VMWARE_DEPOT_HOST))
depot_certs = split_certificates_from_pem(get_certificate_from_host('hostupdate.vmware.com'))
if len(depot_certs) == 0:
print_task_status_warning('FAILED')
print_text("{}Unable to obtain a certificate for {}".format(ColorKey.YELLOW, VMWARE_DEPOT_HOST))
print_text("Ensure the remote repository is accessible.{}".format(ColorKey.NORMAL))
return
else:
depot_issuer_cert_name = get_certificate_name(depot_certs[0], 'issuer')
logger.info("Certificate for {} is issued by {}".format(VMWARE_DEPOT_HOST, depot_issuer_cert_name))
if depot_issuer_cert_name == VMWARE_DEPOT_CERT_ISSUER:
print_task_status('OK')
print()
print_text("Issuing CA for {} is {}{}{}".format(VMWARE_DEPOT_HOST, ColorKey.GREEN, depot_issuer_cert_name, ColorKey.NORMAL))
return
else:
print_task_status_warning('WARNING')
print()
print_text("Issuing CA for {} is {}{}{}".format(VMWARE_DEPOT_HOST, ColorKey.YELLOW, depot_issuer_cert_name, ColorKey.NORMAL))
print_text("Expected issuer is {}{}{}".format(ColorKey.GREEN, VMWARE_DEPOT_CERT_ISSUER, ColorKey.NORMAL))
print()
print_text("{}SSL interception is likely taking place!{}".format(ColorKey.YELLOW, ColorKey.NORMAL))
env = Environment.get_environment()
menu = Menu.load_menu_from_config('config/check_config/ssl_interception/menu_manage_ssl_interception.yaml')
try:
menu.run()
except MenuExitException:
pass
@@ -0,0 +1,234 @@
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import logging
import xml.etree.ElementTree as ET
from lib.console import (
print_task, print_task_status, print_task_status_error,
print_task_status_warning, print_header, ColorKey, print_text
)
from lib.constants import STS_SERVER_CONFIG_FILE_PATH, STS_SERVER_CONFIG_PROPERTY_FILE_PATH
from lib.environment import Environment
from lib.exceptions import OperationFailed
from lib.host_utils import VcVersion, get_vc_version, is_file_exists
from lib.input import MenuInput
from lib.java_utils import load_property_file
from lib.vcdb import get_vmca_configuration_from_vcdb
from lib.vecs import (create_vecs_store_template, get_current_vecs_store_permissions, get_template_vecs_store_permissions,
get_missing_stores, manage_missing_vecs_stores, manage_missing_vecs_permissions
)
from lib.vmdir import get_sso_domain_nodes, perform_ldap_search
from operation.check_certificate import CertificateStatus, add_certificate_status
logger = logging.getLogger(__name__)
def check_sts_configuration():
print_header('Checking STS Server Configuration')
check_sts_certificate_configuration()
check_sts_connectionstring_configuration()
def check_sts_certificate_configuration():
env = Environment.get_environment()
vc_version = get_vc_version()
vc_build = int(env.get_value('VC_BUILD'))
print_task('Checking VECS store configuration')
# List of (versionConstraint, buildConstraint, key) tuples that identify
# what VECS stores are valid for the provided vCenter version.
versionConfigDB = [
# Version Build VECS Key
# ------------- ----- ----------------------------
(VcVersion.V7, None, 'localhost_connector_store'),
(VcVersion.V8, None, 'localhost_connector_store'),
(VcVersion.V9, None, 'localhost_connector_store'),
(VcVersion.V7, None, 'localhost_certificate_store'),
(VcVersion.V8, None, 'localhost_certificate_store'),
# Beginning with 9.0 there is just one certificate store.
(VcVersion.V9, None, 'certificate_store'),
(VcVersion.V7, 20845200, 'clientauth_connector_store'),
(VcVersion.V8, None, 'clientauth_connector_store'),
(VcVersion.V9, None, 'clientauth_connector_store'),
(VcVersion.V7, 20845200, 'clientauth_certificate_store'),
(VcVersion.V8, None, 'clientauth_certificate_store'),
]
# Initialize the STS configuration based on the VC version constraints.
sts_config = {}
for versionConstraint, buildConstraint, key in versionConfigDB:
if vc_version != versionConstraint:
continue
if buildConstraint and vc_build < buildConstraint:
continue
sts_config[key] = ''
# Obtain the STS configuration from vCenter.
if vc_version >= VcVersion.V9:
# Note: All entries in the versionConfigDB that match 9.0 must have an
# entry in the keyMap.
keyMap = {
'vmidentity.server.connector.ssl.localhost.certificate.keystore.file': 'localhost_connector_store',
'vmidentity.server.connector.ssl.store': 'certificate_store',
'vmidentity.server.connector.ssl.client.auth.certificate.keystore.file': 'clientauth_connector_store',
}
config = load_property_file(STS_SERVER_CONFIG_PROPERTY_FILE_PATH)
for key, value in config.items():
if key in keyMap:
sts_config[keyMap[key]] = value
else:
config = ET.parse(STS_SERVER_CONFIG_FILE_PATH)
root = config.getroot()
conn_cfg = {}
for connector in root.iter('Connector'):
if connector.attrib['port'] == '${bio-ssl-localhost.https.port}':
conn_cfg['localhost_connector_store'] = connector.attrib['store']
conn_cfg['localhost_certificate_store'] = connector.find('SSLHostConfig').find('Certificate').attrib['certificateKeystoreFile']
elif connector.attrib['port'] == '${bio-ssl-clientauth.https.port}':
conn_cfg['clientauth_connector_store'] = connector.attrib['store']
conn_cfg['clientauth_certificate_store'] = connector.find('SSLHostConfig').find('Certificate').attrib['certificateKeystoreFile']
for key, value in conn_cfg.items():
if key in sts_config:
sts_config[key] = value
else:
# This should never happen. It means somehow a key is in the
# STS server config file that doesn't belong for the given VC
# version, as indicated by versionConfigDB.
logger.warning('Unexpected STS server config key: %s' % key)
logger.info('Checking STS configuration on vCenter {} build {}'.format(vc_version, vc_build))
vecs_stores_to_replace = set()
update_sts_config = False
for key, value in sts_config.items():
description = key.replace('_', ' ').capitalize()
logger.info('{}: {}'.format(description, value))
if value != 'MACHINE_SSL_CERT':
update_sts_config = True
vecs_stores_to_replace.add(value)
sts_config[key] = value
sts_config['vecs_stores_to_replace'] = vecs_stores_to_replace
if update_sts_config:
add_certificate_status(CertificateStatus.CERT_STATUS_STS_VECS_CONFIG)
print_task_status_warning('LEGACY')
else:
print_task_status('OK')
return update_sts_config, sts_config
def check_sts_connectionstring_configuration():
domain_dn = Environment.get_environment().get_value('SSO_DOMAIN_DN')
sso_domain = Environment.get_environment().get_value('SSO_DOMAIN')
print_task('Checking STS ConnectionStrings')
search_dn = 'cn={},cn=IdentityProviders,cn={},cn=Tenants,cn=IdentityManager,cn=Services,{}'.format(sso_domain, sso_domain, domain_dn)
search_filter = '(objectclass=vmwSTSIdentityStore)'
search_attributes = ['vmwSTSConnectionStrings']
result = perform_ldap_search(search_dn, search_filter, search_attributes)
if result:
if len(get_sso_domain_nodes()) == 1:
print_task_status('OK')
return True, search_dn
sts_connection_strings = result[0]['vmwSTSConnectionStrings']
if len(sts_connection_strings) > 1:
add_certificate_status(CertificateStatus.CERT_STATUS_STS_CONNECTION_STRINGS_NUMBER)
print_task_status_warning('MISCONFIG')
return False, search_dn
if sts_connection_strings[0] != 'ldap://localhost:389':
add_certificate_status(CertificateStatus.CERT_STATUS_STS_CONNECTION_STRINGS_HOSTNAME)
print_task_status_warning('MISCONFIG')
return False, search_dn
print_task_status('OK')
return True, search_dn
else:
print_task_status_error('FAILED')
error_message = 'Unable to get the vmwSTSConnectionStrings attribute from VMware Directory'
raise OperationFailed(error_message)
def check_vmca_configuration_vcdb():
settings =get_vmca_configuration_from_vcdb()
for setting in settings:
setting_parts = setting.split('|')
setting_name = setting_parts[0].strip()
setting_value = setting_parts[1].strip()
if not setting_value:
add_certificate_status(CertificateStatus.CERT_STATUS_VMCA_EMPTY_CONFIG)
print_text("{:<48}{}'EMPTY'{}".format(setting_name, ColorKey.YELLOW, ColorKey.NORMAL))
elif setting_name == 'vpxd.certmgmt.mode' and 'setting_value' == 'thumbprint':
add_certificate_status(CertificateStatus.CERT_STATUS_VMCA_MODE)
print_text("{:<48}{}'{}'{}".format(setting_name, ColorKey.YELLOW, setting_value, ColorKey.NORMAL))
else:
print_text("{:<48}{}'{}'{}".format(setting_name, ColorKey.GREEN, setting_value, ColorKey.NORMAL))
def check_vecs_store_permissions(check_only=True):
env = Environment.get_environment()
vc_version = env.get_value('VC_VERSION_LONG')
vc_build = env.get_value('VC_BUILD')
vecs_permissions_template = '{}/config/vecs_permissions/vcsa-{}-{}.json'.format(env.get_value('SCRIPT_DIR'), vc_version, vc_build)
if is_file_exists(vecs_permissions_template):
print_text('Checking status and permissions for VECS stores:')
current_permissions = get_current_vecs_store_permissions()
template_permissions = get_template_vecs_store_permissions(vecs_permissions_template)
missing_stores = get_missing_stores()
missing_permissions = {'read' : {},
'write' : {}}
for store, permissions in template_permissions.items():
print_task(' {}'.format(store))
for expected_read_permission in permissions['read']:
if expected_read_permission not in current_permissions[store]['read']:
if store not in missing_permissions['read']:
missing_permissions['read'][store] = []
missing_permissions['read'][store].append(expected_read_permission)
for expected_write_permission in permissions['write']:
if expected_write_permission not in current_permissions[store]['write']:
if store not in missing_permissions['write']:
missing_permissions['write'][store] = []
missing_permissions['write'][store].append(expected_write_permission)
if store in missing_stores:
print_task_status_warning('MISSING')
add_certificate_status(CertificateStatus.CERT_STATUS_STORE_MISSING)
elif store in missing_permissions['read'] or store in missing_permissions['write']:
print_task_status_warning('PERMISSIONS')
add_certificate_status(CertificateStatus.CERT_STATUS_STORE_PERMISSIONS)
else:
print_task_status('OK')
if check_only:
return
if missing_stores:
manage_missing_vecs_stores(missing_stores)
if missing_permissions['read'] or missing_permissions['write']:
manage_missing_vecs_permissions(missing_permissions)
else:
print_text('{}VECS store template for {}-{} not found.{}'.format(ColorKey.YELLOW, vc_version, vc_build, ColorKey.NORMAL))
if check_only:
return
prompt = MenuInput('Create VECS store template? [N]: ', acceptable_inputs=['Y', 'N'], default_input='N')
print()
if prompt.get_input() == 'Y':
create_vecs_store_template(vecs_permissions_template)
+206
View File
@@ -0,0 +1,206 @@
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import ldap3 as ldap
import logging
from lib import vecs
from lib import vmdir
from lib.certificate_utils import get_certificate_fingerprint, get_x509_certificate, split_certificates_from_pem
from lib.console import print_text_error
from lib.constants import VMCAM_CERT_FILE_PATH
from lib.environment import Environment
from lib.exceptions import CommandExecutionError
from lib.host_utils import VcVersion, get_file_contents, get_vc_version, get_hostname
from lib.input import MenuInput
from lib.ldap_utils import Ldap, LdapException, get_user_dn
logger = logging.getLogger(__name__)
def prevalidate_credential(**_):
"""
Enforce that SSO credentials are properly populated
"""
env = Environment.get_environment()
_ = env.get_value('SSO_USERNAME')
_ = env.get_value('SSO_PASSWORD')
vmdir.init_env_identity_source()
vmdir.init_env_cac()
def verify_sso_credential(sso_username, sso_password):
"""
Verify the SSO credential via LDAP authentication
:param sso_username: Single Sign-On user name
:param sso_password: Single Sign-on password
:return: True if the credential can be verified, otherwise False
"""
hostname = get_hostname()
user_dn = get_user_dn(sso_username)
connection = None
try:
connection = Ldap.open_ldap_connection(node=hostname, user_dn=user_dn,
password=sso_password)
if Ldap.ldap_search(connection, "cn=schemacontext", "(objectClass=*)",
ldap.BASE, ["cn"]):
logger.info("Password verified successfully for user %s", user_dn)
return True
except LdapException:
pass
finally:
if connection:
Ldap.close_ldap_connection(connection)
logger.error("Password verification failed for user %s on host %s", user_dn, hostname)
return False
def populate_sso_credential():
"""
Callback method to populate SSO_USERNAME and SSO_PASSWORD environment variables
:return: dict containing both verified SSO_USERNAME and SSO_PASSWORD values
:raise CommandExecutionError: if the credential verification failed
"""
env = Environment.get_environment()
default_input = 'administrator@{}'.format(env.get_value('SSO_DOMAIN'))
counter_username = counter_password = 0
sso_username = None
print()
while counter_username < 3 and counter_password < 3:
if sso_username:
default_input = sso_username
menu_input = MenuInput('Please enter a Single Sign-On administrator account [{}]: '.format(default_input),
default_input=default_input, case_insensitive=False)
sso_username = menu_input.get_input()
if not check_sso_credential(sso_username):
sso_username = None
counter_username += 1
continue
menu_input = MenuInput('Please provide the password for {}: '.format(sso_username),
masked=True,
case_insensitive=False)
sso_password = menu_input.get_input()
if not verify_sso_credential(sso_username, sso_password):
counter_password += 1
continue
print()
return [('SSO_USERNAME', sso_username), ('SSO_PASSWORD', sso_password)]
raise CommandExecutionError('Invalid SSO credential')
def check_sso_credential(sso_username, sso_password=None):
"""
Check and verify the SSO credentials. It will validate that the SSO domain matches to
values obtained from VC. If password is provided, it will try to verify the credential
by performing LDAP authentication
:param sso_username: SSO user name
:param sso_password: SSO user password
:return: True if the validation passed, otherwise False
"""
env = Environment.get_environment()
sso_domain = env.get_value('SSO_DOMAIN')
user_name = None
user_sso_domain = None
if '@' in sso_username:
user_name, user_sso_domain = tuple(sso_username.split('@', 2))
if user_sso_domain != sso_domain:
print_text_error('Invalid domain, please provide an account in the SSO domain [{}].'.format(sso_domain))
return False
elif not user_name:
print_text_error('Invalid user name')
return False
if sso_password is None:
return True
return verify_sso_credential(sso_username, sso_password)
def get_vcenter_extensions():
"""
Get list of vCenter extensions
"""
env = Environment.get_environment()
vc_version = get_vc_version()
vc_build = int(env.get_value('VC_BUILD'))
# The following table entries are based on the output of:
# /usr/bin/psql -d VCDB -U postgres -c "SELECT ext_id, thumbprint FROM vpx_ext" -t
# Only the entries for a VC version that have a defined thumbprint should
# be in the table.
vcenterExtensionDB = [
# Version Build Extension
# ------------- -------- -------------------------
(VcVersion.V7, None, 'com.vmware.vsan.health'),
(VcVersion.V8, None, 'com.vmware.vsan.health'),
(VcVersion.V9, None, 'com.vmware.vsan.health'),
(VcVersion.V7, None, 'com.vmware.vcIntegrity'),
(VcVersion.V8, None, 'com.vmware.vcIntegrity'),
(VcVersion.V9, None, 'com.vmware.vcIntegrity'),
(VcVersion.V7, None, 'com.vmware.rbd'),
# At or after the following build number com.vmware.rbd is excluded.
(VcVersion.V8, 22385739, 'com.vmware.rbd'),
(VcVersion.V7, None, 'com.vmware.imagebuilder'),
(VcVersion.V8, None, 'com.vmware.imagebuilder'),
(VcVersion.V7, None, 'com.vmware.vmcam'),
(VcVersion.V8, None, 'com.vmware.vmcam'),
(VcVersion.V9, None, 'com.vmware.vmcam'),
(VcVersion.V7, None, 'com.vmware.vim.eam'),
(VcVersion.V8, None, 'com.vmware.vim.eam'),
(VcVersion.V9, None, 'com.vmware.vim.eam'),
(VcVersion.V8, None, 'com.vmware.vlcm.client'),
(VcVersion.V9, None, 'com.vmware.vlcm.client'),
]
vcenter_extensions = []
for versionConstraint, buildConstraint, key in vcenterExtensionDB:
if vc_version != versionConstraint:
continue
if buildConstraint and vc_build >= buildConstraint:
continue # The build number is an exclusion not inclusion.
vcenter_extensions.append(key)
return vcenter_extensions
def get_vcenter_extension_expected_thumbprints(vcenter_extensions):
"""
Get expected vcenter extension's thumbprint
:param vcenter_extensions: list of vcenter extension
:return: dict { vc_extension: (thumbprint, cert_type, cert_pem) }
"""
vpxd_ext_pem_cert = vecs.get_certificate('vpxd-extension', 'vpxd-extension')
vpxd_ext_thumbprint = get_certificate_fingerprint(get_x509_certificate(vpxd_ext_pem_cert))
machine_ssl_pem_cert = vecs.get_certificate('MACHINE_SSL_CERT', '__MACHINE_CERT')
machine_ssl_thumbprint = get_certificate_fingerprint(get_x509_certificate(machine_ssl_pem_cert))
vmcam_pem_cert = get_file_contents(VMCAM_CERT_FILE_PATH)
vmcam_thumbprint = get_certificate_fingerprint(get_x509_certificate(vmcam_pem_cert))
result = dict()
for extension in vcenter_extensions:
if extension == 'com.vmware.vmcam':
expected_thumbprint = vmcam_thumbprint
expected_cert_type = 'Authentication Proxy'
cert_pem = split_certificates_from_pem(vmcam_pem_cert)[0]
elif extension == 'com.vmware.vsan.health':
expected_thumbprint = machine_ssl_thumbprint
expected_cert_type = 'Machine SSL'
cert_pem = split_certificates_from_pem(machine_ssl_pem_cert)[0]
else:
expected_thumbprint = vpxd_ext_thumbprint
expected_cert_type = 'vpxd-extension'
cert_pem = split_certificates_from_pem(vpxd_ext_pem_cert)[0]
result[extension] = (expected_thumbprint, expected_cert_type, cert_pem)
return result
@@ -0,0 +1,884 @@
#!/usr/bin/env python3
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import stat
from lib.console import print_text, print_task_status, print_task_status_warning, ColorKey, \
print_text_warning, print_header, print_task_status_error
import logging
import os
import urllib3
import re
from lib.command_runner import CommandRunner
from lib.exceptions import OperationFailed, CommandExecutionError
from lib.input import MenuInput
from lib.menu import Menu
from lib import vcdb, vmdir
from lib import vecs
from lib.environment import Environment
from lib.certificate_utils import (
get_x509_certificate, build_certification_path, get_certificate_fingerprint, split_certificates_from_pem,
get_certificate_from_host, get_certificate_end_date, get_certificate_start_date,
get_subject_and_issuer_dn, get_subject_alternative_names, get_certificate_fetcher_from_list,
is_ca_certificate, detect_and_convert_to_pem, get_key_modulus_from_pem_text, generate_csr
)
from lib.console import (
print_task, print_text_error
)
from lib.constants import SPACE
from lib.host_utils import (
get_file_contents, save_text_to_file, find_files,
set_file_mode, is_file_exists
)
from operation.manage_certificate import find_matched_private_key, verify_certificate_and_key, obtain_ca_chain, \
get_csr_info, get_san_entries, generate_openssl_config, clear_csr_info, get_timestamp
urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
logger = logging.getLogger(__name__)
def check_esxi_vcenter_trust():
"""
Entry point for checking esxi vcenter certificate trust
"""
num_hosts = vcdb.get_number_of_hosts()
title = "Check ESXi/vCenter certificate trust"
text_color = ColorKey.YELLOW if num_hosts == 0 else ColorKey.GREEN
sub_title = "There are {}{}{} hosts connected to vCenter.\n".format(text_color, num_hosts, ColorKey.NORMAL)
menu = Menu()
if not get_hosts():
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, is_hidden=False, key='R',
is_default=True)
else:
menu.set_menu_options(title, sub_title, input_text='Select an option [Return to the previous menu]: ')
menu.add_menu_item("Perform check on all hosts (requires uniform root password on all hosts)",
view_certificate_for_all_esxi_hosts)
menu.add_menu_item("Perform check on all hosts in a cluster (requires uniform root password on all hosts)",
view_esxi_for_all_cluster)
menu.add_menu_item("Perform check on single host", view_esxi_certificate_for_specific_host)
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, is_hidden=True, key='R',
is_default=True)
menu.run()
def view_esxi_test_response():
print(True)
def display_num_hosts():
num_hosts = vcdb.get_number_of_hosts()
text_color = ColorKey.YELLOW if num_hosts == 0 else ColorKey.GREEN
print_text("There are {}{}{} hosts connected to vCenter.".format(text_color, num_hosts, ColorKey.NORMAL))
if num_hosts == 0:
print_text_error("Hosts not detected on the system")
print("\n")
def get_hosts():
hosts_str = vcdb.get_hosts()
if hosts_str == "":
print_text("No hosts available")
return []
hosts = []
for host_info in hosts_str:
host_info = host_info.strip()
cluster_id = host_info.split("|")[0].strip()
name = host_info.split("|")[1].strip()
ip = host_info.split("|")[2].strip()
host = [cluster_id, name, ip]
hosts.append(host)
return hosts
def get_host_password(pass_str):
password = ""
while password == "":
password = MenuInput(pass_str, masked=True, case_insensitive=False).get_input()
print()
return password
def get_host_details(hosts):
host_detail = ""
while host_detail == "":
host_detail = MenuInput('Enter FQDN or IP of the ESXi host: ',
allow_empty_input=False, case_insensitive=False).get_input()
logger.info("The entered FQDN or IP is : {}".format(host_detail))
if host_detail == "":
print_text("Please enter valid input")
continue
for cluster_id, name, ip in hosts:
if host_detail == name or host_detail == ip:
print()
return [name, ip]
host_detail = ""
print_text("Please enter valid input from below list")
print(flush=True)
for cluster_id, name, ip in hosts:
print_text("Host name : {} Host ip : {}".format(name, ip))
print(flush=True)
print()
return []
def get_host_name_or_ip(host_name, host_ip):
if host_name == "":
return host_ip
else:
return host_name
def get_certificate_management_mode():
return vcdb.get_certificate_management_mode()
def view_certificate_for_all_esxi_hosts():
hosts = get_hosts()
print("\n")
password = get_host_password("Enter root password for all ESXi hosts: ")
certificate_management_mode = get_certificate_management_mode()
print_text("Certificate Management Mode = {}{}{}\n".format(ColorKey.GREEN, certificate_management_mode, ColorKey.NORMAL))
for cluster_id, name, ip in hosts:
print_text("Host : {}{}{}".format(ColorKey.CYAN, get_host_name_or_ip(name, ip), ColorKey.NORMAL))
view_esxi_certificate_trust(get_host_name_or_ip(name, ip), certificate_management_mode, password)
print("\n\n")
def view_esxi_certificate_for_specific_host():
host_name, host_ip = get_host_details(get_hosts())
host_name_or_ip = get_host_name_or_ip(host_name, host_ip)
password = get_host_password("Enter root password for ESXi host {}: ".format(host_name_or_ip))
certificate_management_mode = get_certificate_management_mode()
print_text("Certificate Management Mode = {}{}{}".format(ColorKey.GREEN, certificate_management_mode, ColorKey.NORMAL))
print_text("\nHost : {}{}{}".format(ColorKey.CYAN, host_name_or_ip, ColorKey.NORMAL))
view_esxi_certificate_trust(host_name_or_ip, certificate_management_mode, password)
print("\n\n")
def get_host_and_cluster():
host_and_cluster_id = vcdb.get_host_and_cluster_id()
if host_and_cluster_id == "":
print_text("No hosts or Clusters available")
return {}
host_and_cluster_id_dict = {}
for host_and_cluster_id_info in host_and_cluster_id:
host_info = host_and_cluster_id_info.strip()
host_id = host_info.split("|")[0].strip()
cluster_id = host_info.split("|")[1].strip()
host_and_cluster_id_dict[host_id] = cluster_id
return host_and_cluster_id_dict
def find_hosts_in_cluster(input_cluster_id, hosts, host_and_cluster):
hosts_in_cluster = []
for host_id, name, ip in hosts:
if host_and_cluster[host_id] == input_cluster_id:
host_in_cluster = [host_id, name, ip]
hosts_in_cluster.append(host_in_cluster)
if not hosts_in_cluster:
print_text("No hosts were found in cluster : {} ".format(input_cluster_id))
return []
return hosts_in_cluster
def get_input_cluster_id(cluster_no_dict, cluster_dict):
input_cluster_id = ""
while input_cluster_id == "":
try:
input_cluster_id = MenuInput('Select custer: ',
allow_empty_input=False, case_insensitive=False).get_input()
logger.info("The entered cluster id is: {}".format(input_cluster_id))
if input_cluster_id not in cluster_no_dict:
print_text("Please enter a valid cluster id from below list : \n")
for id_no, cluster_id in cluster_no_dict.items():
print_text("cluster_id : {} cluster : {} ".format(id_no, cluster_dict[cluster_id]))
input_cluster_id = ""
except ValueError:
print("Invalid input. Please enter appropriate entry")
for id_no, cluster_id in cluster_no_dict.items():
print_text("cluster id : {} cluster name : {} ".format(id_no, cluster_dict[cluster_id]))
continue
except IndexError:
print_text("Please enter a valid cluster id from below list : \n")
for id_no, cluster_id in cluster_no_dict.items():
print_text("cluster id : {} cluster name : {} ".format(id_no, cluster_dict[cluster_id]))
continue
print("\n")
return cluster_no_dict[input_cluster_id]
def get_clusters():
clusters = vcdb.get_clusters()
print_text("\nCompute clusters:")
if clusters == "":
print_text("There are no clusters present ")
return {}
cluster_no = 1
cluster_dict = {}
cluster_no_dict = {}
for cluster_details in clusters:
cluster_id = cluster_details.split("|")[0].strip()
cluster_name = cluster_details.split("|")[1].strip()
cluster_no_dict["{}".format(cluster_no)] = cluster_id
cluster_dict[cluster_id] = cluster_name
print_text(" {}. {}".format(cluster_no, cluster_name))
cluster_no += 1
print_text("\n")
return cluster_no_dict, cluster_dict
def view_esxi_for_all_cluster():
cluster_no_dict, cluster_dict = get_clusters()
if not cluster_dict:
return
input_cluster_id = get_input_cluster_id(cluster_no_dict, cluster_dict)
hosts_list = get_hosts()
host_and_cluster_id_dict = get_host_and_cluster()
hosts_in_cluster = find_hosts_in_cluster(input_cluster_id, hosts_list, host_and_cluster_id_dict)
if hosts_in_cluster:
password = get_host_password("Enter root password for all ESXi hosts in cluster: ")
certificate_management_mode = get_certificate_management_mode()
print_text("Certificate Management Mode = {}{}{}".format(ColorKey.GREEN, certificate_management_mode, ColorKey.NORMAL))
for cluster_id, name, ip in hosts_in_cluster:
print_text("\nHost : {}{}{}".format(ColorKey.CYAN, get_host_name_or_ip(name, ip), ColorKey.NORMAL))
view_esxi_certificate_trust(get_host_name_or_ip(name, ip), certificate_management_mode, password)
def delete_and_validate_vecs_entry(store, alias):
vecs.delete_entry(store, alias)
# This is to check if pem_cert has been deleted
vecs_cert = vecs.get_certificate(store, alias)
if not vecs_cert:
return True
else:
return False
def create_vecs_certificate(access_string, cert_file, vecs_store, vecs_alias):
pem_certs = get_certificate_from_host(access_string, 443)
if pem_certs:
for cert in pem_certs:
# Get and write x509 certificates to TEMP_DIR
save_text_to_file(cert, cert_file)
vecs.add_entry(vecs_store, vecs_alias, cert_file)
updated_vecs_certs = vecs.get_certificate(vecs_store, vecs_alias)
if updated_vecs_certs:
print_text("{:<6}IOFILTER provider certificate created!".format(SPACE))
logger.info("vecs certificate Created")
else:
print_text_warning("Unable to re-create the IOFILTER provider certificate in VECS!")
logger.info("Unable to create vecs certificate")
return True
else:
print_text_warning("Unable to obtain host's {} SSL certificate on port 443!".format(access_string))
logger.info("Unable to obtain host's {} SSL certificate on port 443!".format(access_string))
return False
def display_host_certificate_details(host_hash):
output_format = '%b %e %H:%M:%S %Y GMT'
x509_cert = get_x509_certificate(host_hash)
host_cert_subject, host_cert_issuer = get_subject_and_issuer_dn(x509_cert)
host_cert_valid_start = get_certificate_start_date(x509_cert).strftime(output_format)
host_cert_valid_end = get_certificate_end_date(x509_cert).strftime(output_format)
host_cert_fingerprint = get_certificate_fingerprint(x509_cert, 'sha1')
host_cert_algorithm = x509_cert.get_signature_algorithm().decode('utf-8')
host_cert_san = get_subject_alternative_names(x509_cert)
host_cert_list = []
for certificate in split_certificates_from_pem(host_hash):
x509_certificate = get_x509_certificate(certificate)
host_cert_subject, host_cert_issuer = get_subject_and_issuer_dn(x509_certificate)
host_cert_list.append([host_cert_subject, host_cert_issuer])
print_text("{:<3}Issuer: {}".format(SPACE, host_cert_issuer))
print_text("{:<3}Subject: {}".format(SPACE, host_cert_subject))
print_text("{:<6}Not Before: {}".format(SPACE, host_cert_valid_start))
print_text("{:<6}Not After: {}".format(SPACE, host_cert_valid_end))
print_text("{:<6}SHA1 Fingerprint: {}".format(SPACE, host_cert_fingerprint))
print_text("{:<6}Signature Algorithm: {}".format(SPACE, host_cert_algorithm))
print_text("{:<6}Subject Alternative Name entries:".format(SPACE))
if host_cert_san:
for entry in host_cert_san.split(', '):
print_text("{:<9}|_{}".format(SPACE, entry))
print_text("{:<6}Certificates in the rui.crt file:".format(SPACE))
if host_cert_list:
i=1
for subject, issuer in host_cert_list:
if i > 1:
print_text('{:<9}|'.format(SPACE))
print_text("{:<9}|_Subject={} ".format(SPACE, subject))
print_text("{:<9}|_Issuer={} ".format(SPACE, issuer))
i += 1
return host_cert_fingerprint
def display_host_certificate_for_cert_management_thumbprint(access_string, host_cert_fingerprint):
env = Environment.get_environment()
store = "SMS"
alias = "https:/{}:9080/version.xml".format(access_string)
pem_cert = vecs.get_certificate(store, alias)
current_host_sms_thumbprint = get_certificate_fingerprint(get_x509_certificate(pem_cert))
cert_file = "{}/{}.crt".format(env.get_value('TEMP_DIR'), access_string)
if current_host_sms_thumbprint:
print_text("{:<6}Host IOFILTER provider found in VECS, checking certificate...".format(SPACE))
print(flush=True)
if current_host_sms_thumbprint != host_cert_fingerprint:
print_task_status_warning("Mismatch found, re-creating entry...")
if delete_and_validate_vecs_entry(store, alias):
logger.info("vecs cert deleted. Will recreate.")
create_vecs_certificate(access_string, cert_file, store, alias)
else:
print_task_status_warning("Unable to delete the IOFILTER provider certificate from VECS!")
else:
print_task_status("Certificates match. No need to update")
else:
print_text("{:<3}Host IOFILTER provider certificate not found in VECS. Creating entry...".format(SPACE))
print(flush=True)
logger.info("Attempting to create vecs certificate")
create_vecs_certificate(access_string, cert_file, store, alias)
def url_request(method, url, username=None, password=None, data=None):
# we use plain curl instead of requests library for automatic replay support in CommandRunner
env = Environment.get_environment()
temp_dir = env.get_value('TEMP_DIR')
output_file = "{}/url_request_{}_{}_output.dat".format(temp_dir, method, get_timestamp())
args = ['curl', '-k', '-X', method, url, '-o', output_file, '-s', '-w', '%{http_code}']
if username and password:
args.extend(['-u', "{}:{}".format(username, password)])
if data:
data_file = "{}/url_request_{}_{}_data.dat".format(temp_dir, method, get_timestamp())
save_text_to_file(data, data_file)
args.extend(['--data-binary', '@{}'.format(data_file)])
status_code = CommandRunner(*args).run_and_get_output()
output = get_file_contents(output_file)
return status_code, output
def update_ca_store(access_string, password, vcenter_machine_ssl_cert, sps_cert):
env = Environment.get_environment()
url_text = "https://{}/host/castore".format(access_string)
status_code, content = url_request('GET', url_text, 'root', password)
logger.info("Getting castore.pem file from host {}".format(access_string))
logger.info("Response from host {} - HTTP status code: {}".format(access_string, status_code))
if status_code == '200':
save_text_to_file(content, '{}/{}-castore.pem'.format(env.get_value('TEMP_DIR'), access_string))
cert_list = split_certificates_from_pem(content)
print_task("{:<6}vCenter Machine SSL cert: ".format(SPACE))
if cert_list:
if check_for_ca_certs(vcenter_machine_ssl_cert, cert_list):
print_task_status("Trusted by host")
else:
print_task_status("Not trusted by host")
print_task("{:<6}SPS service connection: ".format(SPACE))
if check_for_ca_certs(sps_cert, cert_list):
print_task_status("Trusted by host")
else:
print_task_status("Not trusted by host (maybe)")
else:
print_task_status_warning(" No CA certs in /etc/vmware/ssl/castore.pem")
print("\n")
return True
elif status_code == '401':
print_task("{:<6}vCenter Machine SSL cert:".format(SPACE))
print_task_status_warning("unknown (possible bad ESXi root password)")
print_task("{:<6}SPS service connection: ".format(SPACE))
print_task_status_warning("unknown (possible bad ESXi root password)")
else:
print_task("{:<6}vCenter Machine SSL cert:".format(SPACE))
print_task_status_warning(" unknown")
print_task("{:<6}SPS service connection: ".format(SPACE))
print_task_status_warning("unknown")
print("\n")
return False
def view_esxi_reverse_proxy_cert_status(host_rhttpproxy_cert, host_iofilterrvp_cert):
input_search_cert_list = find_files("/etc/vmware-vpx/docRoot/certs/*")
search_cert_list = []
for cert_file in input_search_cert_list:
pattern = re.compile(r".*\.r[0-9]*")
if not pattern.match(cert_file):
search_cert_list.append(cert_file)
print_task("{:<6}Reverse Proxy cert (port 443): ".format(SPACE))
certificate_list = []
for search_cert in search_cert_list:
certificate_list.append(get_file_contents(search_cert))
if check_for_ca_certs(host_rhttpproxy_cert, certificate_list):
print_task_status("Trusted by vCenter")
else:
print_task_status("Not trusted by vCenter")
print_task("{:<6}IOFilter VASA provider cert (port 9080): ".format(SPACE))
if host_iofilterrvp_cert:
certificate_list = []
for search_cert in search_cert_list:
certificate_list.append(get_file_contents(search_cert))
if check_for_ca_certs(host_iofilterrvp_cert, certificate_list):
print_task_status("Trusted by vCenter")
else:
print_task_status("Not trusted by vCenter")
else:
print_task_status_warning("unknown")
def view_esxi_certificate_trust(access_string, certificate_management_mode, password):
host_hash = get_certificate_from_host(access_string, 443)
if host_hash:
host_cert_fingerprint = display_host_certificate_details(host_hash)
print_text("\n{:<3}Certificate Trusts:".format(SPACE))
if certificate_management_mode == "thumbprint":
display_host_certificate_for_cert_management_thumbprint(access_string, host_cert_fingerprint)
else:
host_rhttpproxy_cert = get_certificate_from_host(access_string, 443)
host_iofilterrvp_cert = get_certificate_from_host(access_string, 9080)
vcenter_machine_ssl_cert = vecs.get_certificate("MACHINE_SSL_CERT", "__MACHINE_CERT")
sps_cert = vecs.get_certificate("SMS", "sms_self_signed")
view_esxi_reverse_proxy_cert_status(host_rhttpproxy_cert, host_iofilterrvp_cert)
update_ca_store(access_string, password, vcenter_machine_ssl_cert, sps_cert)
else:
print_task_status_warning("Unable to obtain SSL certificate from host {}".format(access_string))
def check_for_ca_certs(cert, search_cert_list):
subject_keyids, fetcher = get_certificate_fetcher_from_list(search_cert_list)
cert_path = build_certification_path(cert, subject_keyids, fetcher)
if cert_path:
return True
return False
def check_esxi_certificate_against_vcdb():
# Entry point for checking esxi vcenter certificate trust against VCDB
num_hosts = vcdb.get_number_of_hosts()
title = "Checking ESXi SSL Thumbprints Against vCenter Database"
text_color = ColorKey.YELLOW if num_hosts == 0 else ColorKey.GREEN
sub_title = "There are {}{}{} hosts connected to vCenter.\n".format(text_color, num_hosts, ColorKey.NORMAL)
menu = Menu()
if not get_hosts():
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, is_hidden=False, key='R',
is_default=True)
else:
menu.set_menu_options(title, sub_title, input_text='Select an option [Return to the previous menu]: ')
menu.add_menu_item("Perform check on all hosts", check_certificate_for_all_esxi_hosts_against_vcdb)
menu.add_menu_item("Perform check on all hosts in a cluster", check_clustered_esxi_against_vcdb)
menu.add_menu_item("Perform check on single host", check_specific_esxi_certificate_against_vcdb)
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, is_hidden=True, key='R',
is_default=True)
menu.run()
def check_certificate_for_all_esxi_hosts_against_vcdb():
hosts = get_hosts()
if hosts:
certificate_management_mode = get_certificate_management_mode()
access_strings = []
for cluster_id, name, ip in hosts:
access_strings.append(get_host_name_or_ip(name, ip))
check_esxi_against_vcdb(access_strings, certificate_management_mode)
def check_clustered_esxi_against_vcdb():
cluster_no_dict, cluster_dict = get_clusters()
input_cluster_id = get_input_cluster_id(cluster_no_dict, cluster_dict)
hosts_list = get_hosts()
host_and_cluster_id_dict = get_host_and_cluster()
hosts_in_cluster = find_hosts_in_cluster(input_cluster_id, hosts_list, host_and_cluster_id_dict)
if hosts_in_cluster:
certificate_management_mode = get_certificate_management_mode()
access_strings = []
for cluster_id, name, ip in hosts_in_cluster:
access_strings.append(get_host_name_or_ip(name, ip))
check_esxi_against_vcdb(access_strings, certificate_management_mode)
def check_specific_esxi_certificate_against_vcdb():
certificate_management_mode = get_certificate_management_mode()
host_name, host_ip = get_host_details(get_hosts())
host_name_or_ip = get_host_name_or_ip(host_name, host_ip)
check_esxi_against_vcdb([host_name_or_ip], certificate_management_mode)
def check_esxi_against_vcdb(access_strings, certificate_management_mode):
if certificate_management_mode == "thumbprint":
print_task_status_warning("The Certificate Management mode has been set to thumbprint")
else:
for access_string in access_strings:
print_text("\n{}{}{}".format(ColorKey.CYAN, access_string, ColorKey.NORMAL))
view_esxi_against_vcdb(access_string)
def get_ssl_thumbprint(access_string):
ssl_thumbprint_str = vcdb.get_ssl_thumbprint_from_vcdb(access_string)
if ssl_thumbprint_str == "":
print_text_warning("No ssl thumbprints available")
return ['', '']
ssl_thumbprint = []
for thumbprint_info in ssl_thumbprint_str:
expected_ssl_thumbprint = thumbprint_info.split("|")[0].strip()
host_ssl_thumbprint = thumbprint_info.split("|")[1].strip()
ssl_thumbprint = [expected_ssl_thumbprint, host_ssl_thumbprint]
return ssl_thumbprint
def view_esxi_against_vcdb(access_string):
expected_ssl_thumbprint, host_ssl_thumbprint = get_ssl_thumbprint(access_string)
pem_certs = get_certificate_from_host(access_string, 443)
certs = ""
if pem_certs:
certs = split_certificates_from_pem(pem_certs)
actual_ssl_thumbprint = get_certificate_fingerprint(get_x509_certificate(certs[0]))
if expected_ssl_thumbprint:
print_text("{:<3}Expected Thumbprint (VCDB) : {}".format(SPACE, expected_ssl_thumbprint))
else:
print_task_status_warning("Expected Thumbprint missing")
if host_ssl_thumbprint:
print_text("{:<3}Host SSL Thumbprint (VCDB) : {}".format(SPACE, host_ssl_thumbprint))
else:
print_task_status_warning("Host SSL Thumbprint missing")
if actual_ssl_thumbprint:
print_text("{:<3}Actual Thumbprint (openssl) : {} ".format(SPACE, actual_ssl_thumbprint))
if expected_ssl_thumbprint != actual_ssl_thumbprint and host_ssl_thumbprint != actual_ssl_thumbprint \
and expected_ssl_thumbprint != host_ssl_thumbprint:
print_text("{:<3}Status : {}MISMATCH{}".format(SPACE, ColorKey.YELLOW, ColorKey.NORMAL))
else:
print_text("{:<3}Status : {}MATCH{}".format(SPACE, ColorKey.GREEN, ColorKey.NORMAL))
else:
print_text(
"{:<3}Actual Thumbprint (openssl) : {}Cannot connect to host{}".format(SPACE, ColorKey.YELLOW, ColorKey.NORMAL))
print_text("{:<3}Status : {}UNKNOWN{}".format(SPACE, ColorKey.YELLOW, ColorKey.NORMAL))
def replace_esxi_certificate():
"""
Entry point for checking esxi vcenter certificate trust
"""
num_hosts = vcdb.get_number_of_hosts()
title = "Replace ESXi certificate"
text_color = ColorKey.YELLOW if num_hosts == 0 else ColorKey.GREEN
sub_title = "There are {}{}{} hosts connected to vCenter.\n".format(text_color, num_hosts, ColorKey.NORMAL)
menu = Menu()
if not get_hosts():
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, is_hidden=False, key='R',
is_default=True)
else:
menu.set_menu_options(title, sub_title, input_text='Select an option [Return to the previous menu]: ')
menu.add_menu_item("Generate Certificate Signing Request and Private Key",
generate_cert_signing_request_non_custom)
menu.add_menu_item("Generate Certificate Signing Request and Private Key from custom OpenSSL configuration "
"file", generate_cert_signing_request_with_custom_openssl)
menu.add_menu_item("Import CA-signed certificate and key", import_ca_signed_cert)
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, is_hidden=True, key='R',
is_default=True)
menu.run()
def get_esxi_cn_input():
esxi_cn_input = ""
while not esxi_cn_input:
esxi_cn_input = MenuInput("Enter a value for the {}CommonName{} of the certificate: ".format(
ColorKey.CYAN, ColorKey.NORMAL), allow_empty_input=False, case_insensitive=False).get_input()
logger.info("The entered esxi cn input : {}".format(esxi_cn_input))
if not esxi_cn_input:
print_text("Please enter valid input")
continue
print()
return esxi_cn_input
def generate_cert_signing_request_non_custom():
generate_cert_signing_request(True)
def generate_cert_signing_request_with_custom_openssl():
generate_cert_signing_request(False)
def generate_csr_and_private_key(esxi_cn_input, custom_openssl_config=False):
"""
Entry point for CSR generation
:param esxi_cn_input: esxi cn input
:param custom_openssl_config: True if a custom OpenSSL config is used
"""
clear_csr_info()
env = Environment.get_environment()
request_dir = env.get_value('REQUEST_DIR')
timestamp = get_timestamp()
csr_file = "{}/{}-{}.csr".format(request_dir, esxi_cn_input, timestamp)
key_file = "{}/{}-{}.key".format(request_dir, esxi_cn_input, timestamp)
if custom_openssl_config:
logger.info('User has chosen to generate the ESXi SSL private key and CSR from a custom OpenSSL '
'configuration file')
user_input = MenuInput('Enter path to custom OpenSSL configuration file: ', allow_empty_input=False,
case_insensitive=False)
print()
while True:
config_file = user_input.get_input()
if not is_file_exists(config_file):
print_text_error('Error: file not found')
continue
break
else:
logger.info('User has chosen to generate the Esxi SSL private key and CSR')
hostname = env.get_value('HOSTNAME')
config_file = "{}/{}.cfg".format(request_dir, esxi_cn_input)
csr_info = get_csr_info(esxi_cn_input)
san_entries = get_san_entries('ESXi', csr_info, esxi_cn_input)
generate_openssl_config(config_file, csr_info, esxi_cn_input, san_entries)
try:
print_header("Replace ESXi Certificate")
print_task("Generating Certificate Signing Request")
generate_csr(config_file, csr_file, key_file)
# OpenSSL 1.0 on VC 7.x doesn't set the file permission correctly
set_file_mode(key_file, stat.S_IRUSR | stat.S_IWUSR)
except CommandExecutionError as e:
error_message = "Unable to generate Certificate Signing Request and Private Key: {}".format(str(e))
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status("OK")
print()
print_text("Certificate Signing Request generated at {}{}{}".format(ColorKey.CYAN, csr_file, ColorKey.NORMAL))
print_text("Private Key generated at {}{}{}".format(ColorKey.CYAN, key_file, ColorKey.NORMAL))
print("\n")
def generate_cert_signing_request(generate_csr_and_private_key_option):
clear_csr_info()
esxi_cn_input = get_esxi_cn_input()
if generate_csr_and_private_key_option:
generate_csr_and_private_key(esxi_cn_input, False)
else:
generate_csr_and_private_key(esxi_cn_input, True)
def path_to_new_esxi_cert():
file_path = ""
while file_path == "":
file_path = MenuInput("Enter path to new ESXi certificate: ",
allow_empty_input=False, case_insensitive=False).get_input()
logger.info("Entered path to new ESXi certificate: {}".format(file_path))
if file_path == "" or not is_file_exists(file_path):
file_path = ""
print_text("Please enter valid input. File not found, enter path to new ESXi certificate:")
return file_path
def get_valid_new_cert():
cert_file = path_to_new_esxi_cert()
pem_cert, pem_cert_key = detect_and_convert_to_pem(cert_file, allow_input=True)
while is_ca_certificate(get_x509_certificate(pem_cert)):
print_task_status_warning("Provided certificate is designated as a Certificate Authority, and is "
"not an appropriate replacement.")
cert_file = path_to_new_esxi_cert()
pem_cert, pem_cert_key = detect_and_convert_to_pem(cert_file, allow_input=True)
env = Environment.get_environment()
temp_dir = env.get_value('TEMP_DIR')
pem_cert_file_name = "{}/{}-converted.pem".format(temp_dir, os.path.basename(cert_file[:cert_file.rfind('.')]))
save_text_to_file(pem_cert, pem_cert_file_name)
logger.info("Provided new certificate file : {}".format(pem_cert_file_name))
pem_key_file_name = "{}/{}-converted.key".format(temp_dir, os.path.basename(cert_file[:cert_file.rfind('.')]))
logger.info("Provided new key file : {}".format(pem_key_file_name))
if not pem_cert_key:
cert_modulus = get_key_modulus_from_pem_text(pem_cert)
pem_cert_key = find_matched_private_key(cert_modulus, cert_file)
if not pem_cert_key:
raise OperationFailed('Failed to obtain key file')
save_text_to_file(pem_cert_key, pem_key_file_name)
logger.info("Provided new certificate key file : {}".format(pem_key_file_name))
return pem_cert_file_name, pem_key_file_name, cert_file, pem_cert, pem_cert_key
def import_ca_signed_cert():
logger.info("User has chosen to import a CA-signed ESXi SSL certificate and key")
host_name, host_ip = get_host_details(get_hosts())
host_name_or_ip = get_host_name_or_ip(host_name, host_ip)
password = get_host_password("Enter root password for ESXi host {}: ".format(host_name_or_ip))
pem_cert_file_name, pem_key_file_name, cert_file, pem_cert, pem_cert_key = get_valid_new_cert()
verify_certificate_and_key(pem_cert_file_name, pem_key_file_name, pem_cert_file_name, None)
ca_pem, ca_certs_pem, cert_pem_updated = obtain_ca_chain(pem_cert)
env = Environment.get_environment()
temp_dir = env.get_value('TEMP_DIR')
pem_ca_file_name = "{}/{}-ca_file.pem".format(temp_dir, os.path.basename(cert_file[:cert_file.rfind('.')]))
if cert_pem_updated is not None:
cert_pem = cert_pem_updated
save_text_to_file(cert_pem, pem_cert_file_name)
save_text_to_file(ca_pem, pem_ca_file_name)
print_header("Replace ESXi Certificate")
print_task('Publish CA signing certificates')
vmdir.publish_trusted_certificate(pem_ca_file_name)
try:
vecs.force_refresh()
print_task_status('OK')
except CommandExecutionError as e:
error_message = "Unable to perform a force-refresh of CA certificates to VECS: {}".format(str(e))
logger.error(error_message)
raise OperationFailed(error_message)
input_search_cert_list = find_files("/etc/vmware-vpx/docRoot/certs/*")
search_cert_list = []
pattern = re.compile(r".*\.r[0-9]*")
for cert_file in input_search_cert_list:
if not pattern.match(cert_file):
search_cert_list.append(cert_file)
env = Environment.get_environment()
temp_dir = env.get_value('TEMP_DIR')
overwrite_file = '{}/{}-castore.pem'.format(temp_dir, host_name_or_ip)
if find_files(overwrite_file):
save_text_to_file("", overwrite_file)
certs_pem, aliases = vecs.get_all_ca_certificates()
cert_text = ""
for cert_pem in certs_pem:
if is_ca_certificate(get_x509_certificate(cert_pem)):
cert_text += cert_pem
cert_text += '\n'
cert_text += vecs.get_certificate('SMS', 'sms_self_signed')
save_text_to_file(cert_text, overwrite_file)
print_task("Replace ESXi certificate")
logger.info("Replacing ESXi certificate with {}".format(overwrite_file))
url_text = "https://{}/host/ssl_cert".format(host_name_or_ip)
status_code, _ = url_request('PUT', url_text, 'root', password, data=pem_cert)
if status_code == '200':
print_task_status("OK")
else:
error_message = "Unable to replace certificate, HTTP return code: {}".format(status_code)
print_task_status_error(error_message)
logger.error(error_message)
raise OperationFailed(error_message)
print_task("Replace ESXi private key")
logger.info("Replacing ESXi private key with {}".format(pem_key_file_name))
url_text = "https://{}/host/ssl_key".format(host_name_or_ip)
status_code, _ = url_request('PUT', url_text, 'root', password, data=pem_cert_key)
if status_code == '200':
print_task_status("OK")
else:
error_message = "Unable to replace private key, HTTP return code: {}".format(status_code)
print_task_status_error(error_message)
logger.error(error_message)
raise OperationFailed(error_message)
print_task("Replace castore.pem")
logger.info("Replacing ESXi private key with {}".format(pem_key_file_name))
url_text = "https://{}/host/castore".format(host_name_or_ip)
status_code, _ = url_request('PUT', url_text, 'root', password, data=cert_text)
if status_code == '200':
print_task_status("OK")
else:
error_message = "Unable to replace castore.pem, HTTP return code: {}\n".format(status_code)
print_task_status_error(error_message)
logger.error(error_message)
raise OperationFailed(error_message)
addition_steps_text = """\nAdditional steps are necessary to complete this process:
1. Run the following command on the ESXi host to save
the new certificate and key to the bootbank: /bin/auto-backup.sh
2. Either reboot the ESXi host, or restart the Management Agents (rhttpproxy, hostd, vpxa, etc.)
3. Disconnect and Re-connect the host in vCenter to update certificate information in the vCenter database
"""
print_text_warning(addition_steps_text)
@@ -0,0 +1,432 @@
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import datetime
import os
import pathlib
import re
from OpenSSL.crypto import load_crl, dump_crl, FILETYPE_PEM, FILETYPE_TEXT
from lib import certificate_utils as certutil
from lib import vcdb
from lib import vmdir
from lib import vecs
from lib.certificate_utils import get_x509_certificate
from lib.constants import (
VCERT_DESC, VMCAM_CERT_FILE_PATH, RBD_CERT_FILE_PATH, RHTTPPROXY_CONFIG_FILE_PATH,
VMCA_CERT_FILE_PATH, REPORT_FILE_PATH
)
from lib.environment import Environment
from lib.exceptions import CommandExecutionError
from lib.console import init_env_colormap, print_text_warning, print_text_error
from lib.host_utils import get_file_contents
from lib.services import is_service_running
from lib.text_utils import TextFilter
from operation.common import get_vcenter_extensions
def write_report(file, text, end='\n', sep=' '):
print(text, file=file, end=end, sep=sep)
print(text, end=end, sep=sep)
def generate_header():
if not is_service_running('vmware-vpostgres'):
print_text_error(
'The vPostgres service is stopped!\n'
'Please ensure this service is running before generating a certificate report.\n'
'Hint: Check the number of CRL entries in VECS')
raise CommandExecutionError('vPostgres is not running')
report_path = pathlib.Path(REPORT_FILE_PATH)
if not os.path.exists(report_path.parent):
# this may necessary for remote execution on a restricted environment
report_file_path = str(pathlib.Path('/tmp').joinpath(report_path.relative_to('/')))
else:
report_file_path = REPORT_FILE_PATH
env = Environment.get_environment()
try:
file = open(report_file_path, 'w')
set_report_file(file)
except IOError as e:
error_message = "Failed to create report file {}: {}".format(REPORT_FILE_PATH, e)
print_text_error(error_message)
raise CommandExecutionError(error_message)
env.set_value('COLORS_DISABLED', True)
init_env_colormap()
write_report(file, '=' * 130)
write_report(file, 'SSL Certificate Report')
write_report(file, VCERT_DESC)
write_report(file, "Host: {}".format(env.get_value('HOSTNAME')))
write_report(file, "Date: {}".format(datetime.datetime.now().ctime()))
write_report(file, "Version: {}".format(env.get_value('VC_VERSION')))
write_report(file, "Build: {}".format(env.get_value('VC_BUILD')))
write_report(file, "Machine ID: {}".format(env.get_value('MACHINE_ID')))
write_report(file, "PNID: {}".format(env.get_value('PNID')))
write_report(file, "Certificate Management Mode: {}".format(vcdb.get_certificate_management_mode()))
write_report(file, '=' * 130)
def report_certificate_details(file, pem_cert, options=None):
x509_cert = certutil.get_x509_certificate(pem_cert)
sign_alg = x509_cert.get_signature_algorithm().decode('utf-8')
subject_kid = certutil.get_subject_keyid(x509_cert)
extensions = certutil.get_certificate_extensions(x509_cert)
report_certificate_basic(file, x509_cert)
write_report(file, " Signature Algorithm: {}".format(sign_alg))
write_report(file, " Subject Key Identifier: {}".format(subject_kid))
write_report(file, ' Authority Key Identifier:')
auth_kid_list = get_authority_keyid_list(x509_cert)
if auth_kid_list:
for auth_key in auth_kid_list:
write_report(file, " |_{}".format(auth_key))
else:
akid = certutil.get_authority_keyid(x509_cert)
if akid:
write_report(file, " |_keyid:{}".format(akid))
write_report(file, ' Key Usage:')
key_usage = extensions.get('keyUsage')
if key_usage:
for ku in key_usage.split(', '):
write_report(file, " |_{}".format(ku))
write_report(file, ' Extended Key Usage:')
ext_key_usage = extensions.get('extendedKeyUsage')
if ext_key_usage:
for ku in ext_key_usage.splitlines():
write_report(file, " |_{}".format(ku))
write_report(file, ' Subject Alternative Name entries:')
san = extensions.get('subjectAltName')
if san:
for entry in san.split(', '):
if re.match('^[A-Za-z]+: ', entry):
entry = re.sub('^([A-Za-z]+:) ', '\\1', entry)
write_report(file, " |_{}".format(entry))
write_report(file, ' Other Information:')
write_report(file, " |_Is a Certificate Authority: {}"
.format('Yes' if certutil.is_ca_certificate(x509_cert) else 'No'))
auth_keyid_alt = certutil.get_authority_keyid(x509_cert, allow_alternate=True)
subject_kids_in_vecs = get_subject_keyids_in_vecs()
subject_kids_in_vmdir = get_subject_keyids_in_vmdir()
found_in_vecs = auth_keyid_alt in subject_kids_in_vecs
found_in_vmdir = auth_keyid_alt in subject_kids_in_vmdir
if not found_in_vmdir and not found_in_vecs:
status = "No{}".format(' (Self-Signed)' if certutil.is_self_signed_certificate(x509_cert)
else '')
elif found_in_vmdir and not found_in_vecs:
status = 'Yes, in VMware Directory'
elif not found_in_vmdir and found_in_vecs:
status = 'Yes, in VECS'
else:
status = 'Yes, in both'
write_report(file, " |_Issuing CA in VMware Directory/VECS: {}".format(status))
if options:
if 'checkCurrentMachineSSLUsage' in options:
report_current_machine_ssl_usage(file)
if 'checkCurrentExtensionThumbprints' in options:
report_current_extension_thumbprints(file)
def report_certificate_basic(file, x509_cert):
subject_dn, issuer_dn = certutil.get_subject_and_issuer_dn(x509_cert)
output_format = '%b %e %H:%M:%S %Y GMT'
start_date = certutil.get_certificate_start_date(x509_cert).strftime(output_format)
end_date = certutil.get_certificate_end_date(x509_cert).strftime(output_format)
fingerprint = certutil.get_certificate_fingerprint(x509_cert)
write_report(file, " Issuer: {}".format(issuer_dn))
write_report(file, " Subject: {}".format(subject_dn))
write_report(file, " Not Before: {}".format(start_date))
write_report(file, " Not After : {}".format(end_date))
write_report(file, " SHA1 Fingerprint: {}".format(fingerprint))
def report_current_machine_ssl_usage(file):
machine_cert = certutil.get_certificate_from_host('localhost', 443)
machine_cert_fingerprint = certutil.get_certificate_fingerprint(get_x509_certificate(machine_cert)) \
if machine_cert else ''
write_report(file, " |_Current certificate used by the reverse proxy: {}"
.format(machine_cert_fingerprint))
vpxd_cert = certutil.get_certificate_from_host('localhost', 8089)
vpxd_cert_fingerprint = certutil.get_certificate_fingerprint(get_x509_certificate(vpxd_cert)) \
if vpxd_cert else ''
write_report(file, " |_Current certificate used by vCenter (vpxd) : {}"
.format(vpxd_cert_fingerprint))
def report_current_extension_thumbprints(file):
vcenter_extensions = get_vcenter_extensions()
max_len = len(max(vcenter_extensions, key=len))
write_report(file, ' |_Thumbprints in VCDB for extensions that should use the vpxd-extension certificate')
for extension in vcenter_extensions:
thumbprint = vcdb.get_extension_thumbprint(extension)
write_report(file, " |_{:<{}}: {}".format(extension, max_len, thumbprint))
def report_crl_details(file, pem_text):
crl = load_crl(FILETYPE_PEM, pem_text)
output = dump_crl(FILETYPE_TEXT, crl).decode('utf-8')
issuer = TextFilter(output).contain('Issuer:').cut('Issuer:', [1]).get_text().strip()
if issuer.startswith('/'):
issuer = re.sub('/([A-Z]+=)', ', \\1', issuer[1:])
last_update = TextFilter(output).contain('Last Update:').cut('Update:', [1]).get_text().strip()
next_update = TextFilter(output).contain('Next Update:').cut('Update:', [1]).get_text().strip()
sign_alg = TextFilter(output).contain('Signature Algorithm:').head(1).cut('Algorithm:', [1]).get_text().strip()
write_report(file, " Issuer: {}".format(issuer))
write_report(file, " Last Update: {}".format(last_update))
write_report(file, " Next Update: {}".format(next_update))
write_report(file, " Signature Algorithm: {}".format(sign_alg))
def get_authority_keyid_list(x509_cert):
"""
Get Authority KeyId list from X509 certificate
:param x509_cert: X509Certificate object
:return: list of as keyId:, or DirName:.../serial:
"""
extensions = certutil.get_certificate_extensions(x509_cert)
akid = extensions.get('authorityKeyIdentifier')
if akid:
return TextFilter(akid).match('^keyid:.*|^DirName:.*|^serial:.*').get_lines()
else:
return []
def get_subject_keyids_in_vecs(pem_certs=None):
env = Environment.get_environment()
subject_kids = env.get_value('SUBJECT_KEYIDS_IN_VECS')
if subject_kids is not None:
return subject_kids
if pem_certs is None:
pem_certs, _ = vecs.get_all_ca_certificates()
subject_kids = []
for pem_cert in pem_certs:
x509_cert = get_x509_certificate(pem_cert)
subject_kids.append(certutil.get_subject_keyid(x509_cert, allow_alternate=True))
env.set_value('SUBJECT_KEYIDS_IN_VECS', subject_kids)
return subject_kids
def get_subject_keyids_in_vmdir(pem_certs=None):
env = Environment.get_environment()
subject_kids = env.get_value('SUBJECT_KEYIDS_IN_VMDIR')
if subject_kids is not None:
return subject_kids
if pem_certs is None:
pem_certs, _ = vmdir.get_all_ca_certificates()
subject_kids = []
for pem_cert in pem_certs:
x509_cert = get_x509_certificate(pem_cert)
subject_kids.append(certutil.get_subject_keyid(x509_cert, allow_alternate=True))
env.set_value('SUBJECT_KEYIDS_IN_VMDIR', subject_kids)
return subject_kids
def generate_report_ca_certs_in_vmdir():
file = get_report_file()
write_report(file, 'VMware Directory Certificates')
write_report(file, ' CA Certificates')
pem_certs, aliases = vmdir.get_all_ca_certificates()
get_subject_keyids_in_vmdir(pem_certs)
get_subject_keyids_in_vecs()
for pem_cert, alias in zip(pem_certs, aliases):
write_report(file, " CN(id): {}".format(alias))
report_certificate_details(file, pem_cert)
def generate_report_ca_certs_in_vecs():
file = get_report_file()
write_report(file, 'VECS Certificates')
get_subject_keyids_in_vecs()
get_subject_keyids_in_vmdir()
for store in vecs.get_store_list():
if store == 'APPLMGMT_PASSWORD':
continue
write_report(file, " Store: {}".format(store))
pem_certs, aliases = vecs.get_all_certificates(store)
for pem_text, alias in zip(pem_certs, aliases):
write_report(file, " Alias: {}".format(alias))
if store == 'TRUSTED_ROOT_CRLS':
report_crl_details(file, pem_text)
else:
if store == 'MACHINE_SSL_CERT' and alias == '__MACHINE_CERT':
options = ['checkCurrentMachineSSLUsage']
elif store == 'vpxd-extension' and alias == 'vpxd-extension':
options = ['checkCurrentExtensionThumbprints']
else:
options = None
report_certificate_details(file, pem_text, options)
def generate_report_solution_user_certs_in_vmdir():
file = get_report_file()
write_report(file, ' Service Principal (Solution User) Certificates')
solution_users = vmdir.get_solution_users()
machine_id = Environment.get_environment().get_value('MACHINE_ID')
for solution_user in solution_users:
write_report(file, " Service Principal: {}-{}".format(solution_user, machine_id))
pem_cert = vmdir.get_solution_user_certificate(solution_user)
report_certificate_details(file, pem_cert)
def generate_report_sts_tenant_certs():
file = get_report_file()
write_report(file, ' Single Sign-On Secure Token Service Certificates')
tenant_certs = vmdir.get_sts_tenant_certificates()
for tenant in tenant_certs.keys():
certs = tenant_certs[tenant]
for cert in certs:
x509_cert = get_x509_certificate(cert)
cert_type = 'CA Certificate' if certutil.is_ca_certificate(x509_cert) else 'Signing Certificate'
write_report(file, " {} {}".format(tenant, cert_type))
report_certificate_details(file, cert)
def generate_report_certs_in_filesystem():
file = get_report_file()
write_report(file, 'Filesystem Certificates')
write_report(file, ' VMCA Certificate')
write_report(file, " Certificate: {}".format(VMCAM_CERT_FILE_PATH))
pem_cert = get_file_contents(VMCA_CERT_FILE_PATH)
report_certificate_details(file, pem_cert)
write_report(file, ' Authentication Proxy Certificate')
write_report(file, " Certificate: {}".format(VMCAM_CERT_FILE_PATH))
pem_cert = get_file_contents(VMCAM_CERT_FILE_PATH)
report_certificate_details(file, pem_cert)
write_report(file, ' Auto Deploy CA Certificate')
write_report(file, " Certificate: {}".format(RBD_CERT_FILE_PATH))
pem_cert = get_file_contents(RBD_CERT_FILE_PATH)
report_certificate_details(file, pem_cert)
config = get_file_contents(RHTTPPROXY_CONFIG_FILE_PATH)
cert_file = TextFilter(config).match('.*<clientCAListFile>.*</clientCAListFile>.*')\
.set_options(invert_match=True).contain('<!--').get_text().strip()
if cert_file:
write_report(file, ' Smart Card Whitelist Certificates')
cert_file = TextFilter(config).cut('clientCAListFile', [1])[1:-2].strip()
file_content = get_file_contents(cert_file)
certs = TextFilter(file_content).match_block(
'-----BEGIN CERTIFICATE-----', '-----END CERTIFICATE-----',
True).get_lines()
for index, cert in enumerate(certs, 1):
write_report(file, " Certificate {}".format(index))
report_certificate_details(file, cert)
def generate_report_ca_certs_sca():
file = get_report_file()
certs = vmdir.get_smart_card_issuing_ca_certs()
if certs:
write_report(file, ' Smart Card Issuing CA Certificates')
for index, cert in enumerate(certs, 1):
write_report(file, " Smart Card Issuing CA {}".format(index))
report_certificate_details(file, certutil.build_pem_certificate(cert))
def generate_report_ca_certs_ad_ldaps():
file = get_report_file()
identity_sources = vmdir.get_identity_sources()
certs = []
for source in identity_sources:
if source['type'] == 'AD over LDAP':
certs.extend(source['certificates'])
if certs:
write_report(file, ' AD Over LDAPS Domain Controller Certificates')
for index, cert in enumerate(certs, 1):
write_report(file, " Certificate {}".format(index))
report_certificate_details(file, certutil.build_pem_certificate(cert))
def generate_report_ssl_trust_anchors():
file = get_report_file()
write_report(file, 'Lookup Service Registration Trust Anchors')
endpoints = vmdir.get_all_lookup_service_endpoints()
trust_anchors = get_ssl_trust_anchors(endpoints)
for index, cert in enumerate(trust_anchors, 1):
write_report(file, " Endpoint Certificate {}".format(index))
service_ids, uris = get_service_ids_and_uri_by_certificate(endpoints, cert)
report_trust_anchor_details(file, certutil.build_pem_certificate(cert), service_ids, uris)
def get_ssl_trust_anchors(endpoints):
trust_anchors = []
for endpoint in endpoints:
endpoint['certificates'] = []
for cert in sum([endpoint['vmwLKUPEndpointSslTrust'], endpoint['vmwLKUPSslTrustAnchor']], []):
pem_cert = certutil.build_pem_certificate(cert)
endpoint['certificates'].append(pem_cert)
if pem_cert not in trust_anchors:
trust_anchors.append(pem_cert)
return trust_anchors
def get_service_ids_and_uri_by_certificate(endpoints, cert):
service_ids = []
uris = []
for endpoint in endpoints:
if cert in endpoint['certificates']:
service_id = endpoint['dn'].split(',')[1].replace('cn=', '')
if service_id not in service_ids:
service_ids.append(service_id)
uri = endpoint['vmwLKUPURI']
if uri not in uris:
uris.append(endpoint['vmwLKUPURI'])
return sorted(service_ids), sorted(uris)
def report_trust_anchor_details(file, pem_cert, service_ids, uris):
x509_cert = get_x509_certificate(pem_cert)
report_certificate_basic(file, x509_cert)
write_report(file, ' Service IDs:')
for service_id in service_ids:
service_type = vmdir.get_endpoint_service_type(service_id)
write_report(file, " |_{} ({})".format(service_id, service_type))
write_report(file, ' Endpoints:')
for uri in uris:
write_report(file, " |_{}".format(uri))
def set_report_file(file):
env = Environment.get_environment()
env.set_value('REPORT_FILE_DESC', file)
def get_report_file():
env = Environment.get_environment()
return env.get_value('REPORT_FILE_DESC')
def finalize_report():
file = get_report_file()
if file is not None:
file.close()
set_report_file(None)
env = Environment.get_environment()
env.set_value('COLORS_DISABLED', False)
init_env_colormap()
print()
print_text_warning("Certificate report is available at {}".format(REPORT_FILE_PATH))
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,129 @@
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import logging
import stat
from lib.command_runner import CommandRunner
from lib.console import (
print_task, print_task_status, print_task_status_warning, print_header, ColorKey, print_text
)
from lib.constants import STS_SERVER_CONFIG_FILE_PATH, STS_SERVER_CONFIG_PROPERTY_FILE_PATH
from lib.environment import Environment
from lib.exceptions import OperationFailed, CommandExecutionError
from lib.host_utils import VcVersion, get_vc_version, set_file_mode, save_text_to_file
from lib.ldap_utils import LdapException
from lib.menu import MenuInput
from lib.vecs import get_certificate, get_key, delete_store, create_store, grant_vecs_permission
from lib.vmdir import perform_ldap_modify
from operation.check_configuration import check_sts_certificate_configuration, check_sts_connectionstring_configuration
from operation.restart_service import restart_vmware_services
logger = logging.getLogger(__name__)
def manage_sts_configuration():
print_header('Checking STS Server Configuration')
need_to_update_configuration, sts_config = check_sts_certificate_configuration()
if not need_to_update_configuration:
print_text('\nThe STS server is using the {}MACHINE_SSL_CERT{} VECS store.\n'.format(ColorKey.GREEN, ColorKey.NORMAL))
else:
manage_sts_certificate_configuration(sts_config)
need_to_update_connectionstrings, connectionstring_dn = check_sts_connectionstring_configuration()
if not need_to_update_connectionstrings:
manage_sts_connectionstrings(connectionstring_dn)
def manage_sts_certificate_configuration(sts_config):
env = Environment.get_environment()
vc_version = get_vc_version()
# Note: The descriptionMap must be kept in sync with the possible keys in
# the versionConfigMap in check_sts_certificate_configuration().
descriptionMap = {
'localhost_connector_store': 'Server > Service > Connector (localhost port)',
'localhost_certificate_store': 'Server > Service > Connector (localhost port) > SSLHostConfig > Certificate',
'certificate_store': 'Server > Service > Connector > SSLHostConfig > Certificate',
'clientauth_connector_store': 'Server > Service > Connector (client auth port)',
'clientauth_certificate_store': 'Server > Service > Connector (client auth port) > SSLHostConfig > Certificate',
}
print_text('\nThe STS server is using the following VECS stores:')
for key, value in sts_config.items():
if key == 'vecs_stores_to_replace':
continue
description = descriptionMap[key]
print_text('{}: {}{}{}'.format(description, ColorKey.YELLOW, value, ColorKey.NORMAL))
vecs_stores_to_replace = sts_config['vecs_stores_to_replace']
user_input = MenuInput('\nUpdate STS server configuration to use the {}MACHINE_SSL_CERT{} store? [N]: '.format(ColorKey.GREEN, ColorKey.NORMAL),
acceptable_inputs=['Y', 'N'], default_input='N')
if user_input.get_input() == 'N':
print()
return
backup_dir = env.get_value('BACKUP_DIR')
print_header('Updating STS server configuration')
print_text('Backing up configuration')
if vc_version >= VcVersion.V9:
sts_config_file = STS_SERVER_CONFIG_PROPERTY_FILE_PATH
sts_config_backup = '{}.sts-server.properties'.format(backup_dir)
else:
sts_config_file = STS_SERVER_CONFIG_FILE_PATH
sts_config_backup = '{}/sts-server.xml'.format(backup_dir)
try:
CommandRunner('cp', sts_config_file, sts_config_backup).run_and_get_output()
set_file_mode(sts_config_backup, stat.S_IRUSR | stat.S_IWUSR)
except CommandExecutionError:
print_task_status_warning('FAILED')
error_message = 'Unable to backup STS server configuration'
logger.error(error_message)
raise OperationFailed(error_message)
for store in vecs_stores_to_replace:
print_task('Store {}: Backing up certificate'.format(store))
sts_backup_cert = get_certificate(store, '__MACHINE_CERT')
save_text_to_file(sts_backup_cert, '{}/sts-{}.crt'.format(backup_dir, store))
print_task_status('OK')
print_task('Store {}: Backing up key'.format(store))
sts_backup_key = get_key(store, '__MACHINE_CERT')
save_text_to_file(sts_backup_key, '{}/sts-{}.key'.format(backup_dir, store))
print_task_status('OK')
print_task('Changing STS server configuration')
try:
CommandRunner('sed', '-i', 's/{}/MACHINE_SSL_CERT/g'.format(store), sts_config_file).run()
except:
print_task_status_warning('FAILED')
error_message = 'Unable to change STS server configuration'
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
if store == 'STS_INTERNAL_SSL_CERT':
print_task('Remove legacy STS VECS store')
delete_store('STS_INTERNAL_SSL_CERT')
print_task_status('OK')
restart_vmware_services('vmware-stsd')
def manage_sts_connectionstrings(search_dn):
user_input = MenuInput('\nUpdate STS ConnectionStrings value to {}ldap://localhost:389{}? [N]: '.format(ColorKey.GREEN, ColorKey.NORMAL),
acceptable_inputs=['Y', 'N'], default_input='N')
if user_input.get_input() == 'N':
print()
return
print_header('Update STS ConnectionStrings')
print_task('Change vmwSTSConnectionStrings value')
try:
perform_ldap_modify(search_dn, 'vmwSTSConnectionStrings', 'ldap://localhost:389')
print_task_status('OK')
except LdapException:
error_message = "Failed updating vmwSTSConnectionStrings to 'ldap://localhost:389'"
logger.error(error_message)
raise OperationFailed(error_message)
@@ -0,0 +1,290 @@
# Copyright (c) 2024-2025 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import logging
import os
import datetime
import stat
from lib import vmdir
from lib.certificate_utils import is_cert_file_expired, CERTOOL_CLI
from lib.command_runner import CommandRunner
from lib.console import (
print_task, print_task_status, print_text_error, print_task_status_error, print_task_status_warning,
print_header, print_text, set_text_color, ColorKey
)
from lib.constants import (VMCA_CERT_FILE_PATH, VMCA_KEY_FILE_PATH, VMCA_SSO_FILE_PATH)
from lib.environment import Environment
from lib.exceptions import OperationFailed, CommandExecutionError
from lib.host_utils import is_file_exists
from lib.menu import Menu, MenuInput
from lib.vmdir import update_solution_user_certificate_in_vmdir, replace_service_principal_certificates
from operation.manage_certificate import (get_csr_info, update_vc_ext_thumbprints,
update_auto_deploy_db, restart_vmware_services,
backup_vecs_cert_key, update_vecs,
generate_csr_and_private_key, verify_certificate_and_key,
publish_ca_signing_certificate,
replace_machine_ssl_certificate_with_vmca_signed,
update_ssl_trust_anchors, clear_csr_info, get_san_entries,
get_timestamp, get_custom_ca_signed_certificate_files,
generate_openssl_config, is_file_exists, generate_csr,
set_file_mode, get_cert_usage_value,
verify_certification_path,
set_default_csr_input,
replace_sts_signing_certificate_with_vmca_signed,
generate_certool_config, backup_filesystem_cert_key,
confirm_replace_sts_signing_certificate)
logger = logging.getLogger(__name__)
def reset_all_certificates():
"""
Entry point for reset certificates with VMCA-signed
"""
if is_cert_file_expired(VMCA_CERT_FILE_PATH):
print_text_error('The VMCA certificate is expired and will first need to be replaced.')
menu = Menu()
menu.set_menu_options('Replace VMCA Certificate', run_once=True)
menu.add_menu_item('Replace VMCA with self-signed certificate', replace_vmca_cert, is_default=True)
menu.add_menu_item('Replace VMCA with CA-signed certificate', import_custom_ca_signed_vmca_certificate)
menu.run()
# to get initial csr info input wrt vCert.sh
get_csr_info()
machine_ssl_cert_file = replace_machine_ssl_certificate_with_vmca_signed()
replace_solution_user_certificates_vmca_signed()
# update the rest settings
update_vc_ext_thumbprints()
update_auto_deploy_db()
if confirm_replace_sts_signing_certificate():
replace_sts_signing_certificate_with_vmca_signed()
update_ssl_trust_anchors(machine_ssl_cert_file)
#todo: need to add additional methods in reset after closing current MR.
# restart services
restart_vmware_services()
def manage_solution_user_certificate_with_vmca_signed():
replace_solution_user_certificates_vmca_signed()
update_vc_ext_thumbprints()
# restart services
restart_vmware_services()
clear_csr_info()
def replace_solution_user_certificates_vmca_signed():
env = Environment.get_environment()
output_dir = env.get_value('TEMP_DIR')
solution_users = env.get_value('SOLUTION_USERS')
print_header('Replace Solution User Certificates')
set_default_csr_input()
vmdir.verify_service_principals()
print('Generate new certificates and keys:')
for soluser in solution_users:
print_task(f" {soluser}")
privkey_arg = "--privkey={}/{}.key".format(output_dir, soluser)
pubkey_arg = "--pubkey={}/{}.pub".format(output_dir, soluser)
cert_arg = "--cert={}/{}.crt".format(output_dir, soluser)
client = CommandRunner(CERTOOL_CLI, '--genkey', privkey_arg, pubkey_arg)
return_code, stdout, stderr = client.run()
if return_code != 0:
print_text_error(f"Unable to generate a key pair for {soluser}")
args = [CERTOOL_CLI, '--gencert', '--server',
"localhost", '--Name', soluser, '--genCIScert',
privkey_arg, cert_arg, '--config=/dev/null', '--Country',
env.get_value('CSR_COUNTRY_DEFAULT'), '--State',
env.get_value('CSR_STATE_DEFAULT'), '--Locality',
env.get_value('CSR_LOCALITY_DEFAULT'), '--Organization',
env.get_value('CSR_ORG_DEFAULT'),
'--OrgUnit',
"mID-{}".format(env.get_value('MACHINE_ID'))]
if soluser in ['wcp', 'wcpsvc']:
args.extend('--dataencipherment')
else:
args.extend(['--FQDN', env.get_value('PNID')])
client = CommandRunner(*args)
return_code, stdout, stderr = client.run()
if return_code != 0:
print_text_error(f"Unable to generate a VMCA-signed cert for {soluser}")
else:
print_task_status("OK")
print('\nBackup certificate and private key:')
for soluser in solution_users:
backup_vecs_cert_key(soluser)
print('\nUpdating certificates and keys in VECS:')
for soluser in solution_users:
cert_file = "{}/{}.crt".format(output_dir, soluser)
key_file = "{}/{}.key".format(output_dir, soluser)
update_vecs(soluser, cert_file, key_file)
if is_file_exists('/storage/vsan-health/vpxd-extension.cert') and is_file_exists('/storage/vsan-health/vpxd-extension.key'):
print('\nUpdating vpxd-extension certificate for vSAN Health')
try:
CommandRunner('cp', '{}/vpxd-extension.crt'.format(output_dir), '/storage/vsan-health/vpxd-extension.cert').run()
CommandRunner('cp', '{}/vpxd-extension.key'.format(output_dir), '/storage/vsan-health/vpxd-extension.key').run()
except CommandExecutionError as e:
error_message = "Unable to update vpxd-extension cert and key for vSAN Health: {}".format(str(e))
logger.error(error_message)
raise OperationFailed(error_message)
print('\nUpdating solution user certificates in VMware Directory:')
for soluser in solution_users:
cert_file = "{}/{}.crt".format(output_dir, soluser)
replace_service_principal_certificates(soluser, cert_file)
def generate_csr_and_private_key_of_soluser(custom_openssl_config=False):
"""
Entry point for generating csr and private key of solution users
"""
env = Environment.get_environment()
request_dir = env.get_value('REQUEST_DIR')
timestamp = get_timestamp()
solution_users = env.get_value('SOLUTION_USERS')
hostname = env.get_value('HOSTNAME')
if custom_openssl_config:
#san_entries = get_san_entries('soluser', get_csr_info(), hostname)
for soluser in solution_users:
csr_file = "{}/{}-{}.csr".format(request_dir, soluser, timestamp)
key_file = "{}/{}-{}.key".format(request_dir, soluser, timestamp)
logger.info('User has chosen to generate the {} private key and CSR from a custom OpenSSL configuration file'.format(soluser))
user_input = MenuInput('Enter path to custom OpenSSL configuration file for {}{}{}: '
.format('{COLORS[CYAN]}', soluser, '{COLORS[NORMAL]}'),
allow_empty_input=False, case_insensitive=False)
config_file = "{}/{}-{}.cfg".format(request_dir, soluser, timestamp)
while True:
config_file = user_input.get_input()
if not is_file_exists(config_file):
print_text_error('Error: file not found, enter path to custom OpenSSL configuration file for {}: '.format(soluser))
continue
break
try:
generate_csr(config_file, csr_file, key_file)
# OpenSSL 1.0 on VC 7.x doesn't set the file permission correctly
set_file_mode(key_file, stat.S_IRUSR | stat.S_IWUSR)
except CommandExecutionError as e:
error_message = "Unable to generate Certificate Signing Request and Private Key: {}".format(str(e))
logger.error(error_message)
raise OperationFailed(error_message)
else:
csr_info = get_csr_info('soluser')
san_entries = get_san_entries('soluser', csr_info, hostname)
for soluser in solution_users:
config_file = "{}/{}-{}.cfg".format(request_dir, soluser, timestamp)
csr_file = "{}/{}-{}.csr".format(request_dir, soluser, timestamp)
key_file = "{}/{}-{}.key".format(request_dir, soluser, timestamp)
cn = "{}-{}".format(soluser, env.get_value('MACHINE_ID'))
generate_openssl_config(config_file, csr_info, cn, san_entries)
try:
generate_csr(config_file, csr_file, key_file)
# OpenSSL 1.0 on VC 7.x doesn't set the file permission correctly
set_file_mode(key_file, stat.S_IRUSR | stat.S_IWUSR)
except CommandExecutionError as e:
error_message = "Unable to generate Certificate Signing Request and Private Key: {}".format(str(e))
logger.error(error_message)
raise OperationFailed(error_message)
print_text("Certificate Signing Request generated at:")
for soluser in solution_users:
csr_file = "{}/{}-{}.csr".format(request_dir, soluser, timestamp)
set_text_color(ColorKey.CYAN)
print_text("{}".format(csr_file))
set_text_color(ColorKey.NORMAL)
print_text("Private Keys generated at:")
for soluser in solution_users:
key_file = "{}/{}-{}.key".format(request_dir, soluser, timestamp)
set_text_color(ColorKey.CYAN)
print_text("{}".format(key_file))
set_text_color(ColorKey.NORMAL)
def import_custom_ca_signed_vmca_certificate():
cert_pem_file, key_pem_file, ca_pem_file, cert_pem, ca_pem = get_custom_ca_signed_certificate_files('vmca', False)
backup_filesystem_cert_key(VMCA_CERT_FILE_PATH, VMCA_KEY_FILE_PATH, 'VMCA')
reconfigure_vmca(cert_pem_file, key_pem_file)
publish_ca_signing_certificate(ca_pem_file)
update_vmca_cert_in_filesystem(VMCA_CERT_FILE_PATH)
return True
def import_custom_ca_signed_vmca_certificate_and_reset_all():
if import_custom_ca_signed_vmca_certificate():
reset_all_certificates()
def replace_vmca_cert():
logger.info("User selected to replace VMCA certificate with a SELF-SIGNED certificate")
logger.info("Certificates will NOT be regenerated")
env = Environment.get_environment()
csr_info = get_csr_info()
vmca_cn_default = env.get_value('VMCA_CN_DEFAULT')
output_dir = env.get_value('TEMP_DIR')
vmca_cn_input = MenuInput('Enter a value for the {}CommonName{} of the certificate [{}]: '
.format('{COLORS[CYAN]}', '{COLORS[NORMAL]}', vmca_cn_default),
default_input=vmca_cn_default).get_input()
print_header('Replace VMCA Certificate')
generate_certool_config(output_dir, 'vmca', csr_info, vmca_cn_input)
print_task('Generate VMCA certificate')
config_arg = "--config={}/vmca.cfg".format(output_dir)
outcert_arg = "--outcert={}/vmca.crt".format(output_dir)
outprivkey_arg = "--outprivkey={}/vmca.key".format(output_dir)
try:
CommandRunner(CERTOOL_CLI, '--genselfcacert', outcert_arg, outprivkey_arg, config_arg, expected_return_code=0).run_and_get_output()
print_task_status('OK')
except CommandExecutionError as e:
print_task_status('FAILED')
error_message = "Unable to generate new VMCA certificate"
logger.error(error_message)
raise OperationFailed(error_message)
backup_filesystem_cert_key(VMCA_CERT_FILE_PATH, VMCA_KEY_FILE_PATH, 'VMCA')
reconfigure_vmca("{}/vmca.crt".format(output_dir), "{}/vmca.key".format(output_dir) )
update_vmca_cert_in_filesystem(VMCA_CERT_FILE_PATH)
return True
def update_vmca_cert_in_filesystem(vmca_cert):
if is_file_exists(VMCA_SSO_FILE_PATH):
print_task('Update VMCA certificate on filesystem')
try:
CommandRunner('mv', VMCA_SSO_FILE_PATH, '/etc/vmware-sso/keys/ssoserverRoot.crt.old', expected_return_code=0).run_and_get_output()
except CommandExecutionError as e:
print_task_status_warning('FAILED')
error_message = 'Unable to backup old SSO server root certificate'
logger.error(error_message)
raise OperationFailed(error_message)
try:
CommandRunner('cp', vmca_cert, VMCA_SSO_FILE_PATH, expected_return_code=0).run_and_get_output()
except CommandExecutionError as e:
print_task_status_warning('FAILED')
error_message = 'Unable to update SSO server root certificate'
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
def reconfigure_vmca(new_vmca_cert, new_vmca_key):
print_task('Reconfigure VMCA')
cert_arg = '--cert={}'.format(new_vmca_cert)
privkey_arg = '--privkey={}'.format(new_vmca_key)
try:
CommandRunner(CERTOOL_CLI, '--rootca', cert_arg, privkey_arg, expected_return_code=0).run_and_get_output()
except CommandExecutionError as e:
print_task_status_warning('FAILED')
error_message = 'Unable to reconfigure the VMCA with the new certificate'
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
def replace_vmca_cert_and_regenerate_all():
if replace_vmca_cert():
reset_all_certificates()
@@ -0,0 +1,87 @@
# Copyright (c) 2024 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import logging
from lib.exceptions import OperationFailed, CommandExecutionError, MenuExitException
from lib.menu import Menu, MenuInput
from lib.services import (
list_vmware_services, start_vmware_services, stop_vmware_services
)
from lib.console import (
print_task, print_task_status, print_task_status_warning, print_header, print_text_error
)
logger = logging.getLogger(__name__)
def restart_vmware_services(services=None):
"""
Restart VMware services
"""
print()
acceptable_inputs = ['Y', 'N']
services = [services] if type(services) == str else services
if services is None:
user_input = MenuInput('Restart VMware services [N]: ', acceptable_inputs=acceptable_inputs,
default_input='N')
else:
user_input = MenuInput("Restart service(s) {}{}{} [N]: "
.format('{COLORS[CYAN]}', ', '.join(services), '{COLORS[NORMAL]}'),
acceptable_inputs=acceptable_inputs, default_input='N')
if user_input.get_input() == 'N':
return
print_header('Restarting Services')
if services is None:
print_task('Stopping VMware services')
if not stop_vmware_services():
print_task_status_warning('FAILED')
error_message = 'Unable to stop all VMware services, check log for details'
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
print_task('Starting VMware services')
if not start_vmware_services():
print_task_status_warning('FAILED')
error_message = 'Unable to start all VMware services, check log for details'
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
return
for service in services:
print_task("Stopping {}".format(service))
if not stop_vmware_services(service):
print_task_status_warning('FAILED')
error_message = "Unable to stop service {}, check log for details".format(service)
print_text_error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
print_task("Starting {}".format(service))
if not start_vmware_services(service):
print_task_status_warning('FAILED')
error_message = "Unable to start service {}, check log for details".format(service)
logger.error(error_message)
raise OperationFailed(error_message)
print_task_status('OK')
def restart_specific_vmware_service():
vmware_services = list_vmware_services()
print()
print('VMware services:')
print(*vmware_services)
service_to_restart = input('\nEnter VMware service to restart from the list above: ')
while not service_to_restart:
service_to_restart = input('\nEnter VMware service to restart: ')
if service_to_restart in vmware_services:
restart_vmware_services(service_to_restart)
else:
print_text_error(f'Error: Unknown service: {service_to_restart}')
@@ -0,0 +1,332 @@
# Copyright (c) 2024 Broadcom. All Rights Reserved.
# The term "Broadcom" refers to Broadcom Inc.
# and/or its subsidiaries.
import logging
import OpenSSL
from datetime import datetime
from lib import vcdb
from lib import vecs
from lib import vmdir
from lib.environment import Environment
from lib.certificate_utils import (
get_certificate_info, get_certificate_fetcher_from_list, get_certificate_extensions,
build_certification_path, get_x509_certificate, get_subject_and_issuer_dn, get_subject_keyid,
is_ca_certificate, split_certificates_from_pem
)
from lib.console import print_header, print_text, print_text_warning, print_text_error, ColorKey
from lib.input import MenuInput
from lib.host_utils import get_file_contents
from lib.constants import VMCA_CERT_FILE_PATH
from lib.menu import Menu, MenuInput
from lib.vmdir import get_smart_card_issuing_ca_certs
from operation.check_certificate import get_ids_domain_and_certificates, get_ids_domain_and_certificates_by_domain
from operation.common import get_vcenter_extensions, get_vcenter_extension_expected_thumbprints
logger = logging.getLogger(__name__)
def view_certificate_dummy(**kwargs):
print_text("{}=== Unsupported view certificate operation! ==={}".format(ColorKey.YELLOW,
ColorKey.NORMAL))
def view_certificate(pem_cert):
print_header('Certificate Information')
print(get_certificate_info(pem_cert))
ca_certs, _ = vecs.get_all_certificates('TRUSTED_ROOTS')
subject_keyids, fetcher = get_certificate_fetcher_from_list(ca_certs)
cert_path = build_certification_path(pem_cert, subject_keyids, fetcher)
print_header('Certification Path')
print_certification_path(cert_path)
def print_certification_path(cert_path, status_text=False):
for idx, cert_info in enumerate(cert_path):
plus_green = "{}+{}".format(ColorKey.GREEN, ColorKey.NORMAL)
if idx == 0:
untrusted = '[UNTRUSTED]' if cert_info['is_trusted'] is not True else ''
incomplete = '[INCOMPLETE]' if cert_info['is_selfsigned'] is not True else ''
if untrusted or incomplete:
mark = "{}!{}".format(ColorKey.RED, ColorKey.NORMAL)
else:
mark = plus_green
if status_text:
print_text("[ {} ] {} {}{}".format(mark, cert_info['cert_name'], incomplete, untrusted))
else:
print_text("[ {} ] {}".format(mark, cert_info['cert_name']))
else:
print_text("{}|_[ {} ] {}".format(' ' * (4 * idx - 2), plus_green,
cert_info['cert_name']))
def view_machine_ssl_certificate():
cert = vecs.get_certificate('MACHINE_SSL_CERT', '__MACHINE_CERT')
view_certificate(cert)
def view_solution_user_certificate():
env = Environment.get_environment()
for soluser in env.get_value('SOLUTION_USERS'):
print(f"\nSolution User: {soluser}")
cert = vecs.get_certificate(soluser, soluser)
view_certificate(cert)
def view_data_encipherment_certificate():
cert = vecs.get_certificate('data-encipherment', 'data-encipherment')
view_certificate(cert)
def view_ca_certificates_in_vmdir(show_list_only=False):
print_header('CA Certificates in VMware Directory')
skids = vmdir.get_all_ca_subject_keyids(use_cache=False)
for index, subject_key_id in enumerate(skids):
cert_pem = vmdir.get_ca_certificate(subject_key_id)
brief = get_certificate_info_brief(cert_pem)
if index != 0:
print()
print("{:>2}. {}".format(index + 1, brief))
if show_list_only:
return
keys = [str(i) for i in range(1, len(skids) + 1)]
keys.append('R')
print()
menu_input = MenuInput('Select certificate [Return to menu]: ', acceptable_inputs=keys,
default_input='R')
key = menu_input.get_input()
print()
if key != 'R':
cert = vmdir.get_ca_certificate(skids[int(key)-1])
view_certificate(cert)
def view_certificates_in_vecs(store, show_list_only=False):
certs, aliases = vecs.get_all_certificates(store)
for idx, cert in enumerate(certs):
brief = get_certificate_info_brief(cert, aliases[idx])
print("{:>2}. {}\n".format(idx + 1, brief))
if show_list_only:
return
keys = [str(i) for i in range(1, len(aliases) + 1)]
keys.append('R')
menu_input = MenuInput('Select certificate [Return to menu]: ', acceptable_inputs=keys,
default_input='R')
key = menu_input.get_input()
print()
if key != 'R':
cert = certs[int(key) - 1]
view_certificate(cert)
def view_ca_certificates_in_vecs(show_list_only=False):
view_certificates_in_vecs('TRUSTED_ROOTS', show_list_only)
def view_sms_certificates_in_vecs(show_list_only=False):
view_certificates_in_vecs('SMS', show_list_only)
def view_sts_signing_certificates():
"""
Entry point for STS Tenant certificates view
"""
certs_map = vmdir.get_sts_tenant_certificates(include_tenant_credential=True,
include_certificate_chain=False)
for tenant_idx, tenant in enumerate(certs_map.keys()):
if tenant_idx != 0:
print()
print("{}".format(tenant))
for cert_idx, pem_cert in enumerate(certs_map[tenant]):
if cert_idx != 0:
print()
basic_constraints = get_certificate_extensions(
get_x509_certificate(pem_cert)).get('basicConstraints')
is_ca = basic_constraints is not None and 'CA:TRUE' in basic_constraints
print(" Certificate Type: {} Certificate".format('CA' if is_ca else 'Signing'))
brief = get_certificate_info_brief(pem_cert, indent_first_line=True)
print(brief)
def get_certificate_info_brief(pem_cert, alias=None, computed_skid=None, indent_first_line=False,
domain_name=None, identity_source_type=None):
"""
indent_first_line is used for 4 space indentation for first line of certificate info,
in this case, it is alias or subject.
"""
cert_info = []
try:
x509_cert = get_x509_certificate(pem_cert)
skid = get_subject_keyid(x509_cert)
subject_dn, issuer_dn = get_subject_and_issuer_dn(x509_cert)
# '20240708142450Z' -> 'Jul 8 14:24:50 2024 GMT'
input_format = '%Y%m%d%H%M%SZ'
output_format = '%b %e %H:%M:%S %Y GMT'
end_date = datetime.strptime(x509_cert.get_notAfter().decode('utf-8'), input_format).strftime(output_format)
if skid is None and computed_skid is not None:
skid = "{} (computed)".format(computed_skid)
is_ca = is_ca_certificate(x509_cert)
indentation = ' ' if indent_first_line else ''
if alias:
cert_info.append("{}Alias: {}".format(indentation, alias))
cert_info.append(" Subject: {}".format(subject_dn))
else:
cert_info.append("{}Subject: {}".format(indentation, subject_dn))
cert_info.append(" Issuer: {}".format(issuer_dn))
cert_info.append(" End Date: {}".format(end_date))
cert_info.append(" Subject Key ID: {}".format(skid))
cert_info.append(" Is CA Cert: {}".format('Yes' if is_ca else 'No'))
if domain_name is not None:
cert_info.append(' Domain: {}'.format(domain_name))
if identity_source_type is not None:
cert_info.append(' Identity Source Type: {}'.format(identity_source_type))
except OpenSSL.crypto.Error:
cert_info.append(print_text_error('Invalid format, certificate cannot be parsed', end=''))
cert_info.append(" Subject Key ID: {}".format(computed_skid))
return '\n'.join(cert_info)
def view_ldaps_identity_source_certificates(show_domain='All', identity_source_type=None, show_list_only=False):
header = 'LDAP Certificates' if show_domain == 'All' else 'LDAP Certificates ({})'.format(show_domain)
print_header(header)
identity_sources = vmdir.get_identity_sources()
index_start = 0
identity_source_certs = ['-']
if show_domain == 'All':
# OpenLDAP
index_start, identity_source_certs = ldaps_identity_source_certificate_items(identity_sources, index_start,
identity_source_certs, identity_source_type='OpenLDAP')
# AD over LDAP
index_start, identity_source_certs = ldaps_identity_source_certificate_items(identity_sources, index_start,
identity_source_certs, identity_source_type='AD over LDAP')
# ADFS
index_start, identity_source_certs = ldaps_identity_source_certificate_items(identity_sources, index_start,
identity_source_certs, identity_source_type='ADFS')
else:
index_start, identity_source_certs = ldaps_identity_source_certificate_items(identity_sources, index_start,
identity_source_certs, domain_name=show_domain,
identity_source_type=identity_source_type)
if show_list_only:
return identity_source_certs
print()
keys = [str(i) for i in range(1, len(identity_source_certs) + 1)]
keys.append('R')
menu_input = MenuInput('Select certificate [Return to menu]: ', acceptable_inputs=keys,
default_input='R')
key = menu_input.get_input()
print()
if key != 'R':
cert = identity_source_certs[int(key)]
view_certificate(cert)
return identity_source_certs
def ldaps_identity_source_certificate_items(identity_sources, index_start, identity_source_certs,
domain_name=None, identity_source_type=None):
if domain_name is None:
domain_and_certs = get_ids_domain_and_certificates(identity_sources, identity_source_type)
else:
domain_and_certs = get_ids_domain_and_certificates_by_domain(identity_sources, identity_source_type, domain_name)
if domain_and_certs:
for domain_name, certificates in domain_and_certs:
for index, cert in enumerate(certificates, index_start):
identity_source_certs.append(cert)
cert_info = get_certificate_info_brief(cert, domain_name=domain_name, identity_source_type=identity_source_type)
if index != 0:
print()
print("{:>2}. {}".format(index + 1, cert_info))
index_start += 1
return index_start, identity_source_certs
def view_VMCA_Certificate():
"""
Entry point to view VMCA Certificate
"""
view_certificate(get_file_contents(VMCA_CERT_FILE_PATH))
def view_vcenter_extension_thumbprints():
vcenter_extensions = get_vcenter_extensions()
extension_thumbprints = vcdb.get_extension_thumbprints(vcenter_extensions)
expected_thumbprints = get_vcenter_extension_expected_thumbprints(vcenter_extensions)
for extension, (thumbprint, db_cert_pem) in extension_thumbprints.items():
_, expected_cert_type, _ = expected_thumbprints[extension]
print("{} ({})".format(extension, expected_cert_type))
print(" ", thumbprint)
def view_smart_card_certificates(show_list_only=False):
smart_card_filter_file_certificates = view_smart_card_filter_file_certificates()
smart_card_vmdir_certificates = view_smart_card_vmdir_certificates()
if show_list_only:
return smart_card_filter_file_certificates, smart_card_vmdir_certificates
menu = Menu()
menu.set_menu_options('View Smart Card Certificate Options')
menu.add_menu_item('View Smart Card filter file certificate')
menu.add_menu_item('View Smart Card CA certificate')
menu.add_menu_item('Return to the previous menu', Menu.run_navigation_return, key='R', is_default=True, is_hidden=True,
use_label_as_key=True)
menu.show_menu()
print()
view_certificate_type = menu.get_input().strip()
if view_certificate_type != 'R':
if view_certificate_type == '1':
keys = [str(i) for i in range(1, len(smart_card_filter_file_certificates) + 1)]
keys.append('R')
print()
view_certificate_selection = MenuInput('Select Smart Card filter file certificate [Return to menu]: ', acceptable_inputs=keys, default_input='R').get_input()
if view_certificate_selection != 'R':
view_certificate(smart_card_filter_file_certificates[int(view_certificate_selection) - 1])
else:
keys = [str(i) for i in range(1, len(smart_card_vmdir_certificates) + 1)]
keys.append('R')
print()
view_certificate_selection = MenuInput('Select Smart Card CA certificate [Return to menu]: ', acceptable_inputs=keys, default_input='R').get_input()
if view_certificate_selection != 'R':
view_certificate(smart_card_vmdir_certificates[int(view_certificate_selection) - 1])
def view_smart_card_filter_file_certificates():
env = Environment.get_environment()
smart_card_filter_file = env.get_value('SMART_CARD_FILTER_FILE')
print_header('Smart Card Filter File Certificates')
smart_card_ca_certificates = split_certificates_from_pem(get_file_contents(smart_card_filter_file))
if smart_card_ca_certificates:
for index, cert in enumerate(smart_card_ca_certificates):
cert_info = get_certificate_info_brief(cert)
if index != 0:
print()
print("{:>2}. {}".format(index + 1, cert_info))
return smart_card_ca_certificates
else:
print_text_warning('No certificates found in {}'.format(smart_card_filter_file))
return []
def view_smart_card_vmdir_certificates():
smart_card_ca_certificates = get_smart_card_issuing_ca_certs()
logger.info('Smart Card CA Certifiates from VMware Directory: {}'.format(smart_card_ca_certificates))
print_header('Smart Card Issuing CA Certificates')
if smart_card_ca_certificates:
for index, cert in enumerate(smart_card_ca_certificates):
cert_info = get_certificate_info_brief(cert)
if index != 0:
print()
print("{:>2}. {}".format(index + 1, cert_info))
return smart_card_ca_certificates
else:
print_text_warning('No certificates found in VMware Directory')
return []