How to Monitor SSL Certificates and DNS Changes with Python
SSL expiry alerts are useful, but they only cover one part of a domain's operational state. A changed nameserver, a new mail provider, or an unexpected subdomain can be just as important.
I wanted one scheduled job that could watch all of those signals without maintaining separate WHOIS, DNS, TLS, and Certificate Transparency clients. The Domain Intelligence Suite returns the four data sets in one run, which makes the monitoring code mostly a matter of storing snapshots and comparing them.
What the monitor should detect
A useful domain monitor needs to report changes, not dump a fresh wall of JSON every morning. For each domain, we will track:
- SSL certificate expiry and issuer
- A, AAAA, MX, NS, and TXT records
- Registrar and registration expiry
- Subdomains found in Certificate Transparency logs
- Errors from individual lookup modules
That last item matters. WHOIS can fail while DNS and SSL still work. The actor keeps module failures isolated, so the monitor can use the good results instead of treating the entire check as failed.
The actor accepts a domain, an optional list of modules, an SSL port, DNS record types, and a subdomain limit. Protocols and paths are stripped automatically, but passing a clean hostname keeps your snapshot keys predictable.
Calling the actor synchronously
For a scheduled job, the synchronous dataset endpoint is convenient because the HTTP response contains the result item. There is no run polling loop to maintain.
import os
from typing import Any
import requests
APIFY_TOKEN = os.getenv("APIFY_TOKEN", "YOUR_APIFY_TOKEN")
ACTOR_ID = "weeknds~domain-intelligence-suite"
RUN_URL = (
f"https://api.apify.com/v2/acts/{ACTOR_ID}/"
"run-sync-get-dataset-items"
)
def inspect_domain(domain: str) -> dict[str, Any]:
payload = {
"domain": domain,
"modules": ["whois", "dns", "ssl", "subdomains"],
"sslPort": 443,
"dnsRecordTypes": ["A", "AAAA", "MX", "NS", "TXT"],
"maxSubdomains": 500,
"includeWildcardSubdomains": False,
}
response = requests.post(
RUN_URL,
params={"token": APIFY_TOKEN},
json=payload,
timeout=120,
)
response.raise_for_status()
items = response.json()
if not items:
raise RuntimeError(f"No domain intelligence returned for {domain}")
return items[0]
Keep the token in an environment variable in production. The fallback value is there so the example is runnable after replacing YOUR_APIFY_TOKEN, not as a recommendation to commit tokens.
Reduce the response to stable signals
Raw responses contain query timestamps and other values that change on every run. Comparing the entire item would create noise. Instead, normalize only the fields that should trigger an alert.
DNS MX records are objects, while A and NS records are usually strings. Sorting through a JSON representation gives us a stable comparison for either shape.
import json
from typing import Any
def stable_list(values: list[Any] | None) -> list[Any]:
return sorted(
values or [],
key=lambda value: json.dumps(value, sort_keys=True),
)
def module_error(module: dict[str, Any] | None) -> str | None:
if not module:
return "module missing from response"
return module.get("error")
def normalize_snapshot(item: dict[str, Any]) -> dict[str, Any]:
whois = item.get("whois", {})
dns = item.get("dns", {})
ssl = item.get("ssl", {})
subdomains = item.get("subdomains", {})
whois_data = whois.get("data") or {}
records = dns.get("records") or {}
certificate = ssl.get("certificate") or {}
return {
"domain": item["domain"],
"whois": {
"registrar": whois_data.get("registrar"),
"expiration_date": whois_data.get("expiration_date"),
"name_servers": stable_list(whois_data.get("name_servers")),
"error": module_error(whois),
},
"dns": {
record_type: stable_list(records.get(record_type))
for record_type in ["A", "AAAA", "MX", "NS", "TXT"]
} | {"error": module_error(dns)},
"ssl": {
"issuer": certificate.get("issuer"),
"valid_to": certificate.get("valid_to"),
"days_remaining": certificate.get("days_remaining"),
"expired": certificate.get("expired"),
"subject_alt_names": stable_list(
certificate.get("subject_alt_names")
),
"error": module_error(ssl),
},
"subdomains": {
"names": stable_list(subdomains.get("subdomains")),
"error": module_error(subdomains),
},
}
The actor discovers subdomains through crt.sh, so this is passive discovery from public Certificate Transparency logs. It does not prove that every hostname is live. Treat a new name as a review signal, then resolve or probe it separately if needed.
Save snapshots without a database
A JSON file per domain is enough for a small portfolio. Write through a temporary file so a stopped process cannot leave a half-written snapshot behind.
import json
from pathlib import Path
SNAPSHOT_DIR = Path("domain-snapshots")
def snapshot_path(domain: str) -> Path:
safe_name = domain.lower().replace("/", "_")
return SNAPSHOT_DIR / f"{safe_name}.json"
def load_previous(domain: str) -> dict | None:
path = snapshot_path(domain)
if not path.exists():
return None
return json.loads(path.read_text(encoding="utf-8"))
def save_snapshot(domain: str, snapshot: dict) -> None:
SNAPSHOT_DIR.mkdir(parents=True, exist_ok=True)
destination = snapshot_path(domain)
temporary = destination.with_suffix(".tmp")
temporary.write_text(
json.dumps(snapshot, indent=2, sort_keys=True),
encoding="utf-8",
)
temporary.replace(destination)
On the first run, the file becomes the baseline. Later runs compare against it.
Report only meaningful differences
A recursive diff keeps the comparison reusable. Lists are already normalized, so added DNS records and subdomains show up as clean old/new values.
from typing import Any
def diff_values(old: Any, new: Any, path: str = "") -> list[dict]:
if isinstance(old, dict) and isinstance(new, dict):
changes = []
for key in sorted(set(old) | set(new)):
child_path = f"{path}.{key}" if path else key
if key not in old:
changes.append(
{"field": child_path, "old": None, "new": new[key]}
)
elif key not in new:
changes.append(
{"field": child_path, "old": old[key], "new": None}
)
else:
changes.extend(diff_values(old[key], new[key], child_path))
return changes
if old != new:
return [{"field": path, "old": old, "new": new}]
return []
def urgent_ssl_warning(snapshot: dict, threshold_days: int = 30) -> str | None:
ssl = snapshot["ssl"]
if ssl["error"]:
return f"SSL lookup failed: {ssl['error']}"
days = ssl["days_remaining"]
if days is not None and days <= threshold_days:
return f"SSL certificate expires in {days} days"
return None
One caveat: days_remaining changes daily, so it will appear in the diff every day. If you only want threshold alerts, remove that field from the general comparison and keep urgent_ssl_warning().
Run it across a domain portfolio
The final script reads one domain per line, continues when an individual lookup fails, and only replaces a baseline after a successful result.
import sys
def monitor_domains(domains: list[str]) -> int:
alert_count = 0
for domain in domains:
try:
current = normalize_snapshot(inspect_domain(domain))
previous = load_previous(domain)
if previous is None:
print(f"{domain}: baseline created")
else:
changes = diff_values(previous, current)
for change in changes:
print(
f"{domain}: {change['field']} changed "
f"from {change['old']!r} to {change['new']!r}"
)
alert_count += len(changes)
warning = urgent_ssl_warning(current)
if warning:
print(f"{domain}: {warning}", file=sys.stderr)
alert_count += 1
save_snapshot(domain, current)
except (requests.RequestException, RuntimeError, ValueError) as error:
print(f"{domain}: check failed: {error}", file=sys.stderr)
alert_count += 1
return alert_count
if __name__ == "__main__":
domains = [
line.strip()
for line in Path("domains.txt").read_text().splitlines()
if line.strip() and not line.startswith("#")
]
raise SystemExit(1 if monitor_domains(domains) else 0)
That exit code is useful in cron, systemd timers, and CI jobs: zero means no changes, one means the output deserves attention. If daily days_remaining changes are enabled, switch the exit decision to count only selected fields.
Cost and scheduling
The Domain Intelligence Suite currently uses Apify's pay-per-usage model rather than a fixed subscription for the actor. The store describes a typical run as taking roughly 3 to 10 seconds, with the final charge depending on compute usage and your Apify plan. Running only the modules you need reduces both work and noise.
WHOIS and DNS rarely need minute-by-minute checks. A daily schedule is sensible for SSL and infrastructure monitoring, while a weekly schedule is usually enough for registration due diligence. Keep maxSubdomains conservative for routine runs, and raise it only when you are investigating a larger domain.
Top comments (0)