Panorama Unused Object Finder

#!/usr/bin/env python3
"""
Panorama Unused Object Finder

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/panorama-unused-objects

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

import argparse
import csv
import os
import sys
import xml.etree.ElementTree as ET

sys.stdout.reconfigure(encoding="utf-8", errors="replace")

# Object classes this tool inventories, as (xml tag, human label).
OBJECT_CLASSES = [
    ("address", "address"),
    ("address-group", "address group"),
    ("service", "service"),
    ("service-group", "service group"),
    ("application-group", "application group"),
    ("tag", "tag"),
    ("external-list", "external dynamic list"),
]

# Reference classes scanned inside a rulebase. Every one of these is a place an
# object can be used; missing one is how a tool reports an in-use object as
# unused, which is how somebody deletes something in production.
RULEBASES = [
    "security", "nat", "pbf", "decryption", "application-override",
    "qos", "authentication", "dos", "sdwan-policy-based-forwarding",
    "tunnel-inspect", "network-packet-broker",
]

# Element names inside a rule that can name an object.
RULE_REF_FIELDS = [
    "source", "destination", "service", "application", "tag", "group-tag",
    "source-translation", "destination-translation", "translated-address",
    "dynamic-ip-and-port", "dynamic-ip", "static-ip", "translated-port",
    "address", "snat-address", "to-interface", "source-address",
    "destination-address",
]

# Things this tool does NOT read. Printed in every report, because "I did not
# check this" and "this is unused" are different facts.
NOT_CHECKED = [
    "Template and template-stack configuration (service routes, log forwarding "
    "profiles, interface and zone definitions, DHCP, IKE gateways)",
    "Security profiles and profile groups that reference EDLs or custom objects",
    "Dynamic address group match expressions that resolve by tag at runtime",
    "Objects referenced only on a managed firewall's local configuration, "
    "outside Panorama",
    "Anything referenced by an external system through the API",
]


class Obj:
    __slots__ = ("name", "kind", "location", "used_by", "shadowed_by",
                 "tags", "transitively_dead")

    def __init__(self, name, kind, location):
        self.name = name
        self.kind = kind
        self.location = location
        self.used_by = []
        self.shadowed_by = []
        self.tags = []
        self.transitively_dead = False


def fetch_config(host, api_key, verify):
    """Pull the whole running configuration in one call.

    One export rather than dozens of per-object queries: nested groups,
    device-group inheritance and overrides are all resolvable from a single
    document, and it avoids guessing an xpath per object class.
    """
    import requests
    import urllib3
    if not verify:
        urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
    url = "https://%s/api/" % host
    params = {"type": "config", "action": "get", "xpath": "/config", "key": api_key}
    r = requests.get(url, params=params, verify=verify, timeout=180)
    r.raise_for_status()
    root = ET.fromstring(r.content)
    status = root.get("status")
    if status != "success":
        msg = root.findtext(".//msg") or r.text[:300]
        raise SystemExit("ERROR: Panorama returned status=%s: %s" % (status, msg))
    result = root.find("result")
    if result is None or len(result) == 0:
        raise SystemExit("ERROR: the API returned an empty configuration.")
    return result[0]


def load_config_file(path):
    try:
        tree = ET.parse(path)
    except FileNotFoundError:
        raise SystemExit("ERROR: file not found: %s" % path)
    except ET.ParseError as exc:
        raise SystemExit("ERROR: %s is not valid XML: %s" % (path, exc))
    root = tree.getroot()
    # Accept either a bare <config> or a full API response wrapping it.
    if root.tag == "response":
        result = root.find("result")
        if result is None or len(result) == 0:
            raise SystemExit("ERROR: no <config> inside the API response.")
        return result[0]
    return root


def device_groups(config):
    """Return {name: parent_name_or_None}, honouring the DG hierarchy."""
    out = {}
    for dg in config.findall("./devices/entry/device-group/entry"):
        out[dg.get("name")] = None
    for pe in config.findall("./readonly/devices/entry/device-group/entry") or []:
        pass
    # Parent relationships live in /config/devices/entry/device-group/entry/..
    # or in the Panorama dg-meta-data section depending on version.
    for e in config.findall(".//dg-meta-data/dginfo/entry"):
        name = e.get("name")
        parent = e.findtext("parent-dg")
        if name in out:
            out[name] = parent or None
    return out


def ancestors(dg, parents):
    """shared -> ... -> dg, nearest-last."""
    chain, seen, cur = [], set(), dg
    while cur and cur not in seen:
        seen.add(cur)
        chain.append(cur)
        cur = parents.get(cur)
    chain.append("shared")
    return list(reversed(chain))


def descendants(dg, parents):
    out, changed = {dg}, True
    while changed:
        changed = False
        for child, parent in parents.items():
            if parent in out and child not in out:
                out.add(child)
                changed = True
    return out


def location_roots(config):
    """Yield (location_name, element) for shared and every device group."""
    shared = config.find("./shared")
    if shared is not None:
        yield "shared", shared
    for dg in config.findall("./devices/entry/device-group/entry"):
        yield dg.get("name"), dg


def collect_objects(config):
    objs = {}
    for loc, root in location_roots(config):
        for tag, kind in OBJECT_CLASSES:
            for e in root.findall("./%s/entry" % tag):
                name = e.get("name")
                if not name:
                    continue
                o = Obj(name, kind, loc)
                o.tags = [m.text.strip() for m in e.findall("./tag/member")
                          if m.text and m.text.strip()]
                objs[(loc, kind, name)] = o
    return objs


def iter_text(elem):
    """Every member/text value under an element, flattened."""
    for m in elem.iter():
        if m.text and m.text.strip():
            yield m.text.strip()


def collect_references(config, parents):
    """Build {(scope, name): [reference descriptions]}.

    `scope` is the location whose objects a reference can resolve against:
    a rule in device group DG can resolve to DG, any ancestor, or shared.
    """
    refs = {}

    def add(scope_dg, value, where):
        refs.setdefault((scope_dg, value), []).append(where)

    for loc, root in location_roots(config):
        # 1. Group membership - nested groups are the class most often missed.
        for tag, kind in (("address-group", "address group"),
                          ("service-group", "service group"),
                          ("application-group", "application group")):
            for g in root.findall("./%s/entry" % tag):
                gname = g.get("name")
                for m in g.findall(".//member"):
                    if m.text:
                        add(loc, m.text.strip(), "%s '%s'" % (kind, gname))
                # A dynamic address group matches by tag, not by member.
                df = g.find("./dynamic/filter")
                if df is not None and df.text:
                    for token in df.text.replace("(", " ").replace(")", " ").split():
                        t = token.strip().strip("'\"")
                        if t and t.lower() not in ("and", "or", "not"):
                            add(loc, t, "dynamic group '%s' filter" % gname)

        # 2. Tags attached to objects.
        for tag, _kind in OBJECT_CLASSES:
            for e in root.findall("./%s/entry" % tag):
                for m in e.findall("./tag/member"):
                    if m.text:
                        add(loc, m.text.strip(), "tag on %s '%s'" % (tag, e.get("name")))

        # 3. Every rulebase, pre and post.
        for base in RULEBASES:
            for stage in ("pre-rulebase", "post-rulebase", "rulebase"):
                for rule in root.findall("./%s/%s/rules/entry" % (stage, base)):
                    rname = rule.get("name")
                    where = "%s %s rule '%s'" % (stage, base, rname)
                    for field in RULE_REF_FIELDS:
                        for fe in rule.findall(".//%s" % field):
                            for val in iter_text(fe):
                                add(loc, val, where)
    return refs


def resolve(objs, refs, parents):
    """Mark each object used if anything in a scope that can see it names it."""
    for (loc, kind, name), obj in objs.items():
        if loc == "shared":
            scopes = set(parents.keys()) | {"shared"}
        else:
            scopes = descendants(loc, parents) | {loc}
        for scope in scopes:
            for where in refs.get((scope, name), []):
                obj.used_by.append("%s: %s" % (scope, where))

        # Override: a device group defining the same name shadows the ancestor
        # copy inside that subtree. The ancestor is NOT unused globally.
        if loc == "shared":
            for dg in parents:
                if (dg, kind, name) in objs:
                    obj.shadowed_by.append(dg)

    # An object whose only references come from containers that are themselves
    # unreferenced is dead too - it is being kept alive by a corpse. Commercial
    # tools call this transitive; most home-grown scripts miss it entirely.
    by_name = {}
    for (loc, kind, name), o in objs.items():
        by_name.setdefault(name, []).append(o)

    def container_of(where):
        # "DG: address group 'X'" / "DG: service group 'X'"
        if "group '" in where:
            return where.split("group '", 1)[1].rstrip("'")
        return None

    for _ in range(8):          # iterate: a chain of dead groups collapses
        changed = False
        for o in objs.values():
            if not o.used_by or o.transitively_dead:
                continue
            holders = [container_of(w) for w in o.used_by]
            if not all(holders):
                continue        # at least one reference is a rule, so alive
            alive = False
            for h in holders:
                for cand in by_name.get(h, []):
                    if cand.used_by and not cand.transitively_dead:
                        alive = True
            if not alive:
                o.transitively_dead = True
                changed = True
        if not changed:
            break
    return objs


RESERVED = {"any", "application-default", "none", "all"}


CSV_HEADER = ["location", "object_type", "name", "verdict",
              "safe_to_delete", "referenced_by", "note"]


def write_csv(path, rows):
    """One row per object, with the verdict as a column.

    CSV rather than xlsx on purpose: the standard library writes it, Excel
    and Sheets both open it, and a cleanup project lives in a spreadsheet
    where someone filters by verdict and assigns an owner per row.
    """
    try:
        with open(path, "w", encoding="utf-8-sig", newline="") as fh:
            w = csv.writer(fh)
            w.writerow(CSV_HEADER)
            w.writerows(rows)
    except OSError as exc:
        print("WARNING: could not write %s: %s" % (path, exc))
        return
    print("CSV written: %s (%d rows)" % (path, len(rows)))


def report(objs, parents, show_used, limit, csv_path=None):
    rows = []
    by_loc = {}
    for obj in objs.values():
        by_loc.setdefault(obj.location, []).append(obj)

    total_unused = 0
    print("=" * 78)
    print("panorama-unused-objects")
    print("=" * 78)
    print("Locations: shared plus %d device group(s)" % len(parents))
    if parents:
        for dg, parent in sorted(parents.items()):
            print("  %-32s parent=%s" % (dg, parent or "shared"))
    print()

    for loc in ["shared"] + sorted(parents):
        items = by_loc.get(loc, [])
        if not items:
            continue
        unused = [o for o in items
                  if (not o.used_by or o.transitively_dead)
                  and o.name.lower() not in RESERVED]
        shadowed = [o for o in unused if o.shadowed_by]
        rest = [o for o in unused if not o.shadowed_by]
        # A TAGGED object can be matched at runtime by a dynamic address group,
        # so it is never a plain "unused" - that would be the exact false
        # positive that gets something deleted in production.
        tagged = [o for o in rest if o.tags and not o.transitively_dead]
        transitive = [o for o in rest if o.transitively_dead]
        clean = [o for o in rest if o not in tagged and o not in transitive]

        print("-" * 78)
        print("%s  -  %d object(s), %d with no reference found"
              % (loc, len(items), len(unused)))
        print("-" * 78)

        for o in clean:
            rows.append([loc, o.kind, o.name, "UNUSED", "candidate", "", ""])
        for o in transitive:
            rows.append([loc, o.kind, o.name, "UNUSED_TRANSITIVE", "candidate",
                         o.used_by[0] if o.used_by else "",
                         "only referenced by a group that is itself unused"])
        for o in tagged:
            rows.append([loc, o.kind, o.name, "COULD_NOT_VERIFY", "no",
                         "", "tagged (%s) - a dynamic address group may match "
                             "it at runtime" % ", ".join(o.tags)])
        for o in shadowed:
            rows.append([loc, o.kind, o.name, "COULD_NOT_VERIFY", "no", "",
                         "overridden in: %s" % ", ".join(o.shadowed_by)])
        for o in items:
            if o.used_by and not o.transitively_dead:
                rows.append([loc, o.kind, o.name, "IN_USE", "no",
                             o.used_by[0]
                             + ("" if len(o.used_by) == 1
                                else " (+%d more)" % (len(o.used_by) - 1)), ""])

        if clean:
            print("  UNUSED (no reference found anywhere that can see them):")
            for o in sorted(clean, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-22s %s" % (o.kind, o.name))
            if len(clean) > limit:
                print("    ... and %d more (raise --limit)" % (len(clean) - limit))
            total_unused += len(clean)
        else:
            print("  No unreferenced objects found.")

        if transitive:
            print()
            print("  UNUSED TRANSITIVELY - referenced only by groups that are")
            print("  themselves unreferenced. Delete the container first.")
            for o in sorted(transitive, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-22s %-34s via %s"
                      % (o.kind, o.name, o.used_by[0].split(": ", 1)[-1]))
            total_unused += len(transitive)

        if tagged:
            print()
            print("  COULD NOT VERIFY - tagged, so a dynamic address group may")
            print("  match them at runtime. Check the DAG filters before acting.")
            for o in sorted(tagged, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-22s %-34s tags: %s"
                      % (o.kind, o.name, ", ".join(o.tags)))

        if shadowed:
            print()
            print("  COULD NOT VERIFY - overridden in a child device group.")
            print("  A device group defines an object with the same name, so this")
            print("  copy may still be the one in effect elsewhere. Do not delete")
            print("  on the strength of this report.")
            for o in sorted(shadowed, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-22s %-34s overridden in: %s"
                      % (o.kind, o.name, ", ".join(o.shadowed_by[:4])))

        if show_used:
            used = [o for o in items if o.used_by]
            print()
            print("  IN USE (%d):" % len(used))
            for o in sorted(used, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-22s %-34s %s"
                      % (o.kind, o.name, o.used_by[0]
                         + ("" if len(o.used_by) == 1
                            else " (+%d more)" % (len(o.used_by) - 1))))
        print()

    print("=" * 78)
    print("WHAT THIS REPORT DID NOT CHECK")
    print("=" * 78)
    for line in NOT_CHECKED:
        print("  - %s" % line)
    print()
    print("  An object listed as unused is a CANDIDATE TO INVESTIGATE, not a")
    print("  delete list. This tool never deletes anything.")
    print()
    print("Total candidates: %d" % total_unused)
    if csv_path:
        write_csv(csv_path, rows)
    return total_unused


def main():
    ap = argparse.ArgumentParser(
        description="Find network objects in Panorama that no policy "
                    "references. Read-only; never deletes anything.")
    src = ap.add_mutually_exclusive_group(required=True)
    src.add_argument("--host", help="Panorama hostname or IP")
    src.add_argument("--config-file",
                     help="A configuration XML already exported from Panorama. "
                          "Lets you run this with no credentials at all.")
    ap.add_argument("--insecure", action="store_true",
                    help="Skip TLS verification (known self-signed certificate only)")
    ap.add_argument("--show-used", action="store_true",
                    help="Also list objects that ARE referenced, and by what")
    ap.add_argument("--limit", type=int, default=50,
                    help="Maximum objects listed per section (default 50)")
    ap.add_argument("--csv", metavar="PATH",
                    help="Also write every object and its verdict to a CSV, "
                         "for the spreadsheet the cleanup actually lives in")
    ap.add_argument("--save-config", metavar="PATH",
                    help="Write the fetched configuration to a file for re-runs")
    args = ap.parse_args()

    if args.host:
        key = os.environ.get("PAN_API_KEY")
        if not key:
            raise SystemExit(
                "ERROR: PAN_API_KEY must be set in the environment.\n"
                "\n"
                "  Generate one:\n"
                "    curl -k 'https://PANORAMA/api/?type=keygen&user=USER&password=PASS'\n"
                "\n"
                "  Linux / macOS:  export PAN_API_KEY='LUFRPT...'\n"
                "  PowerShell:     $env:PAN_API_KEY='LUFRPT...'\n"
                "\n"
                "Read from the environment on purpose: an argument is visible in\n"
                "`ps auxww` and lands in shell history.")
        try:
            import requests  # noqa: F401
        except ImportError:
            raise SystemExit("ERROR: pip install requests")
        import requests
        try:
            config = fetch_config(args.host, key, not args.insecure)
        except requests.exceptions.SSLError as exc:
            raise SystemExit(
                "ERROR: TLS verification failed for %s: %s\n"
                "Use --insecure only for a known self-signed certificate."
                % (args.host, exc))
        except requests.exceptions.RequestException as exc:
            raise SystemExit("ERROR: could not reach %s: %s" % (args.host, exc))
        except ET.ParseError as exc:
            raise SystemExit("ERROR: the response was not valid XML: %s" % exc)
        if args.save_config:
            ET.ElementTree(config).write(args.save_config, encoding="utf-8")
            print("Configuration saved to %s" % args.save_config)
    else:
        config = load_config_file(args.config_file)

    parents = device_groups(config)
    objs = collect_objects(config)
    if not objs:
        print("No objects found. Is this a Panorama configuration, and does the")
        print("API key's admin role allow reading it?")
        return 2
    refs = collect_references(config, parents)
    resolve(objs, refs, parents)
    n = report(objs, parents, args.show_used, args.limit, args.csv)
    return 1 if n else 0


if __name__ == "__main__":
    sys.exit(main())