Skip to content

Instantly share code, notes, and snippets.

@fabidick22
Created June 24, 2026 19:06
Show Gist options
  • Select an option

  • Save fabidick22/b99d43e84ce86d25b285853f19213aa3 to your computer and use it in GitHub Desktop.

Select an option

Save fabidick22/b99d43e84ce86d25b285853f19213aa3 to your computer and use it in GitHub Desktop.
#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.11"
# dependencies = ["aiohttp"]
# ///
"""
check_dns_health.py — Find orphaned/dangling DNS records in a Cloudflare zone export.
Downloads your zone file from Cloudflare (DNS > Records > Export), then run this
script to discover which subdomains are alive and which are orphaned (pointing to
dead origins, decommissioned services, or dangling CNAMEs).
Detects:
- Cloudflare origin errors (521-530): web server down, SSL failures, timeouts
- Dangling CNAMEs: records pointing to removed services (Heroku, old ALBs, etc.)
- Orphaned A records: IPs that no longer respond
- Deep-probe verification: for domains behind reverse proxies (ALB, nginx) where
the root path returns 404 but specific paths have live backends
Usage:
uv run check_dns_health.py <zone-file> [options]
Options:
--concurrency N Max parallel requests (default: 20)
--timeout N Seconds per request (default: 10)
--deep-probe URLs Comma-separated domain/path pairs to verify 404s, e.g.:
"app.example.com/api/health,api.example.com/graphql"
--deep-probe-file FILE Text file with one domain/path per line, e.g.:
app.example.com/api/health
api.example.com/graphql
Output files:
dns-alive.txt — domains that respond (2xx, 3xx, 403, or deep-probe hit)
dns-dead.txt — orphaned domains (timeouts, CF errors, unmatched 404s)
dns-report.csv — full CSV with domain, type, target, status, and probe details
Examples:
# Basic scan
uv run check_dns_health.py example.com.txt
# With deep probes for domains behind a shared load balancer
uv run check_dns_health.py example.com.txt \\
--deep-probe "api.example.com/health,app.example.com/login"
# Deep probes from a file, higher concurrency
uv run check_dns_health.py example.com.txt \\
--deep-probe-file probes.txt --concurrency 50
"""
import asyncio
import csv
import re
import sys
from dataclasses import dataclass
from pathlib import Path
import aiohttp
# ── Configuration ─────────────────────────────────────────────────────────
ZONE_FILE = sys.argv[1] if len(sys.argv) > 1 else "lahaus.com.txt"
CONCURRENCY = 20
TIMEOUT = 10 # seconds per request
DEEP_PROBE_URLS: list[str] = []
i = 2
while i < len(sys.argv):
arg = sys.argv[i]
if arg == "--concurrency" and i + 1 < len(sys.argv):
CONCURRENCY = int(sys.argv[i + 1])
i += 2
elif arg == "--timeout" and i + 1 < len(sys.argv):
TIMEOUT = int(sys.argv[i + 1])
i += 2
elif arg == "--deep-probe" and i + 1 < len(sys.argv):
DEEP_PROBE_URLS.extend(sys.argv[i + 1].split(","))
i += 2
elif arg == "--deep-probe-file" and i + 1 < len(sys.argv):
with open(sys.argv[i + 1]) as f:
for line in f:
line = line.strip()
if line and not line.startswith("#"):
DEEP_PROBE_URLS.extend(line.split(","))
i += 2
else:
i += 1
REPORT_FILE = "dns-report.csv"
ALIVE_FILE = "dns-alive.txt"
DEAD_FILE = "dns-dead.txt"
# ── Data model ────────────────────────────────────────────────────────────
@dataclass
class DnsRecord:
domain: str
record_type: str
target: str
cf_proxied: bool
@dataclass
class ProbeResult:
record: DnsRecord
http_code: int
status: str
server_header: str = ""
deep_probe_code: int = 0
deep_probe_path: str = ""
# ── 1. Parse zone file ───────────────────────────────────────────────────
SKIP_PATTERNS = [
re.compile(r"^_[0-9a-f]{8,}\."), # ACM validation CNAMEs
re.compile(r"\._domainkey\."), # DKIM records
re.compile(r"^fwdkim"), # FreshEmail DKIM forwarding
re.compile(r"^em\d+\."), # SendGrid email link tracking
re.compile(r"^url\d+\."), # SendGrid URL tracking
re.compile(r"^\d{5,}\."), # Numeric SendGrid subdomains (954070)
re.compile(r"^\*\."), # Wildcard records
]
def parse_zone_file(path: str) -> list[DnsRecord]:
records: dict[str, DnsRecord] = {}
with open(path) as f:
for line in f:
line = line.strip()
if not line or line.startswith(";;"):
continue
parts = line.split()
if len(parts) < 5:
continue
# Format: domain TTL IN TYPE target [; comments]
domain = parts[0].rstrip(".")
record_type = parts[3]
target = parts[4].rstrip(".")
if record_type not in ("A", "CNAME"):
continue
# Skip non-browsable records
if any(p.search(domain) for p in SKIP_PATTERNS):
continue
cf_proxied = "cf-proxied:true" in line
# Deduplicate by domain (keep first occurrence)
if domain not in records:
records[domain] = DnsRecord(
domain=domain,
record_type=record_type,
target=target,
cf_proxied=cf_proxied,
)
return sorted(records.values(), key=lambda r: r.domain)
# ── 1b. Parse deep-probe URLs ────────────────────────────────────────────
def parse_deep_probes(raw_urls: list[str]) -> dict[str, str]:
"""Parse URLs like 'domain.com/path' into {domain: /path} map."""
probes: dict[str, str] = {}
for url in raw_urls:
url = url.strip()
if not url:
continue
# Strip scheme if present
url = re.sub(r"^https?://", "", url)
# Split domain from path
if "/" in url:
domain, path = url.split("/", 1)
path = "/" + path
else:
domain = url
path = "/"
probes[domain] = path
return probes
# ── 2. Classify HTTP response ────────────────────────────────────────────
CF_ERROR_CODES = {
521: "CF Web Server Down",
522: "CF Connection Timed Out",
523: "CF Origin Unreachable",
524: "CF Timeout",
525: "CF SSL Handshake Failed",
526: "CF Invalid SSL Certificate",
530: "CF Origin Error",
}
def classify(code: int, server_header: str, cf_proxied: bool) -> str:
if code == 0:
return "DEAD (no response / timeout)"
if 200 <= code < 300:
return "ALIVE"
if 300 <= code < 400:
return "ALIVE (redirect)"
if code == 403:
return "ALIVE (403 Forbidden — origin responds)"
if code == 404:
# If the domain is NOT CF-proxied, any response means origin is alive.
if not cf_proxied:
return "ALIVE (404 — origin responds, not proxied)"
# CF-proxied: check if a real backend answered behind CF.
srv = server_header.lower()
backend_markers = [
"nginx", "apache", "hubspot", "cloudfront", "amazon",
"gunicorn", "express", "openresty", "vercel", "netlify",
"heroku", "webflow", "zendesk", "canny", "envoy",
"istio-envoy", "akamai", "varnish", "caddy", "traefik",
]
if any(marker in srv for marker in backend_markers):
return f"ALIVE (404 — backend: {server_header})"
# Will be re-classified after deep probe if available
return "DEAD (404 Not Found)"
if code in CF_ERROR_CODES:
return f"DEAD ({code} {CF_ERROR_CODES[code]})"
if 400 <= code < 500:
return f"ERROR ({code})"
if 500 <= code < 600:
return f"DEAD ({code} Server Error)"
return f"UNKNOWN ({code})"
# ── 3. Probe domains concurrently ────────────────────────────────────────
async def fetch_status(
session: aiohttp.ClientSession,
domain: str,
path: str = "/",
) -> tuple[int, str]:
"""Try HTTPS then HTTP for domain+path. Returns (status_code, server_header)."""
timeout_cfg = aiohttp.ClientTimeout(total=TIMEOUT * 2, connect=TIMEOUT)
for scheme in ("https", "http"):
url = f"{scheme}://{domain}{path}"
try:
async with session.get(
url,
timeout=timeout_cfg,
allow_redirects=False,
ssl=False,
) as resp:
return resp.status, resp.headers.get("server", "")
except Exception:
if scheme == "https":
continue
return 0, ""
return 0, ""
async def probe_one(
session: aiohttp.ClientSession,
record: DnsRecord,
semaphore: asyncio.Semaphore,
deep_probes: dict[str, str],
) -> ProbeResult:
async with semaphore:
# Phase 1: probe /
code, server_header = await fetch_status(session, record.domain)
status = classify(code, server_header, record.cf_proxied)
deep_probe_code = 0
deep_probe_path = ""
# Phase 2: deep probe if root returned 404 and we have a path for this domain
if code == 404 and record.domain in deep_probes:
deep_probe_path = deep_probes[record.domain]
deep_code, deep_server = await fetch_status(
session, record.domain, deep_probe_path,
)
deep_probe_code = deep_code
# If the deep probe got anything other than 404, the backend is alive
if deep_code != 404 and deep_code != 0:
status = f"ALIVE (deep-probe {deep_probe_path} → {deep_code})"
return ProbeResult(
record=record,
http_code=code,
status=status,
server_header=server_header,
deep_probe_code=deep_probe_code,
deep_probe_path=deep_probe_path,
)
async def probe_all(
records: list[DnsRecord],
deep_probes: dict[str, str],
) -> list[ProbeResult]:
semaphore = asyncio.Semaphore(CONCURRENCY)
connector = aiohttp.TCPConnector(limit=CONCURRENCY, ssl=False)
async with aiohttp.ClientSession(connector=connector) as session:
tasks = [probe_one(session, r, semaphore, deep_probes) for r in records]
total = len(tasks)
results: list[ProbeResult] = []
for i, coro in enumerate(asyncio.as_completed(tasks), 1):
result = await coro
results.append(result)
print(
f"\r[*] Progress: {i}/{total} "
f"({result.record.domain} → {result.http_code})",
end="",
flush=True,
)
print() # newline after progress
return results
# ── 4. Write output files ────────────────────────────────────────────────
def write_results(results: list[ProbeResult]) -> None:
results.sort(key=lambda r: r.record.domain)
# CSV report
with open(REPORT_FILE, "w", newline="") as f:
writer = csv.writer(f)
writer.writerow([
"domain", "record_type", "target", "cf_proxied",
"http_code", "server_header", "deep_probe_path",
"deep_probe_code", "status",
])
for r in results:
writer.writerow([
r.record.domain,
r.record.record_type,
r.record.target,
r.record.cf_proxied,
r.http_code,
r.server_header,
r.deep_probe_path,
r.deep_probe_code or "",
r.status,
])
# Alive list
alive = [r for r in results if r.status.startswith("ALIVE")]
with open(ALIVE_FILE, "w") as f:
for r in alive:
f.write(f"{r.record.domain}\n")
# Dead list
dead = [r for r in results if not r.status.startswith("ALIVE")]
with open(DEAD_FILE, "w") as f:
for r in dead:
f.write(f"{r.record.domain}\n")
return alive, dead
# ── 5. Main ──────────────────────────────────────────────────────────────
def main():
print(f"[*] Parsing zone file: {ZONE_FILE}")
records = parse_zone_file(ZONE_FILE)
print(f"[*] Found {len(records)} browsable subdomains to probe")
deep_probes = parse_deep_probes(DEEP_PROBE_URLS)
if deep_probes:
print(f"[*] Loaded {len(deep_probes)} deep-probe paths:")
for domain, path in sorted(deep_probes.items()):
print(f" {domain}{path}")
else:
print("[*] No deep-probe paths (use --deep-probe or --deep-probe-file to verify 404s)")
print(f"[*] Concurrency: {CONCURRENCY}, Timeout: {TIMEOUT}s\n")
results = asyncio.run(probe_all(records, deep_probes))
alive, dead = write_results(results)
print()
print("=" * 60)
print(" RESULTS")
print("=" * 60)
print(f" Total checked: {len(results)}")
print(f" Alive (2xx/3xx): {len(alive)}")
print(f" Dead / Error: {len(dead)}")
print()
print(f" Files:")
print(f" {REPORT_FILE} — full CSV report")
print(f" {ALIVE_FILE} — alive domain list")
print(f" {DEAD_FILE} — dead/orphan domain list")
print("=" * 60)
if dead:
print()
print("── Dead / Orphan domains ──────────────────────────────────")
for r in dead:
print(f" {r.record.domain:<50s} {r.status}")
print()
if __name__ == "__main__":
main()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment