Your Name
982 lines
44 KiB
Python
982 lines
44 KiB
Python
# 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
|