Function add_initial_zgrab_certificates

python3_cron_scripts/create_cert_graphs.py:297–399

Get the list of current certificates from zgrab scans for the specified zones. Append any new entries to the provided array of current_certs. This currently does not check

(
    mongo_connector, zgrab_collection, zone, current_certs
)

def add_initial_zgrab_certificates(
    mongo_connector, zgrab_collection, zone, current_certs
):
    """
    Get the list of current certificates from zgrab scans for the specified zones.
    Append any new entries to the provided array of current_certs.
    This currently does not check
    """

    results = mongo_connector.perform_find(
        zgrab_collection,
        {
            "$or": [
                {
                    "data.http.redirect_response_chain.0.request.tls_handshake.server_certificates.certificate.parsed.subject.common_name": {
                        "$regex": r"^(.+\.)*" + zone + "$"
                    }
                },
                {
                    "data.http.redirect_response_chain.0.request.tls_handshake.server_certificates.certificate.parsed.extensions.subject_alt_name.dns_names": {
                        "$regex": r"^(.+\.)*" + zone + "$"
                    }
                },
            ]
        },
        filter={"data.http.redirect_response_chain": 1},
    )

    for result in results:
        i = next(
            (
                index
                for (index, item) in enumerate(current_certs)
                if item["id"]
                == result["data"]["http"]["redirect_response_chain"][0]["request"][
                    "tls_handshake"
                ]["server_certificates"]["certificate"]["parsed"]["fingerprint_sha256"]
            ),
            None,
        )
        if i is None:
            item = {
                "id": result["data"]["http"]["redirect_response_chain"][0]["request"][
                    "tls_handshake"
                ]["server_certificates"]["certificate"]["parsed"]["fingerprint_sha256"]
            }
            dns_list = []
            try:
                for dns_name in result["data"]["http"]["redirect_response_chain"][0][
                    "request"
                ]["tls_handshake"]["server_certificates"]["certificate"]["parsed"][
                    "subject"
                ][
                    "common_name"
                ]:
                    if dns_name not in dns_list:
                        dns_list.append(dns_name)
            except KeyError: