Monitor certificate transparency logs for a domain.
| 22 | |
| 23 | |
| 24 | class Monitor: |
| 25 | """Monitor certificate transparency logs for a domain.""" |
| 26 | |
| 27 | def __init__(self, domain: str): |
| 28 | """Initialize the monitor with a validated domain.""" |
| 29 | if not self.validate_domain(domain): |
| 30 | raise ValueError("Invalid domain name") |
| 31 | self.domain = domain |
| 32 | self.last_checked = None |
| 33 | |
| 34 | def get_logs(self, count: int = 100): |
| 35 | """Fetch recent certificate transparency log entries.""" |
| 36 | url = f"{BASE_URL}/?q={self.domain}&output=json&limit={count}" |
| 37 | response = requests.get(url) |
| 38 | return response.json() |
| 39 | |
| 40 | def check_one_log(self, log: object): |
| 41 | """Fetch and inspect a single certificate log entry.""" |
| 42 | log_id = log["id"] |
| 43 | cert_url = f"{BASE_URL}/?d={log_id}" |
| 44 | cert_data = requests.get(cert_url).text |
| 45 | # Extract PEM-encoded certificate |
| 46 | import re |
| 47 | |
| 48 | pem_match = re.search( |
| 49 | r"-----BEGIN CERTIFICATE-----.*?-----END CERTIFICATE-----", |
| 50 | cert_data, |
| 51 | re.DOTALL, |
| 52 | ) |
| 53 | if pem_match: |
| 54 | pem_cert = pem_match.group(0) |
| 55 | |
| 56 | # Parse PEM certificate |
| 57 | cert = x509.load_pem_x509_certificate(pem_cert.encode(), default_backend()) |
| 58 | # Extract the public key |
| 59 | public_key = cert.public_key() |
| 60 | pem_public_key = public_key.public_bytes( |
| 61 | encoding=serialization.Encoding.PEM, |
| 62 | format=serialization.PublicFormat.SubjectPublicKeyInfo, |
| 63 | ) |
| 64 | print("Public Key:") |
| 65 | print(pem_public_key.hex()) |
| 66 | # Extract and print the issuer |
| 67 | print("Issuer:") |
| 68 | for attr in cert.issuer: |
| 69 | oid = attr.oid |
| 70 | if oid._name is not None: |
| 71 | name = oid._name |
| 72 | print(f" {name}: {attr.value}") |
| 73 | else: |
| 74 | print(f" {oid.dotted_string}: {attr.value}") |
| 75 | else: |
| 76 | print("No valid certificate found in the response.") |
| 77 | |
| 78 | def check_new_logs(self): |
| 79 | """Check for new log entries since the last check.""" |
| 80 | logs = self.get_logs(count=10000) |
| 81 | print("num logs", len(logs)) |