Wildcard record points at a deprovisioned service

"""Detect a wildcard CNAME pointing at a deprovisioned service and optionally
delete it via Cloudflare. Safe by default. Set DRY_RUN=false to let it write.
"""
import os
import logging

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("dangling_wildcard")

DNS_DOMAIN = os.environ.get("DNS_DOMAIN", "example.com")
CLOUDFLARE_API_TOKEN = os.environ.get("CLOUDFLARE_API_TOKEN", "")
CLOUDFLARE_ZONE_ID = os.environ.get("CLOUDFLARE_ZONE_ID", "")
DRY_RUN = os.environ.get("DRY_RUN", "true").lower() == "true"

CF_API = "https://api.cloudflare.com/client/v4"

KNOWN_VULNERABLE_FINGERPRINTS = {
    "no such app",
    "nosuchbucket",
    "there isn't a github pages site here",
    "unrecognized domain",
}


def is_dangling_wildcard(record, resolved_target_status, http_fingerprint, known_vulnerable_fingerprints):
    """Pure decision function. No I/O.

    record: dict, must have name starting with "*" and type "CNAME".
    resolved_target_status: one of "NXDOMAIN", "SERVFAIL", "OK", pre-resolved
      by the caller via dns.resolver.
    http_fingerprint: a lowercase string from the target's HTTP response, or
      None if no request was made.
    known_vulnerable_fingerprints: set of known "unclaimed resource" strings.

    Returns True if the wildcard CNAME target is dangling.
    """
    if not record.get("name", "").startswith("*"):
        return False
    if record.get("type") != "CNAME":
        return False
    if resolved_target_status != "OK":
        return True
    if http_fingerprint in known_vulnerable_fingerprints:
        return True
    return False


def list_cname_records():
    """List every CNAME record in the zone via the Cloudflare API."""
    import requests

    headers = {"Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}"}
    url = f"{CF_API}/zones/{CLOUDFLARE_ZONE_ID}/dns_records"
    r = requests.get(url, headers=headers, params={"type": "CNAME"}, timeout=30)
    r.raise_for_status()
    return r.json().get("result", [])


def resolve_target_status(hostname):
    """Resolve a CNAME target and classify it as OK, NXDOMAIN, or SERVFAIL."""
    import dns.resolver

    resolver = dns.resolver.Resolver()
    try:
        resolver.resolve(hostname, "A")
        return "OK"
    except dns.resolver.NXDOMAIN:
        return "NXDOMAIN"
    except dns.resolver.NoAnswer:
        return "OK"
    except Exception as exc:
        log.warning("Resolving %s failed: %s", hostname, exc)
        return "SERVFAIL"


def probe_http_fingerprint(hostname):
    """Fetch the hostname over HTTPS and return a lowercase fingerprint string."""
    import requests

    try:
        resp = requests.get(f"https://{hostname}/", timeout=10)
        return resp.text.lower()
    except Exception as exc:
        log.warning("HTTP probe for %s failed: %s", hostname, exc)
        return ""


def delete_record(record_id):
    """Delete a dangling wildcard record through the Cloudflare API."""
    import requests

    headers = {"Authorization": f"Bearer {CLOUDFLARE_API_TOKEN}"}
    url = f"{CF_API}/zones/{CLOUDFLARE_ZONE_ID}/dns_records/{record_id}"

    if DRY_RUN:
        log.info("[dry run] would delete record %s", record_id)
        return

    requests.delete(url, headers=headers, timeout=30).raise_for_status()
    log.info("Deleted dangling wildcard record %s", record_id)


def run():
    if not (CLOUDFLARE_API_TOKEN and CLOUDFLARE_ZONE_ID):
        log.warning("No Cloudflare credentials set. Nothing to scan.")
        return

    records = list_cname_records()
    flagged = 0
    for record in records:
        if not record.get("name", "").startswith("*"):
            continue
        target = record.get("content", "")
        status = resolve_target_status(target)
        fingerprint = probe_http_fingerprint(target) if status == "OK" else ""
        dangling = is_dangling_wildcard(record, status, fingerprint, KNOWN_VULNERABLE_FINGERPRINTS)
        if dangling:
            log.info("Wildcard %s -> %s is dangling (%s)", record["name"], target, status)
            delete_record(record["id"])
            flagged += 1
        else:
            log.info("Wildcard %s -> %s looks fine", record["name"], target)

    log.info("Done. %d dangling wildcard(s) %s.", flagged, "to remove" if DRY_RUN else "removed")


if __name__ == "__main__":
    run()

/**
 * Detect a wildcard CNAME pointing at a deprovisioned service and optionally
 * delete it via Cloudflare. Safe by default. Set DRY_RUN=false to let it write.
 */
import { pathToFileURL } from "node:url";

const DNS_DOMAIN = process.env.DNS_DOMAIN || "example.com";
const CLOUDFLARE_API_TOKEN = process.env.CLOUDFLARE_API_TOKEN || "";
const CLOUDFLARE_ZONE_ID = process.env.CLOUDFLARE_ZONE_ID || "";
const DRY_RUN = (process.env.DRY_RUN || "true").toLowerCase() === "true";
const CF_API = "https://api.cloudflare.com/client/v4";

const KNOWN_VULNERABLE_FINGERPRINTS = new Set([
  "no such app",
  "nosuchbucket",
  "there isn't a github pages site here",
  "unrecognized domain",
]);

/**
 * Pure decision function. No I/O.
 *
 * record: object, must have name starting with "*" and type "CNAME".
 * resolvedTargetStatus: one of "NXDOMAIN", "SERVFAIL", "OK", pre-resolved by
 *   the caller.
 * httpFingerprint: a lowercase string from the target's HTTP response, or
 *   null if no request was made.
 * knownVulnerableFingerprints: Set of known "unclaimed resource" strings.
 *
 * Returns true if the wildcard CNAME target is dangling.
 */
export function isDanglingWildcard(record, resolvedTargetStatus, httpFingerprint, knownVulnerableFingerprints) {
  if (!record.name || !record.name.startsWith("*")) return false;
  if (record.type !== "CNAME") return false;
  if (resolvedTargetStatus !== "OK") return true;
  if (knownVulnerableFingerprints.has(httpFingerprint)) return true;
  return false;
}

/** List every CNAME record in the zone via the Cloudflare API. */
async function listCnameRecords() {
  const url = `${CF_API}/zones/${CLOUDFLARE_ZONE_ID}/dns_records?type=CNAME`;
  const res = await fetch(url, { headers: { Authorization: `Bearer ${CLOUDFLARE_API_TOKEN}` } });
  if (!res.ok) throw new Error(`Cloudflare list failed: ${res.status}`);
  const body = await res.json();
  return body.result || [];
}

/** Resolve a CNAME target and classify it as OK, NXDOMAIN, or SERVFAIL. */
async function resolveTargetStatus(hostname) {
  const dns = await import("node:dns/promises");
  try {
    await dns.resolve(hostname, "A");
    return "OK";
  } catch (err) {
    if (err.code === "ENOTFOUND" || err.code === "ENODATA") return "NXDOMAIN";
    return "SERVFAIL";
  }
}

/** Fetch the hostname over HTTPS and return a lowercase fingerprint string. */
async function probeHttpFingerprint(hostname) {
  try {
    const res = await fetch(`https://${hostname}/`, { signal: AbortSignal.timeout(10000) });
    const text = await res.text();
    return text.toLowerCase();
  } catch {
    return "";
  }
}

/** Delete a dangling wildcard record through the Cloudflare API. */
async function deleteRecord(recordId) {
  const url = `${CF_API}/zones/${CLOUDFLARE_ZONE_ID}/dns_records/${recordId}`;
  const headers = { Authorization: `Bearer ${CLOUDFLARE_API_TOKEN}` };

  if (DRY_RUN) {
    console.log(`[dry run] would delete record ${recordId}`);
    return;
  }

  const res = await fetch(url, { method: "DELETE", headers });
  if (!res.ok) throw new Error(`Cloudflare delete failed: ${res.status}`);
  console.log(`Deleted dangling wildcard record ${recordId}`);
}

async function run() {
  if (!(CLOUDFLARE_API_TOKEN && CLOUDFLARE_ZONE_ID)) {
    console.warn("No Cloudflare credentials set. Nothing to scan.");
    return;
  }

  const records = await listCnameRecords();
  let flagged = 0;
  for (const record of records) {
    if (!record.name || !record.name.startsWith("*")) continue;
    const target = record.content || "";
    const status = await resolveTargetStatus(target);
    const fingerprint = status === "OK" ? await probeHttpFingerprint(target) : "";
    const dangling = isDanglingWildcard(record, status, fingerprint, KNOWN_VULNERABLE_FINGERPRINTS);
    if (dangling) {
      console.log(`Wildcard ${record.name} -> ${target} is dangling (${status})`);
      await deleteRecord(record.id);
      flagged++;
    } else {
      console.log(`Wildcard ${record.name} -> ${target} looks fine`);
    }
  }

  console.log(`Done. ${flagged} dangling wildcard(s) ${DRY_RUN ? "to remove" : "removed"}.`);
}

if (process.argv[1] && import.meta.url === pathToFileURL(process.argv[1]).href) {
  run().catch((err) => {
    console.error(err);
    process.exit(1);
  });
}

from dangling_wildcard import is_dangling_wildcard

VULN = {"no such app", "nosuchbucket"}


def wildcard_record(**over):
    base = {"name": "*.example.com", "type": "CNAME", "content": "old-app.paas.net"}
    base.update(over)
    return base


def test_dangling_when_target_is_nxdomain():
    record = wildcard_record()
    assert is_dangling_wildcard(record, "NXDOMAIN", None, VULN) is True


def test_dangling_when_target_is_servfail():
    record = wildcard_record()
    assert is_dangling_wildcard(record, "SERVFAIL", None, VULN) is True


def test_dangling_when_fingerprint_matches_known_vulnerable():
    record = wildcard_record()
    assert is_dangling_wildcard(record, "OK", "no such app", VULN) is True


def test_not_dangling_when_target_ok_and_fingerprint_unknown():
    record = wildcard_record()
    assert is_dangling_wildcard(record, "OK", "welcome home", VULN) is False


def test_not_dangling_when_not_a_wildcard_name():
    record = wildcard_record(name="shop.example.com")
    assert is_dangling_wildcard(record, "NXDOMAIN", None, VULN) is False


def test_not_dangling_when_not_cname_type():
    record = wildcard_record(type="A")
    assert is_dangling_wildcard(record, "NXDOMAIN", None, VULN) is False

import { test } from "node:test";
import assert from "node:assert/strict";
import { isDanglingWildcard } from "./dangling-wildcard.js";

const VULN = new Set(["no such app", "nosuchbucket"]);

const wildcardRecord = (over = {}) => ({
  name: "*.example.com",
  type: "CNAME",
  content: "old-app.paas.net",
  ...over,
});

test("dangling when target is nxdomain", () => {
  assert.equal(isDanglingWildcard(wildcardRecord(), "NXDOMAIN", null, VULN), true);
});

test("dangling when target is servfail", () => {
  assert.equal(isDanglingWildcard(wildcardRecord(), "SERVFAIL", null, VULN), true);
});

test("dangling when fingerprint matches known vulnerable", () => {
  assert.equal(isDanglingWildcard(wildcardRecord(), "OK", "no such app", VULN), true);
});

test("not dangling when target ok and fingerprint unknown", () => {
  assert.equal(isDanglingWildcard(wildcardRecord(), "OK", "welcome home", VULN), false);
});

test("not dangling when not a wildcard name", () => {
  const record = wildcardRecord({ name: "shop.example.com" });
  assert.equal(isDanglingWildcard(record, "NXDOMAIN", null, VULN), false);
});

test("not dangling when not cname type", () => {
  const record = wildcardRecord({ type: "A" });
  assert.equal(isDanglingWildcard(record, "NXDOMAIN", null, VULN), false);
});