Infoblox Bulk DNS Record Creator

#!/usr/bin/env python3
"""
Infoblox Bulk DNS Record Creator

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/infoblox-bulk-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"
"""

# NOT read-only. With --apply this script CREATES and can UPDATE DNS records
# on the grid. Without --apply it writes nothing: it reads, plans and prints.
#
# EXIT CODES  0 plan clean / every write succeeded
#             1 something needs a human: blocked rows, conflicts, failed writes
#             2 the run could not proceed at all
#
# 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 re
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"

# ---------------------------------------------------------------- outcomes
CREATE = "CREATE"
UPDATE = "UPDATE"
SKIP = "SKIP"
BLOCKED = "BLOCKED"

TYPES = ("a", "ptr", "cname", "host")
WAPI_OBJECT = {"a": "record:a", "ptr": "record:ptr",
               "cname": "record:cname", "host": "record:host"}

LABEL_RE = re.compile(r"^(?!-)[A-Za-z0-9_-]{1,63}(?<!-)$")


# ---------------------------------------------------------------- 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

    # -- the write half. Everything above this line is shared verbatim with
    # -- the read-only Infoblox scripts; only this part can change a grid.
    def post(self, obj, payload):
        """Create one object. Returns (ok, ref_or_error).

        One object per call on purpose. WAPI does have a /request endpoint
        that batches, but a batch fails as a unit and leaves you guessing
        which row did it. One call per row costs round trips and buys an
        exact rollback file.
        """
        if self.offline_dir:
            return True, "SIMULATED/%s/%s" % (obj, payload.get("name")
                                              or payload.get("ptrdname"))
        try:
            resp = self.session.post(self.base + obj, json=payload, timeout=60)
        except requests.exceptions.RequestException as exc:
            return False, str(exc)
        if resp.status_code in (200, 201):
            return True, resp.json()     # WAPI returns the new _ref, as a string
        return False, self._error(resp)

    def put(self, ref, payload):
        """Change one existing object in place. Returns (ok, ref_or_error)."""
        if self.offline_dir:
            return True, "SIMULATED/" + ref
        try:
            resp = self.session.put(self.base + ref, json=payload, timeout=60)
        except requests.exceptions.RequestException as exc:
            return False, str(exc)
        if resp.status_code == 200:
            return True, resp.json()
        return False, self._error(resp)

    @staticmethod
    def _error(resp):
        """WAPI puts the useful sentence in a JSON 'text' field and the useless
        one in the HTTP reason. Prefer the former, fall back to the body."""
        try:
            body = resp.json()
            return "%s: %s" % (resp.status_code, body.get("text") or body)
        except ValueError:
            return "%s: %s" % (resp.status_code, resp.text[:200])


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


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


# ---------------------------------------------------------------- the CSV
def valid_fqdn(name):
    """Syntax only. A name can be perfectly well formed and still have no zone
    on this grid -- that is a separate check, and it needs the grid."""
    name = (name or "").rstrip(".")
    if not name or len(name) > 253 or "." not in name:
        return False
    return all(LABEL_RE.match(label) for label in name.split("."))


class Row(object):
    """One line of the CSV, and everything decided about it afterwards."""

    def __init__(self, line, rtype, name, value, view, ttl, comment, want_ptr):
        self.line = line
        self.type = rtype
        self.name = name
        self.value = value
        self.view = view
        self.ttl = ttl
        self.comment = comment
        self.want_ptr = want_ptr
        self.ops = []           # Op objects, one row can produce two

    def label(self):
        if self.type == "ptr":
            return "%s -> %s" % (self.name, self.value)
        return "%s -> %s" % (self.name, self.value)


class Op(object):
    """One WAPI object this run would create or change."""

    def __init__(self, row, obj, verdict, reason, payload=None, ref=None,
                 previous=None, derived=False):
        self.row = row
        self.obj = obj              # record:a / record:ptr / ...
        self.verdict = verdict      # CREATE / UPDATE / SKIP / BLOCKED
        self.reason = reason
        self.payload = payload or {}
        self.ref = ref              # existing _ref, for UPDATE
        self.previous = previous    # value before an UPDATE, for the rollback
        self.derived = derived      # True for a PTR this script added itself
        self.result = None          # filled in by apply()
        self.created_ref = None

    def what(self):
        if self.obj == "record:ptr":
            return "%s -> %s" % (self.payload.get("ipv4addr", self.row.name),
                                 self.payload.get("ptrdname", self.row.value))
        return "%s -> %s" % (self.payload.get("name", self.row.name),
                             self.payload.get("ipv4addr")
                             or self.payload.get("canonical")
                             or self.row.value)


def read_csv(path, default_view):
    """Parse and validate the whole file before anything touches the grid.

    Every row is checked, not just up to the first bad one: a user who fixes
    one typo per run will give up before the file is clean. Failing at row
    3000 with 2999 records already written is the worst possible outcome, and
    it is why this function exists before the transport is ever used.
    """
    if not os.path.exists(path):
        raise SystemExit("CSV not found: %s" % path)
    with open(path, encoding="utf-8-sig", newline="") as fh:
        reader = csv.DictReader(fh)
        if not reader.fieldnames:
            raise SystemExit("%s is empty -- it needs a header row." % path)
        headers = [(h or "").strip().lower() for h in reader.fieldnames]
        missing = [h for h in ("type", "name", "value") if h not in headers]
        if missing:
            raise SystemExit(
                "CSV header is missing: %s\n"
                "Required columns: type,name,value. Optional: view,ttl,comment,"
                "create_ptr" % ", ".join(missing))
        rows, errors = [], []
        for n, raw in enumerate(reader, start=2):
            clean = {}
            for key, val in raw.items():
                if key is None:
                    continue
                clean[(key or "").strip().lower()] = (val or "").strip()
            if not any(clean.get(k) for k in ("type", "name", "value")):
                continue    # a blank line, not an error
            rtype = clean.get("type", "").lower()
            name = clean.get("name", "").rstrip(".")
            value = clean.get("value", "").rstrip(".")
            view = clean.get("view") or default_view
            ttl = clean.get("ttl") or ""
            comment = clean.get("comment") or ""
            want_ptr = clean.get("create_ptr", "").lower() in ("yes", "true", "1", "y")

            if rtype not in TYPES:
                errors.append((n, "type '%s' is not one of: %s"
                               % (dashed(rtype), ", ".join(TYPES))))
                continue
            if rtype == "ptr":
                try:
                    ipaddress.ip_address(name)
                except ValueError:
                    errors.append((n, "for a ptr row, name must be the IP address "
                                      "('%s' is not an address)" % dashed(name)))
                    continue
                if not valid_fqdn(value):
                    errors.append((n, "ptr value '%s' is not a valid FQDN" % dashed(value)))
                    continue
            else:
                if not valid_fqdn(name):
                    errors.append((n, "name '%s' is not a valid FQDN (needs at least "
                                      "one dot, labels of 1-63 letters, digits, "
                                      "'-' or '_')" % dashed(name)))
                    continue
            if rtype in ("a", "host"):
                try:
                    addr = ipaddress.ip_address(value)
                except ValueError:
                    errors.append((n, "value '%s' is not an IP address" % dashed(value)))
                    continue
                if addr.version != 4:
                    errors.append((n, "this script writes IPv4 only; '%s' is IPv6 "
                                      "(record:aaaa is not implemented)" % value))
                    continue
            if rtype == "cname":
                if not valid_fqdn(value):
                    errors.append((n, "cname target '%s' is not a valid FQDN" % dashed(value)))
                    continue
                if name.lower() == value.lower():
                    errors.append((n, "CNAME points at itself (%s) -- a resolver "
                                      "follows that until it gives up" % name))
                    continue
            if ttl:
                if not ttl.isdigit() or int(ttl) > 2147483647:
                    errors.append((n, "ttl '%s' is not a whole number of seconds" % ttl))
                    continue
                ttl = int(ttl)
            else:
                ttl = None
            if want_ptr and rtype in ("ptr", "cname"):
                errors.append((n, "create_ptr means nothing on a %s row -- remove it, "
                                  "so nobody reads this file and believes a PTR was "
                                  "requested" % rtype))
                continue
            rows.append(Row(n, rtype, name, value, view, ttl, comment, want_ptr))

    errors.extend(cross_row_errors(rows))
    return rows, sorted(errors)


def cross_row_errors(rows):
    """Contradictions BETWEEN rows. The single-row checks cannot see these,
    and they are the ones that produce half-applied files."""
    errors = []
    seen = {}
    forward = {}
    cnames = {}
    for r in rows:
        key = (r.view, r.type, r.name.lower())
        if key in seen:
            errors.append((r.line, "same %s for %s in view '%s' already on line %d"
                           % (r.type, r.name, r.view, seen[key])))
            continue
        seen[key] = r.line
        if r.type in ("a", "host"):
            forward.setdefault((r.view, r.name.lower()), []).append(r)
        if r.type == "cname":
            cnames[(r.view, r.name.lower())] = r
    for key, rs in forward.items():
        kinds = set(r.type for r in rs)
        if len(kinds) > 1:
            lines = ", ".join(str(r.line) for r in sorted(rs, key=lambda x: x.line))
            errors.append((max(r.line for r in rs),
                           "%s in view '%s' is both an 'a' row and a 'host' row "
                           "(lines %s). They are different objects in NIOS and "
                           "cannot share a name" % (key[1], key[0], lines)))
    for key, c in cnames.items():
        if key in forward:
            errors.append((c.line, "%s in view '%s' is a CNAME here and an address "
                                   "record on line %d. A CNAME cannot coexist with "
                                   "other data for the same name"
                           % (key[1], key[0], forward[key][0].line)))
    return errors


# ---------------------------------------------------------------- planning
class Planner(object):
    def __init__(self, wapi, args):
        self.w = wapi
        self.args = args
        self.counts = {}
        self.views = []

    def fetch(self):
        """Read the current state once and index it. Asking WAPI 'does this
        name exist' per row is one round trip per row; on a 4000-row file that
        is the difference between a minute and an afternoon."""
        w = self.w
        self.views = [v.get("name") for v in
                      w.get("view", {"_return_fields": "name,is_default"})]
        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,ttl,comment,disable"})
        self.ptr = w.get("record:ptr", {"_return_fields":
                         "ptrdname,ipv4addr,name,view,disable"})
        self.cname = w.get("record:cname", {"_return_fields":
                           "name,canonical,view,disable"})
        self.host = w.get("record:host", {"_return_fields":
                          "name,ipv4addrs,view,configure_for_dns,disable"})

        self.a_by_name = {}        # (view, name) -> [record]
        self.host_by_name = {}     # (view, name) -> record
        self.cname_by_name = {}    # (view, name) -> record
        self.ptr_by_ip = {}        # (view, ip)   -> [record]
        for r in self.a:
            self.a_by_name.setdefault(
                (r.get("view") or "default", (r.get("name") or "").lower()), []).append(r)
        for h in self.host:
            self.host_by_name[(h.get("view") or "default",
                               (h.get("name") or "").lower())] = h
        for c in self.cname:
            self.cname_by_name[(c.get("view") or "default",
                                (c.get("name") or "").lower())] = c
        for p in self.ptr:
            self.ptr_by_ip.setdefault(
                (p.get("view") or "default", p.get("ipv4addr")), []).append(p)
        self.counts = {"views": len(self.views), "zones": len(self.zones.auth),
                       "A": len(self.a), "PTR": len(self.ptr),
                       "CNAME": len(self.cname), "host": len(self.host)}

    # -- per row ----------------------------------------------------------
    def plan(self, rows):
        for row in rows:
            if self.views and row.view not in self.views:
                row.ops.append(Op(row, WAPI_OBJECT[row.type], BLOCKED,
                                  "DNS view '%s' does not exist on this grid "
                                  "(views here: %s)"
                                  % (row.view, ", ".join(sorted(self.views)))))
                continue
            getattr(self, "_plan_" + row.type)(row)
        self.dedupe(rows)
        return rows

    def _zone_ok(self, row, name, what):
        """The one check that stops a row creating a record nobody can resolve."""
        status, zone = self.zones.status(name, row.view)
        if status == "authoritative":
            return True, zone
        msgs = {
            "absent": "no zone on this grid holds %s in view '%s'" % (name, row.view),
            "delegated": "%s falls inside %s, which this grid DELEGATES away -- "
                         "the record would exist here and never be served" % (name, zone),
            "disabled": "the zone %s exists in view '%s' but is disabled"
                        % (zone, row.view),
        }
        row.ops.append(Op(row, what, BLOCKED, msgs[status]))
        return False, zone

    def _plan_a(self, row):
        ok, _ = self._zone_ok(row, row.name, "record:a")
        if not ok:
            return
        key = (row.view, row.name.lower())
        if key in self.cname_by_name:
            row.ops.append(Op(row, "record:a", BLOCKED,
                              "a CNAME for %s already exists in view '%s'; a name "
                              "cannot be both" % (row.name, row.view)))
            return
        if key in self.host_by_name:
            row.ops.append(Op(row, "record:a", BLOCKED,
                              "a host record named %s already exists in view '%s'. "
                              "record:host and record:a are different objects and "
                              "NIOS will not let them share the name"
                              % (row.name, row.view)))
            return
        payload = {"name": row.name, "ipv4addr": row.value, "view": row.view}
        if row.comment:
            payload["comment"] = row.comment
        if row.ttl is not None:
            payload["ttl"] = row.ttl
            payload["use_ttl"] = True
        existing = self.a_by_name.get(key, [])
        same = [r for r in existing if r.get("ipv4addr") == row.value]
        if same:
            row.ops.append(Op(row, "record:a", SKIP,
                              "already exists with the same address", payload,
                              ref=same[0].get("_ref")))
        elif existing:
            other = ", ".join(sorted(r.get("ipv4addr") or "-" for r in existing))
            self._conflict(row, "record:a", payload, existing[0], other,
                           {"ipv4addr": row.value}, row.value)
        else:
            row.ops.append(Op(row, "record:a", CREATE, "new name", payload))
        self._plan_derived_ptr(row)

    def _plan_derived_ptr(self, row):
        """An A record does NOT bring a PTR with it. In NIOS the reverse entry
        is a separate record:ptr object, and the GUI's 'create associated PTR'
        tickbox is the GUI doing exactly this second call for you. Over WAPI
        nobody does it for you, which is the single most common reason a bulk
        load leaves a zone with forward-only names."""
        if not row.want_ptr:
            return
        rev = ipaddress.ip_address(row.value).reverse_pointer
        status, zone = self.zones.status(rev, row.view)
        if status != "authoritative":
            row.ops.append(Op(row, "record:ptr", BLOCKED,
                              "cannot create the PTR: the reverse zone for %s is %s "
                              "in view '%s'%s. The A record above is unaffected"
                              % (row.value, status, row.view,
                                 " (%s)" % zone if zone else ""),
                              derived=True))
            return
        payload = {"ptrdname": row.name, "ipv4addr": row.value, "view": row.view}
        if row.comment:
            payload["comment"] = row.comment
        existing = self.ptr_by_ip.get((row.view, row.value), [])
        same = [p for p in existing
                if (p.get("ptrdname") or "").lower() == row.name.lower()]
        if same:
            row.ops.append(Op(row, "record:ptr", SKIP,
                              "PTR already points here", payload,
                              ref=same[0].get("_ref"), derived=True))
        elif existing:
            other = ", ".join(sorted(p.get("ptrdname") or "-" for p in existing))
            self._conflict(row, "record:ptr", payload, existing[0], other,
                           {"ptrdname": row.name}, row.name, derived=True)
        else:
            row.ops.append(Op(row, "record:ptr", CREATE,
                              "reverse zone %s is authoritative here" % zone,
                              payload, derived=True))

    def _plan_ptr(self, row):
        rev = ipaddress.ip_address(row.name).reverse_pointer
        ok, zone = self._zone_ok(row, rev, "record:ptr")
        if not ok:
            return
        payload = {"ptrdname": row.value, "ipv4addr": row.name, "view": row.view}
        if row.comment:
            payload["comment"] = row.comment
        if row.ttl is not None:
            payload["ttl"] = row.ttl
            payload["use_ttl"] = True
        existing = self.ptr_by_ip.get((row.view, row.name), [])
        same = [p for p in existing
                if (p.get("ptrdname") or "").lower() == row.value.lower()]
        if same:
            row.ops.append(Op(row, "record:ptr", SKIP, "already points here",
                              payload, ref=same[0].get("_ref")))
        elif existing:
            other = ", ".join(sorted(p.get("ptrdname") or "-" for p in existing))
            self._conflict(row, "record:ptr", payload, existing[0], other,
                           {"ptrdname": row.value}, row.value)
        else:
            row.ops.append(Op(row, "record:ptr", CREATE,
                              "reverse zone %s is authoritative here" % zone, payload))

    def _plan_cname(self, row):
        ok, _ = self._zone_ok(row, row.name, "record:cname")
        if not ok:
            return
        key = (row.view, row.name.lower())
        clash = None
        if key in self.a_by_name:
            clash = "an A record"
        elif key in self.host_by_name:
            clash = "a host record"
        if clash:
            row.ops.append(Op(row, "record:cname", BLOCKED,
                              "%s named %s already exists in view '%s'. A CNAME "
                              "cannot coexist with other data for the same name"
                              % (clash, row.name, row.view)))
            return
        payload = {"name": row.name, "canonical": row.value, "view": row.view}
        if row.comment:
            payload["comment"] = row.comment
        if row.ttl is not None:
            payload["ttl"] = row.ttl
            payload["use_ttl"] = True
        existing = self.cname_by_name.get(key)
        if existing and (existing.get("canonical") or "").lower() == row.value.lower():
            row.ops.append(Op(row, "record:cname", SKIP,
                              "already points at the same target", payload,
                              ref=existing.get("_ref")))
        elif existing:
            self._conflict(row, "record:cname", payload, existing,
                           existing.get("canonical") or "-", {"canonical": row.value},
                           row.value)
        else:
            row.ops.append(Op(row, "record:cname", CREATE, "new alias", payload))

    def _plan_host(self, row):
        ok, _ = self._zone_ok(row, row.name, "record:host")
        if not ok:
            return
        key = (row.view, row.name.lower())
        if key in self.cname_by_name:
            row.ops.append(Op(row, "record:host", BLOCKED,
                              "a CNAME for %s already exists in view '%s'; a name "
                              "cannot be both" % (row.name, row.view)))
            return
        if key in self.a_by_name:
            row.ops.append(Op(row, "record:host", BLOCKED,
                              "an A record named %s already exists in view '%s'. "
                              "A host record is not an A record -- NIOS stores them "
                              "separately and refuses the overlap"
                              % (row.name, row.view)))
            return
        payload = {"name": row.name, "view": row.view,
                   "ipv4addrs": [{"ipv4addr": row.value}],
                   "configure_for_dns": True}
        if row.comment:
            payload["comment"] = row.comment
        if row.ttl is not None:
            payload["ttl"] = row.ttl
            payload["use_ttl"] = True
        existing = self.host_by_name.get(key)
        if existing:
            addrs = [a.get("ipv4addr") for a in existing.get("ipv4addrs") or []]
            if row.value in addrs:
                row.ops.append(Op(row, "record:host", SKIP,
                                  "host record already has this address", payload,
                                  ref=existing.get("_ref")))
            else:
                self._conflict(row, "record:host", payload, existing,
                               ", ".join(sorted(a or "-" for a in addrs)),
                               {"ipv4addrs": [{"ipv4addr": row.value}]}, row.value)
        else:
            row.ops.append(Op(row, "record:host", CREATE, "new host record", payload))
        if row.want_ptr:
            self._host_ptr_note(row)

    def _host_ptr_note(self, row):
        """A host record does not need a record:ptr -- NIOS answers the reverse
        from the host object itself, provided the reverse zone exists here. So
        create_ptr on a host row creates nothing; it only checks the zone, and
        says so rather than silently ignoring the column."""
        rev = ipaddress.ip_address(row.value).reverse_pointer
        status, zone = self.zones.status(rev, row.view)
        if status == "authoritative":
            row.ops.append(Op(row, "record:host", SKIP,
                              "create_ptr: nothing to do. A host record answers the "
                              "reverse itself, and %s is authoritative here" % zone,
                              derived=True))
        else:
            row.ops.append(Op(row, "record:host", BLOCKED,
                              "create_ptr: the reverse zone for %s is %s in view "
                              "'%s', so this host will have no reverse. The host "
                              "record itself is unaffected" % (row.value, status, row.view),
                              derived=True))

    def _conflict(self, row, obj, payload, existing, current, patch, wanted,
                  derived=False):
        """The same name with a DIFFERENT value. This is the normal case in a
        bulk load, not the exception, and the policy has to be explicit:
        skipping silently leaves the CSV and the grid disagreeing, updating
        silently repoints production."""
        policy = self.args.on_conflict
        ref = existing.get("_ref")
        if policy == "skip":
            row.ops.append(Op(row, obj, SKIP,
                              "CONFLICT: exists with %s, wanted %s -- left alone "
                              "(--on-conflict skip)" % (current, wanted),
                              payload, ref=ref, derived=derived))
        elif policy == "update":
            row.ops.append(Op(row, obj, UPDATE,
                              "CONFLICT: %s -> %s" % (current, wanted),
                              patch, ref=ref, previous=current, derived=derived))
        else:   # abort
            row.ops.append(Op(row, obj, BLOCKED,
                              "CONFLICT: exists with %s, CSV says %s "
                              "(--on-conflict abort)" % (current, wanted),
                              payload, ref=ref, derived=derived))

    def dedupe(self, rows):
        """Two CSV rows -- or one row and the PTR another row asked for -- can
        aim at the same object. WAPI would accept the first and reject the
        second, so the plan has to show it as one write, not two.

        This cannot be caught while reading the CSV: an 'a' row with
        create_ptr=yes and a 'ptr' row are different types, different columns
        and the same record:ptr object.
        """
        first = {}
        for row in rows:
            for op in row.ops:
                if op.verdict not in (CREATE, UPDATE):
                    continue
                if op.obj == "record:ptr":
                    key = (op.obj, row.view, op.payload.get("ipv4addr"),
                           (op.payload.get("ptrdname") or "").lower())
                else:
                    key = (op.obj, row.view, (op.payload.get("name") or "").lower())
                if key in first:
                    op.verdict = SKIP
                    op.reason = ("the same object is already planned by line %d%s "
                                 "-- one write, not two"
                                 % (first[key].row.line,
                                    " (its create_ptr)" if first[key].derived else ""))
                else:
                    first[key] = op


# ---------------------------------------------------------------- applying
class Rollback(object):
    """One JSON object per line, flushed after every write.

    Flushed, not buffered: the run that needs this file most is the one that
    died halfway, and a buffered file loses exactly the records nobody knows
    about.
    """

    def __init__(self, path):
        self.path = path
        self.fh = open(path, "w", encoding="utf-8")
        self.count = 0

    def record(self, op, ref):
        entry = {
            "op": "created" if op.verdict == CREATE else "updated",
            "wapi_object": op.obj,
            "ref": ref,
            "what": op.what(),
            "view": op.row.view,
            "csv_line": op.row.line,
            "previous_value": op.previous,
            "at": datetime.now(timezone.utc).isoformat(),
        }
        self.fh.write(json.dumps(entry) + "\n")
        self.fh.flush()
        self.count += 1

    def close(self):
        self.fh.close()


def apply_plan(wapi, ops, rollback):
    done, failed = 0, 0
    for op in ops:
        if op.verdict == CREATE:
            ok, res = wapi.post(op.obj, op.payload)
        else:
            ok, res = wapi.put(op.ref, op.payload)
        if ok:
            op.result = "ok"
            op.created_ref = res if isinstance(res, str) else str(res)
            rollback.record(op, op.created_ref)
            done += 1
        else:
            op.result = "failed: %s" % res
            failed += 1
    return done, failed


# ---------------------------------------------------------------- reporting
def report_plan(planner, rows, args, errors):
    line = "=" * 74
    print(line)
    print("infoblox-bulk-dns-records             NIOS / WAPI %s" % args.wapi_version)
    print(line)
    print("Grid        : %s%s" % (args.offline_dir or args.host,
                                  "  (offline replay)" if args.offline_dir else ""))
    print("CSV         : %s" % args.csv)
    print("Mode        : %s" % ("APPLY -- this run WRITES to the grid" if args.apply
                                else "DRY RUN -- nothing will be written"))
    print("Conflicts   : --on-conflict %s" % args.on_conflict)
    print("Default view: %s" % args.view)
    c = planner.counts
    print("Grid holds  : %d A, %d PTR, %d CNAME, %d host, %d zone(s), %d view(s)"
          % (c["A"], c["PTR"], c["CNAME"], c["host"], c["zones"], c["views"]))
    print("")

    if errors:
        print("CSV REJECTED -- %d row(s) did not survive validation" % len(errors))
        print("-" * 74)
        for ln, msg in errors[:args.limit]:
            print("  line %-5d %s" % (ln, msg))
        if len(errors) > args.limit:
            print("  ... %d more (raise --limit)" % (len(errors) - args.limit))
        print("")
        print("  Nothing was read from the grid and nothing was written. The whole")
        print("  file is validated before the first call on purpose: stopping at")
        print("  row 3000 with 2999 records already created is the worst outcome")
        print("  this script can have.")
        return

    ops = [op for row in rows for op in row.ops]
    buckets = {}
    for op in ops:
        buckets.setdefault(op.verdict, []).append(op)

    titles = [
        (BLOCKED, "BLOCKED -- will not be attempted"),
        (UPDATE, "UPDATE -- an existing record will be CHANGED"),
        (CREATE, "CREATE -- new records"),
        (SKIP, "SKIP -- already correct, or left alone"),
    ]
    for verdict, title in titles:
        got = buckets.get(verdict, [])
        if not got:
            continue
        if verdict == SKIP and not args.show_skipped:
            continue
        print("%s (%d)" % (title, len(got)))
        print("-" * 74)
        for op in got[:args.limit]:
            print("  line %-5d %-12s %s%s"
                  % (op.row.line, op.obj, op.what(),
                     "   [derived PTR]" if op.derived and op.obj == "record:ptr" else ""))
            print("      %s" % op.reason)
        if len(got) > args.limit:
            print("  ... %d more (raise --limit)" % (len(got) - args.limit))
        print("")

    print("THE PLAN")
    print("-" * 74)
    by_obj = {}
    for op in ops:
        by_obj.setdefault(op.obj, {}).setdefault(op.verdict, 0)
        by_obj[op.obj][op.verdict] += 1
    print("  %-14s %8s %8s %8s %8s" % ("object", "create", "update", "skip", "blocked"))
    for obj in sorted(by_obj):
        b = by_obj[obj]
        print("  %-14s %8d %8d %8d %8d"
              % (obj, b.get(CREATE, 0), b.get(UPDATE, 0), b.get(SKIP, 0),
                 b.get(BLOCKED, 0)))
    writes = len(buckets.get(CREATE, [])) + len(buckets.get(UPDATE, []))
    print("  %-14s %8d %8d %8d %8d"
          % ("TOTAL", len(buckets.get(CREATE, [])), len(buckets.get(UPDATE, [])),
             len(buckets.get(SKIP, [])), len(buckets.get(BLOCKED, []))))
    print("")
    print("  %d row(s) in the CSV, %d object(s) would be written." % (len(rows), writes))
    return writes


def report_apply(ops, rollback, args):
    ok = [o for o in ops if o.result == "ok"]
    bad = [o for o in ops if o.result and o.result != "ok"]
    print("")
    print("APPLY")
    print("-" * 74)
    if args.offline_dir:
        print("  SIMULATED. --offline-dir has no grid behind it, so every write was")
        print("  faked and NOTHING was created. This path exists to exercise the")
        print("  plan end to end; it is not a record of a change.")
    for op in bad[:args.limit]:
        print("  line %-5d %-12s %s" % (op.row.line, op.obj, op.what()))
        print("      %s" % op.result)
    print("  %d written, %d failed." % (len(ok), len(bad)))
    if rollback and rollback.count and not args.offline_dir:
        print("  Rollback file: %s  (%d entry/entries)" % (rollback.path, rollback.count))
        print("  It holds the _ref of everything this run touched. That is what you")
        print("  need to undo it -- DNS has no transaction and no 'back' button.")


# ---------------------------------------------------------------- entry
def parse_args(argv):
    p = argparse.ArgumentParser(
        description="Bulk-create A, PTR, CNAME and host records on an Infoblox "
                    "NIOS grid from a CSV, over WAPI v2.x. Dry run unless --apply.")
    p.add_argument("--csv", required=True, help="The CSV of records to create")
    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", default="default",
                   help="DNS view for rows with no view column. Default 'default'")
    p.add_argument("--apply", action="store_true",
                   help="Actually write. Without it the script only plans")
    p.add_argument("--on-conflict", choices=("skip", "update", "abort"), default="skip",
                   help="A name that exists with a DIFFERENT value: leave it (skip, "
                        "the default), repoint it (update), or refuse the row (abort)")
    p.add_argument("--max-objects", type=int, default=500,
                   help="Refuse to apply a plan bigger than this unless "
                        "--confirm-large is given. Default 500")
    p.add_argument("--confirm-large", action="store_true",
                   help="Acknowledge a plan over --max-objects")
    p.add_argument("--rollback-out",
                   help="Where to write the rollback file. Default rollback-<UTC>.jsonl "
                        "next to the CSV. Only written with --apply")
    p.add_argument("--validate-only", action="store_true",
                   help="Check the CSV and stop. No grid, no credentials needed")
    p.add_argument("--show-skipped", action="store_true",
                   help="List the SKIP rows too, not just the counts")
    p.add_argument("--limit", type=int, default=50,
                   help="Rows printed per section. Default 50")
    p.add_argument("--insecure", action="store_true",
                   help="Skip TLS verification. Only for a certificate you recognise")
    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 args.validate_only:
        return args
    if bool(args.host) == bool(args.offline_dir):
        p.error("pass exactly one of --host or --offline-dir (or --validate-only)")
    return args


def main(argv=None):
    args = parse_args(argv)
    rows, errors = read_csv(args.csv, args.view)

    if args.validate_only:
        print("%s: %d row(s) well formed, %d rejected."
              % (args.csv, len(rows), len(errors)))
        if errors:
            for ln, msg in errors[:args.limit]:
                print("  line %-5d %s" % (ln, msg))
            print("Nothing else was done. Fix the file and run this again.")
            return 1
        print("Every row is well formed. This says nothing about whether the zones")
        print("exist on your grid -- that check needs --host or --offline-dir.")
        return 0

    wapi = Wapi(host=args.host, version=args.wapi_version, insecure=args.insecure,
                offline_dir=args.offline_dir, save_dir=args.save_dir)
    planner = Planner(wapi, args)
    if errors:
        planner.counts = {"views": 0, "zones": 0, "A": 0, "PTR": 0,
                          "CNAME": 0, "host": 0}
        report_plan(planner, rows, args, errors)
        return 2

    planner.fetch()
    planner.plan(rows)
    writes = report_plan(planner, rows, args, errors)
    ops = [op for row in rows for op in row.ops]
    blocked = [o for o in ops if o.verdict == BLOCKED]
    conflicts = [o for o in ops if o.verdict == SKIP and o.reason.startswith("CONFLICT")]

    if not args.apply:
        print("")
        print("  Dry run. Nothing was written. Re-read the plan above, then add")
        print("  --apply to the same command to execute exactly it.")
        return 1 if (blocked or conflicts) else 0

    if writes > args.max_objects and not args.confirm_large:
        print("")
        print("  REFUSING TO APPLY: %d objects is over --max-objects %d."
              % (writes, args.max_objects))
        print("  A CSV exported with the wrong filter is how a zone gets rewritten.")
        print("  If %d is really what you meant, add --confirm-large." % writes)
        return 2

    rollback = None
    if not args.offline_dir:
        path = args.rollback_out or os.path.join(
            os.path.dirname(os.path.abspath(args.csv)),
            "rollback-%s.jsonl" % datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ"))
        rollback = Rollback(path)
    else:
        rollback = Rollback(os.devnull)

    todo = [o for o in ops if o.verdict in (CREATE, UPDATE)]
    try:
        done, failed = apply_plan(wapi, todo, rollback)
    finally:
        rollback.close()
    report_apply(todo, rollback, args)
    return 1 if (failed or blocked or conflicts) else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.stderr.write("\ninterrupted\n")
        sys.exit(2)
    except SystemExit as exc:
        # The shared loader raises SystemExit with a sentence, which Python
        # would turn into exit code 1 -- the code this script reserves for
        # "there are findings". A setup failure is a 2.
        if isinstance(exc.code, str):
            sys.stderr.write(exc.code + "\n")
            sys.exit(2)
        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)