Function add_initial_zgrab2_certificates

python3_cron_scripts/create_cert_graphs.py:514–638

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_zgrab2_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.result.redirect_response_chain.0.request.tls_log.handshake_log.server_certificates.certificate.parsed.subject.common_name": {
                        "$regex": r"^(.+\.)*" + zone + "$"
                    }
                },
                {
                    "data.http.result.redirect_response_chain.0.request.tls_log.handshake_log.server_certificates.certificate.parsed.extensions.subject_alt_name.dns_names": {
                        "$regex": r"^(.+\.)*" + zone + "$"
                    }
                },
            ]
        },
        filter={"data.http.result.redirect_response_chain": 1},
    )

    for result in results:
        i = next(
            (
                index
                for (index, item) in enumerate(current_certs)
                if item["id"]
                == result["data"]["http"]["result"]["redirect_response_chain"][0][
                    "request"
                ]["tls_log"]["handshake_log"]["server_certificates"]["certificate"][
                    "parsed"
                ][
                    "fingerprint_sha256"
                ]
            ),
            None,
        )
        if i is None:
            item = {
                "id": result["data"]["http"]["result"]["redirect_response_chain"][0][
                    "request"
                ]["tls_log"]["handshake_log"]["server_certificates"]["certificate"][
                    "parsed"
                ][
                    "fingerprint_sha256"
                ]
            }
            dns_list = []
            try:
                for dns_name in result["data"]["http"]["result"][
                    "redirect_response_chain"