Infoblox Unused IP Space Finder

#!/usr/bin/env python3
"""
Infoblox Unused IP Space Finder

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/infoblox-unused-ip-space

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 anything.
#
# EXIT CODES   0 nothing reclaimable   1 reclaimable found   2 the run failed
#
# CREDENTIALS come from the environment, never from an argument:
#   IB_USER / IB_PASS      (not needed at all with --offline-dir)

import argparse
import csv
import ipaddress
import json
import os
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
UNUSED = "UNUSED"
TRANSITIVE = "UNUSED_TRANSITIVELY"
UNVERIFIABLE = "COULD_NOT_VERIFY"
USED = "USED"

SEVERITY = {UNUSED: 0, TRANSITIVE: 1, UNVERIFIABLE: 2, USED: 3}


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

    def __init__(self, kind, name, view, verdict, reason, detail=""):
        self.kind = kind          # network / range / fixedaddress / networkcontainer
        self.name = name          # the CIDR, the range bounds, or the IP
        self.view = view          # network view: the same CIDR legitimately repeats
        self.verdict = verdict
        self.reason = reason      # why this verdict, in one line
        self.detail = detail      # the numbers behind it

    def key(self):
        return (self.view, self.kind, self.name)


# ---------------------------------------------------------------- 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 network 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 pct(value):
    """network.utilization is a plain percentage.

    range.dhcp_utilization and network.dhcp_utilization are NOT -- they are
    the percentage multiplied by 1000, so 250 means 25.0%. Reading one as the
    other is the single commonest mistake made against this API, and it turns
    a 25%-full range into a 250%-full one or a 2.5%-empty one depending on
    which way you get it wrong.
    """
    return None if value is None else float(value)


def permille(value):
    return None if value is None else float(value) / 10.0


def plural(n, word):
    return "%d %s%s" % (n, word, "" if n == 1 else "s")


def age_days(epoch):
    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 size_of(cidr):
    try:
        return ipaddress.ip_network(cidr, strict=False).num_addresses
    except ValueError:
        return 0


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

    def add(self, f):
        self.findings.append(f)

    # -- networks ---------------------------------------------------------
    def networks(self):
        fields = ("network,network_view,comment,utilization,utilization_update,"
                  "dhcp_utilization,dhcp_utilization_status,total_hosts,extattrs,"
                  "disable")
        params = {"_return_fields": fields}
        if self.args.network_view:
            params["network_view"] = self.args.network_view
        nets = self.w.get("network", params)

        for n in nets:
            cidr = n.get("network", "?")
            view = n.get("network_view", "default")
            util = pct(n.get("utilization"))
            stale = age_days(n.get("utilization_update"))

            if util is None:
                self.add(Finding(
                    "network", cidr, view, UNVERIFIABLE,
                    "the grid has never computed utilisation for this network",
                    "utilization absent"))
                continue

            if util > 0:
                if self.args.show_used:
                    self.add(Finding("network", cidr, view, USED,
                                     "in use", "%.1f%% utilised" % util))
                continue

            # utilisation is 0. That is a claim, and claims have an age.
            if stale is None:
                self.add(Finding(
                    "network", cidr, view, UNVERIFIABLE,
                    "utilisation reads 0% but carries no update timestamp",
                    "utilization_update absent"))
                continue
            if stale > self.args.stale_days:
                self.add(Finding(
                    "network", cidr, view, UNVERIFIABLE,
                    "utilisation reads 0%% but was last computed %d days ago" % stale,
                    "older than --stale-days %d" % self.args.stale_days))
                continue

            self.verify_network(n, cidr, view, stale)

    def verify_network(self, n, cidr, view, stale):
        """0% utilisation is where most tools stop. It is where the work starts.

        Utilisation counts what DHCP and IPAM know. A network that is fully
        deployed with static addresses and documented nowhere reads exactly
        like a network nobody ever used. Deleting it deletes the only record
        that the space is taken -- and the next engineer allocates it again.
        """
        size = size_of(cidr)
        if size > self.args.max_addresses:
            self.add(Finding(
                "network", cidr, view, UNVERIFIABLE,
                "too large to enumerate address by address (%d addresses)" % size,
                "raise --max-addresses to check it, at the cost of a long run"))
            return

        used = self.w.get("ipv4address", {
            "network": cidr,
            "network_view": view,
            "status": "USED",
            "_return_fields": "ip_address,types,names,status,usage,discovered_data",
        })
        # BROADCAST / NETWORK / GATEWAY are structural, not occupancy.
        structural = {"BROADCAST", "NETWORK"}
        real = [a for a in used
                if set(a.get("types") or []) - structural]

        if real:
            kinds = sorted({t for a in real for t in (a.get("types") or [])})
            self.add(Finding(
                "network", cidr, view, USED,
                "utilisation says 0%% but %d %s claimed" % (
                    len(real), "address is" if len(real) == 1 else "addresses are"),
                "types: " + ", ".join(kinds)))
            return

        discovered = [a for a in used if (a.get("discovered_data") or {}).get("last_discovered")]
        if discovered:
            self.add(Finding(
                "network", cidr, view, UNVERIFIABLE,
                "IPAM is empty but discovery has seen %s here" % plural(len(discovered), "live host"),
                "IPAM is wrong, the network is not free"))
            return

        if not self.args.discovery_enabled:
            self.add(Finding(
                "network", cidr, view, TRANSITIVE,
                "empty in IPAM, and no discovery data to contradict it",
                "statically addressed hosts would be invisible; pass "
                "--discovery-enabled only if discovery really covers this view"))
            return

        self.add(Finding(
            "network", cidr, view, UNUSED,
            "empty in IPAM and discovery saw nothing",
            "utilisation computed %d day(s) ago" % stale))

    # -- DHCP ranges ------------------------------------------------------
    def ranges(self):
        fields = ("start_addr,end_addr,network,network_view,comment,disable,"
                  "dhcp_utilization,dhcp_utilization_status,extattrs")
        params = {"_return_fields": fields}
        if self.args.network_view:
            params["network_view"] = self.args.network_view
        for r in self.w.get("range", params):
            name = "%s-%s" % (r.get("start_addr", "?"), r.get("end_addr", "?"))
            view = r.get("network_view", "default")
            util = permille(r.get("dhcp_utilization"))

            if r.get("disable"):
                self.add(Finding("range", name, view, TRANSITIVE,
                                 "administratively disabled, so it serves nobody",
                                 "disable=true; it holds the space without using it"))
                continue
            if util is None:
                self.add(Finding("range", name, view, UNVERIFIABLE,
                                 "no DHCP utilisation reported for this range", ""))
                continue
            if util > 0:
                if self.args.show_used:
                    self.add(Finding("range", name, view, USED, "in use",
                                     "%.1f%% of the range leased" % util))
                continue

            leases = self.w.get("lease", {
                "network_view": view,
                "address>=": r.get("start_addr"),
                "address<=": r.get("end_addr"),
                "_return_fields": "address,binding_state,ends,hardware",
            })
            live = [l for l in leases if l.get("binding_state") in ("ACTIVE", "STATIC")]
            abandoned = [l for l in leases if l.get("binding_state") == "ABANDONED"]

            if live:
                self.add(Finding("range", name, view, USED,
                                 "%s despite 0%% utilisation" % plural(len(live), "active lease"),
                                 "utilisation is computed periodically, leases are live"))
                continue
            if abandoned:
                self.add(Finding(
                    "range", name, view, UNVERIFIABLE,
                    "%s ABANDONED -- the server offered the address and "
                    "something already answered on it" % plural(len(abandoned), "lease"),
                    "that is a static host inside a DHCP range, not free space"))
                continue

            self.add(Finding(
                "range", name, view, UNVERIFIABLE,
                "no leases found, which is weaker evidence than it looks",
                "the grid keeps a lease only for its lease time; a long lease "
                "time plus a recent restart produces an empty table for a "
                "range in daily use. Confirm against the lease time before acting"))

    # -- fixed addresses --------------------------------------------------
    def fixed(self):
        fields = ("ipv4addr,mac,name,network,network_view,comment,disable,extattrs"
                  ",match_client")
        params = {"_return_fields": fields}
        if self.args.network_view:
            params["network_view"] = self.args.network_view
        for f in self.w.get("fixedaddress", params):
            ip = f.get("ipv4addr", "?")
            view = f.get("network_view", "default")
            mac = (f.get("mac") or "").lower()

            if f.get("disable"):
                self.add(Finding("fixedaddress", ip, view, TRANSITIVE,
                                 "disabled, so it reserves the address and serves nothing",
                                 "disable=true"))
                continue

            if mac in ("", "00:00:00:00:00:00"):
                self.add(Finding(
                    "fixedaddress", ip, view, TRANSITIVE,
                    "a reservation with no MAC -- it blocks the address and "
                    "hands it to nobody",
                    "these accumulate for years and are the easiest real win, "
                    "but check the comment first: some are deliberate holes"))
                continue

            leases = self.w.get("lease", {
                "address": ip, "network_view": view,
                "_return_fields": "address,binding_state,ends,hardware",
            })
            if any(l.get("binding_state") in ("ACTIVE", "STATIC") for l in leases):
                if self.args.show_used:
                    self.add(Finding("fixedaddress", ip, view, USED,
                                     "an active lease exists for this reservation", mac))
                continue

            self.add(Finding(
                "fixedaddress", ip, view, UNVERIFIABLE,
                "no lease, and a fixed address carries no last-seen field",
                "a powered-off laptop and a decommissioned server are "
                "indistinguishable here. Age it against DHCP lease history or "
                "discovery, not against this script"))

    def run(self):
        self.networks()
        self.ranges()
        self.fixed()
        self.unchecked = [
            "DNS records (A/PTR/CNAME) that point into a network are NOT traced "
            "here -- that is infoblox-stale-dns-records",
            "IPv6 networks (the ipv6network object) are out of scope",
            "Network containers are skipped: a container is empty by design and "
            "judging it like a leaf network produces nonsense",
            "DHCP option filters, failover associations and MAC filters that "
            "reference a range are not resolved",
            "Objects in network views you did not pass with --network-view",
        ]
        return self.findings


# ---------------------------------------------------------------- reporting
def report(findings, unchecked, args):
    order = sorted(findings, key=lambda f: (SEVERITY[f.verdict], f.view, f.kind, f.name))
    buckets = {}
    for f in order:
        buckets.setdefault(f.verdict, []).append(f)

    line = "=" * 74
    print(line)
    print("infoblox-unused-ip-space")
    print(line)
    print("Grid           : %s" % (args.offline_dir or args.host))
    print("Network views  : %s" % (args.network_view or "all"))
    print("")

    titles = [
        (UNUSED, "RECLAIMABLE -- empty in IPAM and nothing contradicted it"),
        (TRANSITIVE, "RECLAIMABLE WITH A CAVEAT -- read the reason on each line"),
        (UNVERIFIABLE, "COULD NOT VERIFY -- absence of evidence, not evidence of absence"),
        (USED, "IN USE"),
    ]
    for verdict, title in titles:
        rows = buckets.get(verdict, [])
        if not rows:
            continue
        print("%s (%d)" % (title, len(rows)))
        print("-" * 74)
        for f in rows[:args.limit]:
            print("  [%s] %s  (view: %s)" % (f.kind, 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("")

    print("NOT CHECKED BY THIS SCRIPT")
    print("-" * 74)
    for u in unchecked:
        print("  - %s" % u)
    print("")
    print("This script never deletes anything and never will. 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", "kind", "object", "network_view", "reason", "detail"])
        for f in sorted(findings, key=lambda x: (SEVERITY[x.verdict], x.view, x.name)):
            w.writerow([f.verdict, f.kind, f.name, f.view, f.reason, f.detail])


# ---------------------------------------------------------------- entry
def parse_args(argv):
    p = argparse.ArgumentParser(
        description="Find address space Infoblox NIOS believes is empty, and "
                    "say how much that belief is worth. 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("--network-view", action="append",
                   help="Limit to this network view. Repeatable. Default: all")
    p.add_argument("--insecure", action="store_true",
                   help="Skip TLS verification. Only for a certificate you recognise")
    p.add_argument("--stale-days", type=int, default=7,
                   help="Utilisation older than this is not trusted. Default 7")
    p.add_argument("--max-addresses", type=int, default=1024,
                   help="Do not enumerate networks larger than this. Default 1024 (/22)")
    p.add_argument("--discovery-enabled", action="store_true",
                   help="Assert that discovery really covers these views. Without "
                        "it, an empty network is reported as UNUSED_TRANSITIVELY "
                        "rather than UNUSED, because static hosts are invisible")
    p.add_argument("--show-used", action="store_true",
                   help="Also list objects that ARE in use, to prove a specific one is")
    p.add_argument("--limit", type=int, default=50, help="Rows printed per section. Default 50")
    p.add_argument("--csv", help="Write every object 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")
    if args.network_view and len(args.network_view) == 1:
        args.network_view = args.network_view[0]
    elif args.network_view:
        p.error("one --network-view per run for now; run it once per view")
    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)
    findings = audit.run()
    report(findings, audit.unchecked, args)
    if args.csv:
        write_csv(args.csv, findings)
        print("\nCSV written: %s" % args.csv)
    reclaimable = [f for f in findings if f.verdict in (UNUSED, TRANSITIVE)]
    return 1 if reclaimable 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)