Get the list of current certificates from censys for the specified zones. Append any new entries to the provided array of current_certs.
(censys_collection, zone, current_certs)
| 67 | |
| 68 | |
| 69 | def add_censys_certificates(censys_collection, zone, current_certs): |
| 70 | """ |
| 71 | Get the list of current certificates from censys for the specified zones. |
| 72 | Append any new entries to the provided array of current_certs. |
| 73 | """ |
| 74 | |
| 75 | results = censys_collection.find( |
| 76 | { |
| 77 | "$or": [ |
| 78 | { |
| 79 | "p443.https.tls.certificate.parsed.subject.common_name": { |
| 80 | "$regex": r"^(.+\.)*" + zone + "$" |
| 81 | } |
| 82 | }, |
| 83 | { |
| 84 | "p443.https.tls.certificate.parsed.extensions.subject_alt_name.dns_names": { |
| 85 | "$regex": r"^(.+\.)*" + zone + "$" |
| 86 | } |
| 87 | }, |
| 88 | ] |
| 89 | }, |
| 90 | { |
| 91 | "p443.https.tls.certificate.parsed.subject.common_name": 1, |
| 92 | "p443.https.tls.certificate.parsed.extensions.subject_alt_name.dns_names": 1, |
| 93 | "p443.https.tls.certificate.parsed.fingerprint_sha256": 1, |
| 94 | }, |
| 95 | ) |
| 96 | |
| 97 | for result in results: |
| 98 | i = next( |
| 99 | ( |
| 100 | index |
| 101 | for (index, item) in enumerate(current_certs) |
| 102 | if item["id"] |
| 103 | == result["p443"]["https"]["tls"]["certificate"]["parsed"][ |
| 104 | "fingerprint_sha256" |
| 105 | ] |
| 106 | ), |
| 107 | None, |
| 108 | ) |
| 109 | if i is None: |
| 110 | item = { |
| 111 | "id": result["p443"]["https"]["tls"]["certificate"]["parsed"][ |
| 112 | "fingerprint_sha256" |
| 113 | ] |
| 114 | } |
| 115 | dns_list = [] |
| 116 | try: |
| 117 | for dns_name in result["p443"]["https"]["tls"]["certificate"]["parsed"][ |
| 118 | "subject" |
| 119 | ]["common_name"]: |
| 120 | if dns_name not in dns_list: |
| 121 | dns_list.append(dns_name) |
| 122 | except KeyError: |
| 123 | pass |
| 124 | |
| 125 | # Not all certificates contain alternative names. |
| 126 | try: |
no test coverage detected