CertificatesTroubleshoot

Source

import demistomock as demisto  # noqa: F401
from CommonServerPython import *  # noqa: F401
import os
import socket
import ssl
from pathlib import Path
from typing import Any
import subprocess
import re
import pem

# mypy: ignore-errors
from cryptography import x509
from cryptography.x509.extensions import ExtensionNotFound
from cryptography.hazmat.backends import default_backend


def parse_certificate_object_identifier_name(certificate: x509.Name, oid: x509.ObjectIdentifier) -> list[str] | None:
    """Get attribute from decoded certificate.

    Args:
        certificate: Certificate as x509.Certificate .
        oid: Enum value from x509.NameOID .

    Returns:
        list: Decoded values.
    """
    attributes = [attr.value for attr in certificate.get_attributes_for_oid(oid)]

    return attributes if attributes else None


def parse_certificate_object_identifier_extentions(certificate: x509.Extensions, oid: x509.ObjectIdentifier) -> list[str] | None:
    """Get attribute from decoded certificate extension.

    Args:
        certificate: Certificate as x509.Certificate .
        oid: Enum value from x509.ExtentionOID .

    Returns:
        list: Decoded values.
    """
    try:
        values = certificate.get_extension_for_oid(oid).value
        attributes = [item.value for item in values]  # type: ignore[attr-defined]
    except ExtensionNotFound:
        attributes = []

    return attributes if attributes else None


def certificate_to_ec(certificate_name: x509.Name) -> dict:
    """Translate abstrcat object x509.Name to entry context.

    Args:
        certificate_name: Issuer or subject.

    Returns:
        dict: Corresponding enrty context.
    """
    return {
        # Contact details
        "EmailAddress": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.EMAIL_ADDRESS),
        "SurName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.SURNAME),
        # Location related
        "CountryName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.COUNTRY_NAME),
        "StateOrProvinceName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.STATE_OR_PROVINCE_NAME),
        "LocalityName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.LOCALITY_NAME),
        "JurisdictionCountryName": parse_certificate_object_identifier_name(
            certificate_name, x509.NameOID.JURISDICTION_COUNTRY_NAME
        ),
        "JurisdictionLocalityName": parse_certificate_object_identifier_name(
            certificate_name, x509.NameOID.JURISDICTION_LOCALITY_NAME
        ),
        "JurisdictionStateOrProvinceName": parse_certificate_object_identifier_name(
            certificate_name, x509.NameOID.JURISDICTION_STATE_OR_PROVINCE_NAME
        ),
        "PostalAddress": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.POSTAL_ADDRESS),
        "PostalCode": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.POSTAL_CODE),
        "StreetAddress": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.STREET_ADDRESS),
        # Domain / URL
        "DomainNameQualifier": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.DN_QUALIFIER),
        "DomainComponent": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.DOMAIN_COMPONENT),
        "GenerationQualifier": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.GENERATION_QUALIFIER),
        "GivenName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.GIVEN_NAME),
        "CommonName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.COMMON_NAME),
        # Business
        "BusinessCategory": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.BUSINESS_CATEGORY),
        "OrganizationName": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.ORGANIZATION_NAME),
        "OrganizationalUnitName": parse_certificate_object_identifier_name(
            certificate_name, x509.NameOID.ORGANIZATIONAL_UNIT_NAME
        ),
        # General
        "Title": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.TITLE),
        "SerialNumber": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.SERIAL_NUMBER),
        "Pseudonym": parse_certificate_object_identifier_name(certificate_name, x509.NameOID.PSEUDONYM),
    }


def certificate_extentions_to_ec(certificate_ext: x509.Extensions) -> dict:
    """Translate abstrcat object x509.Name to entry context.

    Args:
        certificate_ext: Certificate extension.

    Returns:
        dict: Corresponding enrty context.
    """
    return {
        "IssuerAlternativeName": parse_certificate_object_identifier_extentions(
            certificate_ext, x509.ExtensionOID.ISSUER_ALTERNATIVE_NAME
        ),
        "SubjectAlternativeName": parse_certificate_object_identifier_extentions(
            certificate_ext, x509.ExtensionOID.SUBJECT_ALTERNATIVE_NAME
        ),
    }


def parse_certificate(certificate: str) -> dict:
    """Decode certificate from

    Args:
        certificate: certificate as string.

    Returns:
        dict: Corresponding enrty context.
    """
    decode_certificate = x509.load_pem_x509_certificate(certificate.encode(), default_backend())
    return {
        "Raw": certificate,
        "Decode": {
            "Issuer": certificate_to_ec(decode_certificate.issuer),
            "Subject": certificate_to_ec(decode_certificate.subject),
            "Extentions": certificate_extentions_to_ec(decode_certificate.extensions),
            "NotValidBefore": str(decode_certificate.not_valid_before),
            "NotValidAfter": str(decode_certificate.not_valid_after),
            "Version": decode_certificate.version.value,
        },
    }


def parse_all_certificates(certifcates: str) -> list[dict[Any, Any]]:
    """Parse all certificates in a given string.

    Args:
        certifcates: certificates as a single string.

    Returns:
        list: Corresponding enrty context.
    """
    return [parse_certificate(cert.as_text()) for cert in pem.parse(certifcates.encode())]


def docker_container_details() -> dict:
    """Gather docker container SSL/TLS Certificate information (Which set by demisto engine), The following details:
            1. Global veriables which used by requests module:
                a. SSL_CERT_FILE
                b. REQUESTS_CA_BUNDLE
            2. Custom python ssl file located in docker container - SSL_CERT_FILE

    Returns:
        dict: Corresponding enrty context.
    """
    ssl_cert_file_env = os.environ.get("SSL_CERT_FILE")
    custom_certs = []
    if ssl_cert_file_env:
        container_ca_file = Path(ssl_cert_file_env)
        certificates = "" if not container_ca_file.is_file() else container_ca_file.read_text()
        custom_certs = parse_all_certificates(certificates)
        demisto.debug(f"custom certs len: {len(custom_certs)}")
    return {
        "ShellVariables": {
            "SSL_CERT_FILE": os.environ.get("SSL_CERT_FILE"),
            "CERT_FILE": os.environ.get("REQUESTS_CA_BUNDLE"),
        },
        "CustomCertificateAuthorities": custom_certs[:5],
    }


def get_certificates(endpoint: str, port: str) -> str:
    """Download certificate from remote server.

    Args:
        endpoint: url to get certificate from.
        port: endpoint port.

    Returns:
        str: certificate string containing the certs in PEM format.
    """
    mode = demisto.getArg("mode") or "python"
    if mode == "openssl":
        return get_certificate_openssl(endpoint, port)
    else:
        hostname = endpoint
        conn = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
        sock = context.wrap_socket(conn, server_hostname=hostname)
        sock.settimeout(60)  # make sure we don't get stuck
        sock.connect((hostname, int(port)))
        peer_cert = sock.getpeercert(True)
        if peer_cert:
            return ssl.DER_cert_to_PEM_cert(peer_cert)
        else:
            return ""


def get_certificate_openssl(endpoint, port):
    try:
        debug_args = ["-msg", "-debug"] if is_debug_mode() else []
        demisto.debug("before opening openssl process")
        process = subprocess.Popen(
            ["openssl", "s_client", *debug_args, "-connect", f"{endpoint}:{port}", "-showcerts"],
            text=True,
            stderr=subprocess.STDOUT,
            stdout=subprocess.PIPE,
            stdin=subprocess.PIPE,
        )
        demisto.debug("opened process")
        openssl_res, _ = process.communicate("Q\n", timeout=60)
        demisto.debug(f"openssl process return code {process.returncode}")
        demisto.debug(f"openssl output: {openssl_res}")
        return "\n".join(re.findall(r"^-----BEGIN CERT.*?^-----END CERTIFICATE-----", openssl_res, re.DOTALL | re.MULTILINE))
    except subprocess.TimeoutExpired as e:
        process.kill()  # Terminate the process
        openssl_res, _ = process.communicate()  # Capture any remaining output
        demisto.error(f"openssl command timed out after 60 seconds {e}")
        demisto.error(f"Partial openssl output: {openssl_res}")
        return_error("openssl command timed out, see logs for more details.")


def endpoint_certificate(endpoint: str, port: str) -> dict:
    """Get certificate issuer from endpoint.

    Args:
        endpoint: Enpoint url:port, if no port will be 443 by default.
        port: endpoint port.

    Returns:
        dict: Corresponding enrty context.
    """
    certificates = get_certificates(endpoint, port)
    demisto.debug(f"endpoint: {endpoint} port: {port} certs: {certificates}")
    return {"Identifier": endpoint, "Certificates": parse_all_certificates(certificates)}


def build_human_readable(entry_context: dict) -> str:
    human_readable = ""
    entry_context = entry_context.get("TroubleShoot", {})
    # Engine docker container
    engine: dict = dict_safe_get(entry_context, ["Engine", "SSL/TLS"], {}, dict)
    human_readable += "## Docker container engine - custom certificate\n"
    engine_cer_general = [dict_safe_get(item, ["Decode"]) for item in engine.get("CustomCertificateAuthorities", {})]
    engine_cer_issuer = [dict_safe_get(item, ("Decode", "Issuer")) for item in engine.get("CustomCertificateAuthorities", {})]
    engine_cer_subjects = [dict_safe_get(item, ("Decode", "Subject")) for item in engine.get("CustomCertificateAuthorities", {})]
    engine_cer_extentions = [
        dict_safe_get(item, ("Decode", "Extentions")) for item in engine.get("CustomCertificateAuthorities", {})
    ]
    engine_vars = engine.get("ShellVariables")
    human_readable += tableToMarkdown(name="Enviorment variables", t=engine_vars)
    human_readable += (
        tableToMarkdown(name="General", t=engine_cer_general, headers=["NotValidBefore", "NotValidAfter", "Version"])
        if engine_cer_extentions
        else ""
    )
    human_readable += tableToMarkdown(name="Issuer", t=engine_cer_issuer, removeNull=True) if engine_cer_issuer else ""
    human_readable += tableToMarkdown(name="Subject", t=engine_cer_subjects, removeNull=True) if engine_cer_subjects else ""
    human_readable += (
        tableToMarkdown(name="Extentions", t=engine_cer_extentions, removeNull=True) if engine_cer_extentions else ""
    )
    # Endpoint
    endpoint: dict = entry_context.get("Endpoint", {}).get("SSL/TLS", {})
    endpoint_cer_general = [dict_safe_get(item, ["Decode"]) for item in endpoint.get("Certificates", {})]
    endpoint_cer_issuer = [dict_safe_get(item, ("Decode", "Issuer")) for item in endpoint.get("Certificates", {})]
    endpoint_cer_subject = [dict_safe_get(item, ("Decode", "Subject")) for item in endpoint.get("Certificates", {})]
    endpoint_cer_extentions = [dict_safe_get(item, ("Decode", "Extentions")) for item in endpoint.get("Certificates", {})]
    human_readable += f"\n\n## Endpoint certificate - {endpoint.get('Identifier')}\n"
    human_readable += tableToMarkdown(
        name="General", t=endpoint_cer_general, headers=["NotValidBefore", "NotValidAfter", "Version"]
    )
    human_readable += tableToMarkdown(name="Issuer", t=endpoint_cer_issuer, removeNull=True)
    human_readable += tableToMarkdown(name="Subject", t=endpoint_cer_subject, removeNull=True)
    human_readable += tableToMarkdown(name="Extentions", t=endpoint_cer_extentions, removeNull=True)
    human_readable += "\n"

    return human_readable


def main():
    try:
        entry_context = {
            "TroubleShoot": {
                "Engine": {
                    "SSL/TLS": docker_container_details(),
                },
                "Endpoint": {
                    "SSL/TLS": endpoint_certificate(demisto.getArg("endpoint"), demisto.getArg("port") or "443"),
                },
            }
        }
        human_readable = build_human_readable(entry_context)

        return_outputs(human_readable, entry_context, {})
    except Exception as e:
        return_error(f"Failed to execute Certificate Troubleshoot.\n Error: {str(e)}")


if __name__ in ("__main__", "__builtin__", "builtins"):
    main()

README

{
    "TroubleShoot": {
        "Engine": {
            "SSL/TLS": {
                "ShellVariables": {
                    "SSL_CERT_FILE": "/etc/custom-python-ssl/certs.pem", 
                    "CERT_FILE": "/etc/custom-python-ssl/certs.pem"
                }, 
                "CustomCertificateAuthorities": [
                    {
                        "Decode": {
                            "Subject": {
                                "OrganizationalUnitName": [
                                    "Content"
                                ], 
                                "OrganizationName": [
                                    "Demisto"
                                ], 
                                "BusinessCategory": null, 
                                "Title": null, 
                                "SerialNumber": null, 
                                "StateOrProvinceName": [
                                    "Hamerkaz"
                                ], 
                                "DomainComponent": null, 
                                "GivenName": null, 
                                "Pseudonym": null, 
                                "JurisdictionStateOrProvinceName": null, 
                                "GenerationQualifier": null, 
                                "LocalityName": [
                                    "Tel Aviv"
                                ], 
                                "SurName": null, 
                                "CommonName": [
                                    "Demisto TLS"
                                ], 
                                "JurisdictionLocalityName": null, 
                                "StreetAddress": null, 
                                "PostalCode": null, 
                                "PostalAddress": null, 
                                "JurisdictionCountryName": null, 
                                "CountryName": [
                                    "IL"
                                ], 
                                "EmailAddress": [
                                    "test@gmail.com""
                                ], 
                                "DomainNameQualifier": null
                            }, 
                            "Issuer": {
                                "OrganizationalUnitName": [
                                    "Content"
                                ], 
                                "OrganizationName": [
                                    "Demisto"
                                ], 
                                "BusinessCategory": null, 
                                "Title": null, 
                                "SerialNumber": null, 
                                "StateOrProvinceName": [
                                    "Hamerkaz"
                                ], 
                                "DomainComponent": null, 
                                "GivenName": null, 
                                "Pseudonym": null, 
                                "JurisdictionStateOrProvinceName": null, 
                                "GenerationQualifier": null, 
                                "LocalityName": [
                                    "Tel Aviv"
                                ], 
                                "SurName": null, 
                                "CommonName": [
                                    "Demisto TLS"
                                ], 
                                "JurisdictionLocalityName": null, 
                                "StreetAddress": null, 
                                "PostalCode": null, 
                                "PostalAddress": null, 
                                "JurisdictionCountryName": null, 
                                "CountryName": [
                                    "IL"
                                ], 
                                "EmailAddress": [
                                    "test@gmail.com""
                                ], 
                                "DomainNameQualifier": null
                            }
                        }, 
                        "Raw": "-----BEGIN CERTIFICATE-----\nxxxxx\n-----END CERTIFICATE-----\n"
                    }
                ]
            }
        }, 
        "Endpoint": {
            "SSL/TLS": {
                "Certificates": [
                    {
                        "Decode": {
                            "Subject": {
                                "OrganizationalUnitName": [
                                    "Test"
                                ], 
                                "OrganizationName": [
                                    "Content"
                                ], 
                                "BusinessCategory": null, 
                                "Title": null, 
                                "SerialNumber": null, 
                                "StateOrProvinceName": [
                                    "Demisto"
                                ], 
                                "DomainComponent": null, 
                                "GivenName": null, 
                                "Pseudonym": null, 
                                "JurisdictionStateOrProvinceName": null, 
                                "GenerationQualifier": null, 
                                "LocalityName": null, 
                                "SurName": null, 
                                "CommonName": [
                                    "test.compute-1.amazonaws.com"
                                ], 
                                "JurisdictionLocalityName": null, 
                                "StreetAddress": null, 
                                "PostalCode": null, 
                                "PostalAddress": null, 
                                "JurisdictionCountryName": null, 
                                "CountryName": [
                                    "IL"
                                ], 
                                "EmailAddress": [
                                    "test@gmail.com""
                                ], 
                                "DomainNameQualifier": null
                            }, 
                            "Issuer": {
                                "OrganizationalUnitName": [
                                    "Content"
                                ], 
                                "OrganizationName": [
                                    "Demisto"
                                ], 
                                "BusinessCategory": null, 
                                "Title": null, 
                                "SerialNumber": null, 
                                "StateOrProvinceName": [
                                    "Hamerkaz"
                                ], 
                                "DomainComponent": null, 
                                "GivenName": null, 
                                "Pseudonym": null, 
                                "JurisdictionStateOrProvinceName": null, 
                                "GenerationQualifier": null, 
                                "LocalityName": [
                                    "Tel Aviv"
                                ], 
                                "SurName": null, 
                                "CommonName": [
                                    "Demisto TLS"
                                ], 
                                "JurisdictionLocalityName": null, 
                                "StreetAddress": null, 
                                "PostalCode": null, 
                                "PostalAddress": null, 
                                "JurisdictionCountryName": null, 
                                "CountryName": [
                                    "IL"
                                ], 
                                "EmailAddress": [
                                    "test@gmail.com"
                                ], 
                                "DomainNameQualifier": null
                            }
                        }, 
                        "Raw": "-----BEGIN CERTIFICATE-----\nxxxx\n-----END CERTIFICATE-----\n"
                    }
                ], 
                "Identifier": "test.compute-1.amazonaws.com",
                "NotValidBefore": "2020-09-22 11:37:45",
                "NotValidAfter": "2025-09-21 11:37:45",
                "Version": 0,
                "Extentions: {
                    "IssuerAlternativeName": [*.google.com, *.appengine.google.com],
                    "SubjectAlternativeName": [*.google.com, *.appengine.google.com]
                }
            }
        }
    }
}