Created
June 24, 2026 19:06
-
-
Save fabidick22/b99d43e84ce86d25b285853f19213aa3 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| #!/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