Infoblox Stale DNS Record Finder

#!/usr/bin/env python3
"""
Infoblox Stale DNS Record Finder

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/infoblox-stale-dns-records

It covers what the script does, whether it is safe to run, the
inventory file and credentials it needs, how to read the output,
and what to do when something fails.

    pip install "requests>=2.28"
"""

# SAFE: read-only. Every WAPI call is a GET; there is no POST, PUT or DELETE
# anywhere in this file. It never deletes or edits a record.
#
# EXIT CODES   0 nothing to act on   1 findings to review   2 the run failed
#
# CREDENTIALS come from the environment, never from an argument:
#   IB_USER / IB_PASS      (not needed at all with --offline-dir)
#
# NIOS / WAPI v2.x. Not Universal DDI.

import argparse
import csv
import ipaddress
import json
import os
import socket
import sys
from datetime import datetime, timezone

try:
    import requests
    from requests.auth import HTTPBasicAuth
except ImportError:
    sys.stderr.write("requests is not installed. Run: pip install 'requests>=2.28'\n")
    raise SystemExit(2)

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass

DEFAULT_WAPI = "v2.12"

# ---------------------------------------------------------------- verdicts
STALE = "STALE"
SUSPECT = "SUSPECT"
UNVERIFIABLE = "COULD_NOT_VERIFY"
DHCP = "DHCP_MANAGED"
BY_DESIGN = "BY_DESIGN"

SEVERITY = {STALE: 0, SUSPECT: 1, UNVERIFIABLE: 2, DHCP: 3, BY_DESIGN: 4}

# Where a dangling CNAME stops being untidy and becomes a takeover.
TAKEOVER_SUFFIXES = (
    "cloudfront.net", "s3.amazonaws.com", "elb.amazonaws.com",
    "azurewebsites.net", "cloudapp.azure.com", "trafficmanager.net",
    "github.io", "herokuapp.com", "herokudns.com", "pages.dev",
    "netlify.app", "fastly.net", "wpengine.com", "zendesk.com",
)


class Finding(object):
    """One record, one verdict, and the evidence behind it."""

    def __init__(self, check, kind, name, view, verdict, reason, detail=""):
        self.check = check        # which of the four checks produced this
        self.kind = kind          # record:a / record:ptr / record:cname
        self.name = name          # the FQDN, plus the data that matters
        self.view = view          # DNS view: the same name legitimately repeats
        self.verdict = verdict
        self.reason = reason      # why this verdict, in one line
        self.detail = detail      # what to do about it, or what blocked the check


# ---------------------------------------------------------------- transport
class Wapi(object):
    """Live WAPI, or a replay of saved responses. Same interface either way.

    Offline replay is not a convenience. It is how this script was tested
    without a grid, and it is how you can hand a colleague the exact data a
    finding came from.
    """

    def __init__(self, host=None, version=DEFAULT_WAPI, insecure=False,
                 offline_dir=None, save_dir=None):
        self.offline_dir = offline_dir
        self.save_dir = save_dir
        self.base = None
        self.session = None
        if offline_dir:
            if not os.path.isdir(offline_dir):
                raise SystemExit("--offline-dir does not exist: %s" % offline_dir)
            return
        user = os.environ.get("IB_USER")
        password = os.environ.get("IB_PASS")
        if not user or not password:
            raise SystemExit(
                "IB_USER and IB_PASS must be set in the environment.\n"
                "  Linux/macOS : export IB_USER=admin; read -s IB_PASS; export IB_PASS\n"
                "  PowerShell  : $env:IB_USER='admin'; $env:IB_PASS='...'\n"
                "They are read from the environment on purpose: an argument is "
                "visible in 'ps auxww' and lands in shell history."
            )
        self.base = "https://%s/wapi/%s/" % (host, version)
        self.session = requests.Session()
        self.session.auth = HTTPBasicAuth(user, password)
        self.session.verify = not insecure
        if insecure:
            try:
                import urllib3
                urllib3.disable_warnings()
            except Exception:
                pass

    @staticmethod
    def _slug(obj, params):
        """Filename for one query. Must survive Windows, so everything that is
        not alphanumeric becomes a dash -- '>' and ':' are legal in a WAPI
        search key and illegal in an NTFS filename."""
        def clean(text):
            return "".join(c if c.isalnum() else "-" for c in str(text))

        bits = [clean(obj)]
        for k in sorted(params):
            if k.startswith("_"):
                continue
            bits.append("%s=%s" % (clean(k), clean(params[k])))
        return ("__".join(bits) or "root")[:180] + ".json"

    def get(self, obj, params=None):
        """Return a list of objects, following WAPI paging to the end.

        Paging in WAPI only works with _return_as_object=1 -- without it the
        next_page_id never comes back and you silently get the first page and
        nothing else. _max_results is NEGATIVE on purpose: a positive value
        makes WAPI raise 'Result set too large' instead of truncating, which
        turns a big zone into a failed run rather than a partial one.
        """
        params = dict(params or {})
        slug = self._slug(obj, params)

        if self.offline_dir:
            path = os.path.join(self.offline_dir, slug)
            if not os.path.exists(path):
                return []
            with open(path, encoding="utf-8") as fh:
                return json.load(fh)

        params["_return_as_object"] = 1
        params["_paging"] = 1
        params["_max_results"] = -1000
        out = []
        page = None
        while True:
            call = dict(params)
            if page:
                call = {"_page_id": page, "_return_as_object": 1}
            resp = self.session.get(self.base + obj, params=call, timeout=60)
            if resp.status_code == 401:
                raise SystemExit("WAPI rejected IB_USER/IB_PASS (401).")
            if resp.status_code >= 400:
                raise SystemExit("WAPI %s on %s: %s" % (resp.status_code, obj, resp.text[:300]))
            body = resp.json()
            out.extend(body.get("result", []))
            page = body.get("next_page_id")
            if not page:
                break

        if self.save_dir:
            if not os.path.isdir(self.save_dir):
                os.makedirs(self.save_dir)
            with open(os.path.join(self.save_dir, slug), "w", encoding="utf-8") as fh:
                json.dump(out, fh, indent=1)
        return out


# ---------------------------------------------------------------- helpers
def age_days(epoch):
    """Epoch seconds to whole days ago, or None when the field is absent.

    None is NOT zero. 'last_queried is empty' and 'last queried today' are
    opposite facts, and a dash is the only honest way to print the first.
    """
    if not epoch:
        return None
    try:
        then = datetime.fromtimestamp(float(epoch), timezone.utc)
    except (TypeError, ValueError, OSError):
        return None
    return (datetime.now(timezone.utc) - then).days


def dashed(value):
    return "-" if value in (None, "") else str(value)


def reverse_name(ip):
    try:
        return ipaddress.ip_address(ip).reverse_pointer
    except ValueError:
        return None


def is_suffix_zone(name, zone):
    """True when `name` lives inside `zone`, by label, not by characters.

    Comparing with endswith() alone matches 'evilexample.com' against
    'example.com', which is how a check like this quietly stops working.
    """
    name = (name or "").rstrip(".").lower()
    zone = (zone or "").rstrip(".").lower()
    return bool(zone) and (name == zone or name.endswith("." + zone))


def classless_cidr(zone_fqdn):
    """RFC 2317 reverse zone ('0/25.3.2.10.in-addr.arpa') to a CIDR.

    Classless reverse delegation is normal on /25 and longer, and a plain
    label-suffix match never matches it -- which would report every PTR in
    such a zone as living in no zone at all. Returns None for an ordinary
    zone.
    """
    labels = (zone_fqdn or "").rstrip(".").lower().split(".")
    if len(labels) < 3 or "/" not in labels[0]:
        return None
    first, prefix = labels[0].split("/", 1)
    octets = list(reversed(labels[1:-2]))  # drop in-addr.arpa, then reverse
    try:
        quad = ".".join(octets + [first])
        while quad.count(".") < 3:
            quad += ".0"
        return ipaddress.ip_network("%s/%s" % (quad, prefix), strict=False)
    except ValueError:
        return None


def ip_from_reverse(name):
    labels = (name or "").rstrip(".").lower().split(".")
    if len(labels) < 6 or labels[-2:] != ["in-addr", "arpa"]:
        return None
    try:
        return ipaddress.ip_address(".".join(reversed(labels[-6:-2])))
    except ValueError:
        return None


class Zones(object):
    """Which names this grid authors, which it delegates, and which it does
    not host at all. Every verdict in this script turns on that distinction."""

    def __init__(self, auth, delegated):
        self.auth = [z for z in auth if not z.get("disable")]
        self.disabled = [z for z in auth if z.get("disable")]
        self.delegated = delegated

    @staticmethod
    def _best(name, zones, view):
        """Longest matching zone in this view, classless reverse included."""
        best, best_len = None, -1
        rev_ip = ip_from_reverse(name)
        for z in zones:
            if (z.get("view") or "default") != view:
                continue
            fqdn = z.get("fqdn") or ""
            hit = is_suffix_zone(name, fqdn)
            if not hit and rev_ip is not None:
                net = classless_cidr(fqdn)
                hit = bool(net) and rev_ip in net
            if hit and len(fqdn) > best_len:
                best, best_len = z, len(fqdn)
        return best

    def status(self, name, view):
        """-> ('authoritative'|'delegated'|'disabled'|'absent', zone fqdn|None)

        Longest match wins, and a delegation inside an authoritative zone wins
        over it when it is longer: that is exactly how resolution works, and
        getting the precedence backwards is what makes a tool call a delegated
        subzone 'ours' and then report its records as broken.
        """
        cands = []
        for kind, zones in (("authoritative", self.auth),
                            ("delegated", self.delegated),
                            ("disabled", self.disabled)):
            z = self._best(name, zones, view)
            if z:
                cands.append((len(z.get("fqdn") or ""), kind, z.get("fqdn")))
        if not cands:
            return "absent", None
        # Delegation first on a tie: the delegation is what resolvers follow.
        rank = {"delegated": 0, "authoritative": 1, "disabled": 2}
        cands.sort(key=lambda c: (-c[0], rank[c[1]]))
        _, kind, fqdn = cands[0]
        return kind, fqdn


def probe(ip, ports, timeout):
    """Is anything answering at this address? Three answers, never two.

      'alive'      something completed a handshake, or actively refused one.
                   A refusal is a REPLY: the host is up and said no.
      'no-answer'  nothing came back on any port tried.
      'unknown'    the probe could not be attempted at all.

    'no-answer' is NOT 'dead', and this script never promotes it to one. A
    host can drop every unsolicited packet by policy, sit behind a firewall
    that eats the probe, or be powered off for a maintenance window that ends
    on Monday. ICMP is not used either: a raw socket needs root, and an echo
    reply that never comes would not settle the question.
    """
    try:
        ipaddress.ip_address(ip)
    except ValueError:
        return "unknown"
    for port in ports:
        s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        s.settimeout(timeout)
        try:
            s.connect((ip, int(port)))
            return "alive"
        except ConnectionRefusedError:
            return "alive"
        except (socket.timeout, OSError):
            continue
        finally:
            s.close()
    return "no-answer"


# ---------------------------------------------------------------- analysis
class Audit(object):
    def __init__(self, wapi, args):
        self.w = wapi
        self.args = args
        self.findings = []
        self.probes = {}
        self.counts = {}

    def add(self, check, kind, name, view, verdict, reason, detail=""):
        """One record can fail two different checks, and that is not a
        duplicate: 'no PTR' and 'nobody queries it' are separate facts about
        the same name. The check label is printed so the two never read as
        the same line twice."""
        self.findings.append(Finding(check, kind, name, view, verdict, reason, detail))

    # -- fetch ------------------------------------------------------------
    def fetch(self):
        """One query per object type, then all the set maths locally.

        Asking WAPI 'is there a PTR for this A' once per record is 40,000
        round trips on a medium grid. Reading each object type once and
        indexing it is the difference between seconds and an afternoon.
        """
        w = self.w
        self.zones = Zones(
            w.get("zone_auth", {"_return_fields": "fqdn,view,zone_format,disable"}),
            w.get("zone_delegated", {"_return_fields": "fqdn,view,zone_format,delegate_to"}),
        )
        self.a = w.get("record:a", {"_return_fields":
                       "name,ipv4addr,view,zone,disable,creator,comment,"
                       "last_queried,ddns_protected"})
        self.ptr = w.get("record:ptr", {"_return_fields":
                         "ptrdname,ipv4addr,name,view,zone,disable,creator,"
                         "comment,last_queried"})
        self.cname = w.get("record:cname", {"_return_fields":
                           "name,canonical,view,zone,disable,creator,comment,"
                           "last_queried"})
        self.host = w.get("record:host", {"_return_fields":
                          "name,ipv4addrs,view,disable,comment,configure_for_dns"})
        self.fixed = w.get("fixedaddress", {"_return_fields":
                           "ipv4addr,mac,name,network_view,comment"})
        self.leases = []
        if self.args.leases:
            self.leases = w.get("lease", {"_return_fields":
                                "address,binding_state,ends,client_hostname"})
        if self.w.offline_dir:
            path = os.path.join(self.w.offline_dir, "probe.json")
            if os.path.exists(path):
                with open(path, encoding="utf-8") as fh:
                    self.probes = json.load(fh)

        self.fwd_by_ip = {}        # (view, ip)   -> [names]  from A and host
        self.fwd_names = {}        # (view, name) -> [ips]    from A and host
        self.ptr_by_ip = {}        # (view, ip)   -> [ptrdname]
        self.cname_names = set()   # (view, name)
        self.cname_target = {}     # (view, name) -> canonical
        self.dhcp_ips = set()      # addresses DHCP owns

        for r in self.a:
            if r.get("disable"):
                continue
            v = r.get("view") or "default"
            n = (r.get("name") or "").lower()
            ip = r.get("ipv4addr")
            self.fwd_by_ip.setdefault((v, ip), []).append(n)
            self.fwd_names.setdefault((v, n), []).append(ip)

        # A host record is NOT an A record plus a PTR record: NIOS synthesises
        # both from one object. Treating it as a missing A is the classic false
        # positive here, and it is why host records are indexed as forward data
        # and never reported as orphans.
        for h in self.host:
            if h.get("disable") or h.get("configure_for_dns") is False:
                continue
            v = h.get("view") or "default"
            n = (h.get("name") or "").lower()
            for addr in h.get("ipv4addrs") or []:
                ip = addr.get("ipv4addr")
                self.fwd_by_ip.setdefault((v, ip), []).append(n)
                self.fwd_names.setdefault((v, n), []).append(ip)
                self.ptr_by_ip.setdefault((v, ip), []).append(n)
                if addr.get("configure_for_dhcp"):
                    self.dhcp_ips.add(ip)

        for p in self.ptr:
            if p.get("disable"):
                continue
            self.ptr_by_ip.setdefault(
                (p.get("view") or "default", p.get("ipv4addr")), []
            ).append((p.get("ptrdname") or "").lower())
        for c in self.cname:
            if c.get("disable"):
                continue
            key = ((c.get("view") or "default"), (c.get("name") or "").lower())
            self.cname_names.add(key)
            self.cname_target[key] = (c.get("canonical") or "").lower()
        for f in self.fixed:
            if f.get("ipv4addr"):
                self.dhcp_ips.add(f["ipv4addr"])
        for l in self.leases:
            if l.get("binding_state") in ("ACTIVE", "STATIC") and l.get("address"):
                self.dhcp_ips.add(l["address"])

        self.counts = {
            "A": len(self.a), "PTR": len(self.ptr), "CNAME": len(self.cname),
            "host": len(self.host), "zones": len(self.zones.auth),
            "delegated": len(self.zones.delegated),
        }

    # -- A without PTR ----------------------------------------------------
    def check_a_without_ptr(self):
        """An A record with no PTR is NORMAL in most networks, and reporting
        every one of them is how a tool trains people to ignore it.

        It becomes a finding in exactly one case: the reverse zone exists AND
        this grid authors it. Then the grid could have held the PTR and does
        not. Delegated away, it is invisible. Absent, nothing is missing.
        """
        check = "a-without-ptr"
        for r in self.a:
            if r.get("disable"):
                continue
            v = r.get("view") or "default"
            name = (r.get("name") or "").lower()
            ip = r.get("ipv4addr")
            label = "%s -> %s" % (name, dashed(ip))

            rev = reverse_name(ip)
            if rev is None:
                self.add(check, "record:a", label, v, UNVERIFIABLE,
                         "ipv4addr is missing or not an address this script can parse",
                         "left alone rather than guessed at")
                continue
            existing = self.ptr_by_ip.get((v, ip), [])
            if name in existing:
                continue

            kind, zone = self.zones.status(rev, v)
            if kind == "absent":
                self.add(check, "record:a", label, v, BY_DESIGN,
                         "no PTR, and no reverse zone for this address on the grid",
                         "nothing is missing: there is no zone a PTR could live in")
            elif kind == "delegated":
                self.add(check, "record:a", label, v, UNVERIFIABLE,
                         "the reverse zone %s is DELEGATED away" % zone,
                         "the PTR may well exist on the server you delegated to. This "
                         "grid cannot see it, so this is not a finding")
            elif kind == "disabled":
                self.add(check, "record:a", label, v, SUSPECT,
                         "the reverse zone %s exists but is disabled" % zone,
                         "a disabled zone answers nobody, so the PTR would not resolve "
                         "even after you created it. Fix the zone first")
            elif existing:
                self.add(check, "record:a", label, v, SUSPECT,
                         "the PTR for %s names %s instead" % (ip, existing[0]),
                         "forward and reverse disagree. Decide which name is the real "
                         "one before changing either record")
            else:
                self.add(check, "record:a", label, v, SUSPECT,
                         "no PTR, and you author the reverse zone %s" % zone,
                         "mail servers, logging pipelines and some licence checks do a "
                         "reverse lookup. This is the case worth fixing")

    # -- orphan PTR -------------------------------------------------------
    def check_orphan_ptr(self):
        for p in self.ptr:
            if p.get("disable"):
                continue
            v = p.get("view") or "default"
            target = (p.get("ptrdname") or "").lower()
            ip = p.get("ipv4addr")
            rname = (p.get("name") or reverse_name(ip) or "?")
            label = "%s -> %s" % (rname, dashed(target))

            rkind, rzone = self.zones.status(rname, v)
            if rkind == "delegated":
                self.add("orphan-ptr", "record:ptr", label, v, UNVERIFIABLE,
                         "sits inside %s, a reverse zone this grid delegated away" % rzone,
                         "a PTR in a delegated zone is not orphaned, it is INVISIBLE: "
                         "resolvers follow the delegation and never ask this grid. "
                         "Judge it on the server that answers for the zone")
                continue

            if ip in self.fwd_names.get((v, target), []):
                continue

            fkind, fzone = self.zones.status(target, v)
            if fkind in ("absent", "delegated"):
                where = ("not hosted on this grid" if fkind == "absent"
                         else "inside delegated zone %s" % fzone)
                self.add("orphan-ptr", "record:ptr", label, v, UNVERIFIABLE,
                         "the forward name is %s" % where,
                         "the A record may exist where this grid cannot read it. Check "
                         "against the authoritative server, not against this report")
            elif self.fwd_names.get((v, target)):
                others = ", ".join(dashed(x) for x in self.fwd_names[(v, target)])
                self.add("orphan-ptr", "record:ptr", label, v, SUSPECT,
                         "%s exists but resolves to %s" % (target, others),
                         "the reverse answers with a name that points somewhere else. "
                         "One of the two records was moved and the other left behind")
            elif ip in self.dhcp_ips:
                self.add("orphan-ptr", "record:ptr", label, v, DHCP,
                         "no forward record, and %s is DHCP-managed" % dashed(ip),
                         "a DDNS-created PTR whose A was reaped, or a lease in flight. "
                         "Fix it in DHCP/DDNS: a hand delete of a DDNS record comes back")
            else:
                self.add("orphan-ptr", "record:ptr", label, v, STALE,
                         "nothing in view '%s' is named %s, and you author %s"
                         % (v, target, fzone),
                         "this reverse answers with a name that resolves to nothing")

    # -- dangling CNAME ---------------------------------------------------
    def check_cname(self):
        for c in self.cname:
            if c.get("disable"):
                continue
            v = c.get("view") or "default"
            name = (c.get("name") or "").lower()
            target = (c.get("canonical") or "").lower()
            label = "%s -> %s" % (name, dashed(target))

            if not target:
                self.add("dangling-cname", "record:cname", label, v, STALE,
                         "the canonical name is empty", "there is nothing to point at")
                continue

            if self.chain_loops(v, name):
                self.add("dangling-cname", "record:cname", label, v, STALE,
                         "the CNAME chain loops back on itself",
                         "no resolver ever gets an answer out of this, and nothing in "
                         "the GUI flags it")
                continue

            if self.fwd_names.get((v, target)) or (v, target) in self.cname_names:
                continue

            kind, zone = self.zones.status(target, v)
            takeover = next((s for s in TAKEOVER_SUFFIXES if is_suffix_zone(target, s)), None)

            if kind in ("authoritative", "disabled"):
                self.add("dangling-cname", "record:cname", label, v, STALE,
                         "you author %s and it holds no record for %s" % (zone, target),
                         "a dangling CNAME inside your own zone: the alias returns "
                         "NXDOMAIN today")
            elif self.args.resolve_external:
                state = self.resolve(target)
                if state == "resolves":
                    continue
                if state == "ambiguous":
                    self.add("dangling-cname", "record:cname", label, v, SUSPECT,
                             "the resolver said 'host not found', which on Windows "
                             "means NXDOMAIN *or* a failed lookup"
                             + (" -- and %s is a takeover target" % takeover
                                if takeover else ""),
                             "Windows cannot tell the two apart, so this is reported "
                             "as worth checking rather than as proven dead. Re-run on "
                             "Linux or macOS, or query the authoritative server, to "
                             "settle it")
                elif state == "nxdomain":
                    self.add("dangling-cname", "record:cname", label, v, STALE,
                             "NXDOMAIN from this machine's resolver"
                             + (" and %s is a takeover target" % takeover if takeover else ""),
                             ("whoever registers %s next answers for %s" % (target, name))
                             if takeover else
                             "the alias is live and points at a name nobody owns")
                else:
                    self.add("dangling-cname", "record:cname", label, v, UNVERIFIABLE,
                             "the resolver gave neither an answer nor an NXDOMAIN (%s)" % state,
                             "SERVFAIL or a timeout is a broken lookup, not a missing "
                             "name. Re-run, or ask the authoritative server directly")
            else:
                self.add("dangling-cname", "record:cname", label, v, UNVERIFIABLE,
                         "%s is outside the DNS this grid serves" % target,
                         "'this grid cannot resolve it' is not 'it points at nothing'. "
                         "Pass --resolve-external to ask a real resolver"
                         + (" -- %s is a takeover target, so it is worth doing" % takeover
                            if takeover else ""))

    def chain_loops(self, view, start):
        seen = set()
        cur = start
        while (view, cur) in self.cname_names:
            if cur in seen:
                return True
            seen.add(cur)
            nxt = self.cname_target.get((view, cur))
            if not nxt:
                return False
            if nxt == cur:
                return True
            cur = nxt
        return False

    def resolve(self, name):
        """Ask this machine's resolver, and keep the three outcomes apart.

        NXDOMAIN is a fact. SERVFAIL and a timeout are the absence of one, and
        turning either into 'does not exist' is how a working record ends up
        on a delete list.
        """
        try:
            socket.getaddrinfo(name, None)
            return "resolves"
        except socket.gaierror as exc:
            no_name = {socket.EAI_NONAME}
            nodata = getattr(socket, "EAI_NODATA", None)
            if nodata is not None:
                no_name.add(nodata)
            if exc.errno in no_name:
                # Windows returns WSAHOST_NOT_FOUND (11001) for a real NXDOMAIN
                # AND for a resolver that simply failed, so the two cannot be
                # told apart here. Saying "does not exist" on that evidence is
                # exactly the conflation this script exists to avoid.
                if sys.platform == "win32":
                    return "ambiguous"
                return "nxdomain"
            return "lookup error: %s" % (exc.strerror or exc)
        except OSError as exc:
            return "lookup error: %s" % exc

    # -- idle records -----------------------------------------------------
    def check_idle(self):
        """'Nobody has asked for this name in months' is the closest a DNS
        server comes to evidence of abandonment. It is still not proof, and it
        fails asymmetrically:

          last_queried empty  the grid may not be recording it at all
          last_queried old    or this member is simply not the one being asked
        """
        limit = self.args.idle_days
        rows = ([("record:a", r, (r.get("name") or "").lower(), r.get("ipv4addr"))
                 for r in self.a]
                + [("record:cname", c, (c.get("name") or "").lower(), None)
                   for c in self.cname])
        for kind, r, name, ip in rows:
            if r.get("disable"):
                continue
            v = r.get("view") or "default"
            idle = age_days(r.get("last_queried"))
            label = "%s%s" % (name, " -> %s" % ip if ip else "")

            if idle is None:
                self.add("idle-record", kind, label, v, UNVERIFIABLE,
                         "last_queried is empty, which is not the same as zero",
                         "either nothing has ever queried it, or this grid does not "
                         "record query timestamps. Reported as '-', never as 0 days")
                continue
            if idle < limit:
                continue

            if ip and ip in self.dhcp_ips:
                self.add("idle-record", kind, label, v, DHCP,
                         "idle %d days, and %s is DHCP-managed" % (idle, ip),
                         "a reservation or a live lease owns this address. Deleting the "
                         "record breaks DDNS bookkeeping and it may simply reappear")
                continue

            if ip and self.args.probe:
                state = self.probes.get(ip) or probe(ip, self.args.probe_ports,
                                                     self.args.probe_timeout)
                self.probes[ip] = state
                if state == "alive":
                    self.add("idle-record", kind, label, v, BY_DESIGN,
                             "idle %d days, but %s answered a probe" % (idle, ip),
                             "the host is up. Nobody resolving the name is a monitoring "
                             "or documentation gap, not a dead record")
                    continue
                self.add("idle-record", kind, label, v, UNVERIFIABLE,
                         "idle %d days and %s did not answer any probe" % (idle, ip),
                         "NO ANSWER IS NOT DEAD. ICMP and TCP are filtered on purpose "
                         "in most networks and a host can be off for a maintenance "
                         "window. This is the check that must never become a delete")
                continue

            self.add("idle-record", kind, label, v, SUSPECT,
                     "nothing has queried this name in %d days" % idle,
                     "last_queried resets when DNS restarts and counts only queries "
                     "this member answered. Confirm with --probe, or with the owner, "
                     "before deleting anything")

    def run(self):
        self.fetch()
        self.check_a_without_ptr()
        self.check_orphan_ptr()
        self.check_cname()
        self.check_idle()
        return self.findings

    @staticmethod
    def unchecked():
        return [
            "Anything in a DNS view you excluded with --view, and zones served "
            "by a grid this Grid Master does not manage",
            "IPv6: AAAA records and ip6.arpa reverse zones are out of scope",
            "MX, SRV, NS and TXT targets -- a dangling MX is the same class of "
            "bug and is not looked at here",
            "Whether a name is still referenced by a certificate, a load "
            "balancer pool, or a hard-coded string in an application",
            "DHCP ranges and unused address space -- that is "
            "infoblox-unused-ip-space",
        ]


# ---------------------------------------------------------------- reporting
TITLES = [
    (STALE, "STALE -- broken on this grid's own data"),
    (SUSPECT, "WORTH A LOOK -- real findings, each with its caveat"),
    (UNVERIFIABLE, "COULD NOT VERIFY -- NOT findings. Nothing here was proved bad"),
    (DHCP, "DHCP-MANAGED -- fix in DHCP/DDNS, never by hand"),
    (BY_DESIGN, "BY DESIGN -- looked wrong, is not"),
]


def report(audit, args):
    buckets = {}
    for f in sorted(audit.findings,
                    key=lambda f: (SEVERITY[f.verdict], f.view, f.check, f.name)):
        buckets.setdefault(f.verdict, []).append(f)

    line = "=" * 74
    print(line)
    print("infoblox-stale-dns-records            NIOS / WAPI %s" % args.wapi_version)
    print(line)
    print("Grid        : %s" % (args.offline_dir or args.host))
    print("DNS views   : %s" % (args.view or "all"))
    c = audit.counts
    print("Read        : %d A, %d PTR, %d CNAME, %d host, %d zone(s) authored, "
          "%d delegated" % (c["A"], c["PTR"], c["CNAME"], c["host"],
                            c["zones"], c["delegated"]))
    print("Idle after  : %d days without a query" % args.idle_days)
    print("Probing     : %s" % ("TCP %s, %.1fs per port" % (
        ",".join(str(p) for p in args.probe_ports), args.probe_timeout)
        if args.probe else "off -- enable with --probe"))
    print("")

    for verdict, title in TITLES:
        rows = buckets.get(verdict, [])
        if verdict == BY_DESIGN and not args.show_ok:
            continue
        if not rows:
            continue
        print("%s (%d)" % (title, len(rows)))
        print("-" * 74)
        for f in rows[:args.limit]:
            print("  [%s] %s  (view: %s)" % (f.check, f.name, f.view))
            print("      %s" % f.reason)
            if f.detail:
                print("      %s" % f.detail)
        if len(rows) > args.limit:
            print("  ... %d more (raise --limit, or read the CSV)" % (len(rows) - args.limit))
        print("")

    actionable = len(buckets.get(STALE, [])) + len(buckets.get(SUSPECT, []))
    print("SUMMARY")
    print("-" * 74)
    print("  %d to act on, %d could not be verified, %d DHCP-managed, %d by design."
          % (actionable, len(buckets.get(UNVERIFIABLE, [])),
             len(buckets.get(DHCP, [])), len(buckets.get(BY_DESIGN, []))))
    print("  COULD NOT VERIFY is a separate list on purpose. A record this script")
    print("  could not check and a record it proved broken are different facts,")
    print("  and merging them is how a live record gets deleted.")
    print("")
    print("NOT CHECKED BY THIS SCRIPT")
    print("-" * 74)
    for u in audit.unchecked():
        print("  - %s" % u)
    print("")
    print("This script never deletes or edits a record. Its job is to hand you a "
          "shortlist you can argue with.")


def write_csv(path, findings):
    with open(path, "w", encoding="utf-8-sig", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(["verdict", "check", "record_type", "record", "dns_view",
                    "reason", "detail"])
        for f in sorted(findings, key=lambda x: (SEVERITY[x.verdict], x.view, x.name)):
            w.writerow([f.verdict, f.check, f.kind, f.name, f.view, f.reason, f.detail])


# ---------------------------------------------------------------- entry
def parse_args(argv):
    p = argparse.ArgumentParser(
        description="Find stale DNS technical debt on an Infoblox NIOS grid over "
                    "WAPI v2.x: A records with no PTR, orphaned PTRs, dangling "
                    "CNAMEs and names nobody has queried in months. Read-only.")
    p.add_argument("--host", help="Grid Master hostname or IP. Needs IB_USER and IB_PASS")
    p.add_argument("--wapi-version", default=DEFAULT_WAPI,
                   help="WAPI version path segment. Default %s" % DEFAULT_WAPI)
    p.add_argument("--view", help="Limit the report to one DNS view. Default: all")
    p.add_argument("--insecure", action="store_true",
                   help="Skip TLS verification. Only for a certificate you recognise")
    p.add_argument("--idle-days", type=int, default=180,
                   help="last_queried older than this is suspect. Default 180")
    p.add_argument("--leases", action="store_true",
                   help="Also read the lease table, so an address with a live lease "
                        "counts as DHCP-managed. One more query, and a large one")
    p.add_argument("--probe", action="store_true",
                   help="TCP-probe the address behind an idle record. An answer "
                        "proves it is alive; silence proves NOTHING and lands in "
                        "COULD NOT VERIFY, never in the findings")
    p.add_argument("--probe-ports", default="443,22,80,3389,445",
                   help="Ports to try, in order. Default 443,22,80,3389,445")
    p.add_argument("--probe-timeout", type=float, default=1.0,
                   help="Seconds per port. Default 1.0")
    p.add_argument("--resolve-external", action="store_true",
                   help="Ask this machine's resolver about a CNAME target outside "
                        "the grid. NXDOMAIN becomes a finding; SERVFAIL or a "
                        "timeout stays in COULD NOT VERIFY")
    p.add_argument("--show-ok", action="store_true",
                   help="Also print the BY DESIGN section")
    p.add_argument("--limit", type=int, default=50,
                   help="Rows printed per section. Default 50")
    p.add_argument("--csv", help="Write every record and verdict to a CSV")
    p.add_argument("--save-dir", help="Save every WAPI response here for later offline runs")
    p.add_argument("--offline-dir", help="Replay a --save-dir. No network, no credentials")
    args = p.parse_args(argv)
    if bool(args.host) == bool(args.offline_dir):
        p.error("pass exactly one of --host or --offline-dir")
    try:
        args.probe_ports = [int(x) for x in str(args.probe_ports).split(",") if x.strip()]
    except ValueError:
        p.error("--probe-ports takes a comma-separated list of port numbers")
    if not args.probe_ports:
        p.error("--probe-ports is empty")
    return args


def main(argv=None):
    args = parse_args(argv)
    wapi = Wapi(host=args.host, version=args.wapi_version, insecure=args.insecure,
                offline_dir=args.offline_dir, save_dir=args.save_dir)
    audit = Audit(wapi, args)
    audit.run()
    if args.view:
        audit.findings = [f for f in audit.findings if f.view == args.view]
    report(audit, args)
    if args.csv:
        write_csv(args.csv, audit.findings)
        print("\nCSV written: %s" % args.csv)
    return 1 if any(f.verdict in (STALE, SUSPECT) for f in audit.findings) else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.stderr.write("\ninterrupted\n")
        sys.exit(2)
    except SystemExit:
        raise
    except requests.exceptions.SSLError as exc:
        sys.stderr.write("TLS verification failed: %s\n(use --insecure only for a "
                         "certificate you recognise)\n" % exc)
        sys.exit(2)
    except requests.exceptions.RequestException as exc:
        sys.stderr.write("could not reach the Grid Master: %s\n" % exc)
        sys.exit(2)