FortiManager Unused Object Finder

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

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/fortimanager-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 json
import os
import re
import sys

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

# (api path fragment, human label). Paths are relative to the ADOM root, and
# "global" is its own pseudo-ADOM whose objects every other ADOM can see -
# the same shape as Panorama's shared.
OBJECT_CLASSES = [
    ("firewall/address",        "address"),
    ("firewall/address6",       "address6"),
    ("firewall/addrgrp",        "address group"),
    ("firewall/addrgrp6",       "address group6"),
    ("firewall/service/custom", "service"),
    ("firewall/service/group",  "service group"),
    ("firewall/vip",            "vip"),
    ("firewall/vipgrp",         "vip group"),
    ("firewall/ippool",         "ip pool"),
    ("firewall/schedule/recurring", "schedule"),
    ("firewall/schedule/group", "schedule group"),
    ("application/list",        "application list"),
]

# Fields in a policy that can name an object.
POLICY_REF_FIELDS = [
    "srcaddr", "dstaddr", "srcaddr6", "dstaddr6", "service", "schedule",
    "poolname", "srcintf", "dstintf", "internet-service-name",
    "application-list", "vip", "nat-ippool",
]

# Group member fields, per class.
GROUP_MEMBER_FIELDS = ["member", "srcaddr", "dstaddr"]

NOT_CHECKED = [
    "Device-level configuration installed outside FortiManager",
    "VPN phase1/phase2 interfaces and IPsec tunnel definitions",
    "SD-WAN rules and performance SLA members",
    "Security profiles: webfilter, IPS, antivirus, DLP and their groups",
    "Scripts, TCL templates and CLI templates",
    "Local-in policies, proxy policies and traffic shapers",
]


class FortiManager:
    """FortiManager JSON-RPC client.

    Everything is a POST to /jsonrpc with a method and a url, which is why
    this looks nothing like the Panorama or FMC clients even though it does
    the same job.
    """

    def __init__(self, host, user, password, verify=True, save_dir=None):
        import requests
        import urllib3
        self.requests = requests
        self.url = "https://%s/jsonrpc" % host
        self.host = host
        self.verify = verify
        self.save_dir = save_dir
        self.session = None
        self.calls = 0
        self._id = 0
        if not verify:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        self.s = requests.Session()
        self.s.verify = verify
        self._login(user, password)

    def _rpc(self, method, params, with_session=True):
        self._id += 1
        body = {"id": self._id, "method": method, "params": params}
        if with_session and self.session:
            body["session"] = self.session
        r = self.s.post(self.url, json=body, timeout=120)
        r.raise_for_status()
        self.calls += 1
        data = r.json()
        return data

    def _login(self, user, password):
        data = self._rpc("exec", [{"url": "/sys/login/user",
                                   "data": {"user": user, "passwd": password}}],
                         with_session=False)
        status = (data.get("result") or [{}])[0].get("status", {})
        if status.get("code") != 0:
            raise SystemExit(
                "ERROR: %s rejected the credentials: %s"
                % (self.host, status.get("message", "unknown")))
        self.session = data.get("session")
        if not self.session:
            raise SystemExit("ERROR: FortiManager did not return a session.")

    def logout(self):
        if self.session:
            try:
                self._rpc("exec", [{"url": "/sys/logout"}])
            except Exception:
                pass

    def get(self, url, option=None):
        params = {"url": url}
        if option:
            params["option"] = option
        data = self._rpc("get", [params])
        res = (data.get("result") or [{}])[0]
        status = res.get("status", {})
        code = status.get("code")
        # -6 and -3 are "object does not exist" on this ADOM. Not an error:
        # a given class may simply not be configured.
        if code not in (0, -3, -6):
            print("  WARNING: %s returned %s (%s)"
                  % (url, code, status.get("message")))
            return []
        out = res.get("data") or []
        if isinstance(out, dict):
            out = [out]
        if self.save_dir:
            self._save(url, out)
        return out

    def _save(self, url, data):
        name = re.sub(r"[^A-Za-z0-9]+", "_", url).strip("_")
        with open(os.path.join(self.save_dir, name[:180] + ".json"), "w",
                  encoding="utf-8") as fh:
            json.dump({"url": url, "data": data}, fh, indent=1)


class Offline:
    def __init__(self, directory):
        self.store = {}
        self.calls = 0
        if not os.path.isdir(directory):
            raise SystemExit("ERROR: not a directory: %s" % directory)
        for f in sorted(os.listdir(directory)):
            if f.endswith(".json"):
                with open(os.path.join(directory, f), encoding="utf-8") as fh:
                    try:
                        blob = json.load(fh)
                    except ValueError as exc:
                        raise SystemExit("ERROR: %s is not valid JSON: %s" % (f, exc))
                self.store[blob.get("url")] = blob.get("data") or []
        if not self.store:
            raise SystemExit("ERROR: no saved responses found in %s" % directory)

    def get(self, url, option=None):
        self.calls += 1
        return self.store.get(url, [])

    def logout(self):
        pass


class Obj:
    __slots__ = ("name", "kind", "adom", "used_by", "transitively_dead")

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


def adom_root(adom):
    return "/pm/config/global" if adom == "global" else "/pm/config/adom/%s" % adom


def list_adoms(api):
    out = []
    for a in api.get("/dvmdb/adom"):
        name = a.get("name")
        if name:
            out.append(name)
    return out


def names_from(value):
    """FortiManager returns a member as a string, a list, or a list of dicts."""
    if value is None:
        return []
    if isinstance(value, str):
        return [value]
    if isinstance(value, dict):
        n = value.get("name")
        return [n] if n else []
    if isinstance(value, list):
        out = []
        for v in value:
            out.extend(names_from(v))
        return out
    return []


def collect(api, adom):
    root = adom_root(adom)
    objs, raw = {}, {}
    for path, kind in OBJECT_CLASSES:
        items = api.get("%s/obj/%s" % (root, path))
        raw[kind] = items
        for it in items:
            name = it.get("name")
            if not name:
                continue
            objs[(kind, name)] = Obj(name, kind, adom)
    return objs, raw


def collect_refs(api, adom, objs, raw, global_objs=None):
    """Group membership first, then every policy package.

    A reference resolves against this ADOM's own objects AND against the
    global ADOM, because global objects are visible everywhere. Resolving
    only against the local ADOM makes every global object look unused -
    the same hierarchy mistake as checking Panorama device groups in
    isolation, and it is how somebody deletes an object three ADOMs use.
    """
    root = adom_root(adom)
    by_name = {}
    for o in objs.values():
        by_name.setdefault(o.name, []).append(o)
    if global_objs and adom != "global":
        for o in global_objs.values():
            by_name.setdefault(o.name, []).append(o)

    def record(name, where):
        for o in by_name.get(name, []):
            o.used_by.append("%s: %s" % (adom, where))

    # 1. Group membership, including nesting.
    for kind, items in raw.items():
        if "group" not in kind and kind not in ("vip group",):
            continue
        for g in items:
            gname = g.get("name")
            for field in GROUP_MEMBER_FIELDS:
                for m in names_from(g.get(field)):
                    record(m, "%s '%s'" % (kind, gname))

    # 2. A VIP names an address indirectly through its mapped IP; and an
    #    ippool is named by policies. Both are covered by the policy walk.

    scanned = []
    packages = api.get("%s/pkg" % root)

    def walk_package(pkg, prefix=""):
        pname = pkg.get("name")
        full = prefix + pname if not prefix else "%s/%s" % (prefix.rstrip("/"), pname)
        # A folder has subobj; a real package has a policy block.
        subs = pkg.get("subobj") or []
        if subs:
            for sp in subs:
                walk_package(sp, full + "/")
            return
        rules = api.get("%s/pkg/%s/firewall/policy" % (root, full))
        if rules:
            scanned.append("policy package '%s' (%d rules)" % (full, len(rules)))
        for rule in rules:
            where = "policy %s in '%s'" % (rule.get("policyid", "?"), full)
            for field in POLICY_REF_FIELDS:
                for nm in names_from(rule.get(field)):
                    record(nm, where)

    for pkg in packages:
        walk_package(pkg)

    return scanned


def mark_transitive(objs):
    by_name = {}
    for o in objs.values():
        by_name.setdefault(o.name, []).append(o)

    def container_of(where):
        if "group '" in where:
            return where.split("group '", 1)[1].rstrip("'")
        return None

    for _ in range(8):
        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
            alive = any(c.used_by and not c.transitively_dead
                        for h in holders for c in by_name.get(h, []))
            if not alive:
                o.transitively_dead = True
                changed = True
        if not changed:
            break


# Objects FortiOS ships with. Flagging them is noise and they cannot be
# deleted anyway.
BUILTIN = {"all", "none", "any", "ALL", "ALL_TCP", "ALL_UDP", "ALL_ICMP",
           "ALL_ICMP6", "always", "none-schedule", "default", "g-All",
           "login-time", "FABRIC_DEVICE", "SSLVPN_TUNNEL_ADDR1"}


def report(per_adom, limit, show_used, csv_path, global_consumers):
    rows, total = [], 0
    print("=" * 78)
    print("fortimanager-unused-objects")
    print("=" * 78)
    print("ADOMs analysed: %s" % ", ".join(per_adom))
    print()

    for adom, (objs, scanned) in per_adom.items():
        unused = [o for o in objs.values()
                  if (not o.used_by or o.transitively_dead)
                  and o.name not in BUILTIN]
        transitive = [o for o in unused if o.transitively_dead]
        rest = [o for o in unused if not o.transitively_dead]
        # A global object is visible to every ADOM. If some ADOMs were not
        # analysed, "unused" is a claim this run cannot make.
        unverifiable = rest if (adom == "global" and global_consumers) else []
        clean = [] if unverifiable else rest

        print("-" * 78)
        print("ADOM %s  -  %d object(s), %d with no reference found"
              % (adom, len(objs), len(unused)))
        print("-" * 78)
        print("  Policy packages scanned:")
        for s in (scanned or ["(none found)"]):
            print("    - %s" % s)
        print()


        if unverifiable:
            print("  COULD NOT VERIFY - global objects, and not every ADOM was")
            print("  analysed. Re-run with --all-adoms for a usable verdict.")
            for o in sorted(unverifiable, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-18s %s" % (o.kind, o.name))
            print()

        if clean:
            print("  UNUSED (no reference found in any ADOM analysed):")
            for o in sorted(clean, key=lambda x: (x.kind, x.name))[:limit]:
                print("    %-18s %s" % (o.kind, o.name))
            if len(clean) > limit:
                print("    ... and %d more (raise --limit)" % (len(clean) - limit))
            total += 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("    %-18s %-30s via %s" % (o.kind, o.name, o.used_by[0]))
            total += len(transitive)

        if show_used:
            used = [o for o in objs.values() if o.used_by and not o.transitively_dead]
            print()
            print("  IN USE (%d):" % len(used))
            for o in sorted(used, key=lambda x: (x.kind, x.name))[:limit]:
                extra = "" if len(o.used_by) == 1 else " (+%d more)" % (len(o.used_by) - 1)
                print("    %-18s %-30s %s%s" % (o.kind, o.name, o.used_by[0], extra))
        print()

        for o in clean:
            rows.append([adom, o.kind, o.name, "UNUSED", "candidate", "", ""])
        for o in unverifiable:
            rows.append([adom, o.kind, o.name, "COULD_NOT_VERIFY", "no", "",
                         "global object and not every ADOM was analysed - "
                         "re-run with --all-adoms"])
        for o in transitive:
            rows.append([adom, 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 objs.values():
            if o.used_by and not o.transitively_dead:
                rows.append([adom, 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)), ""])

    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)

    if csv_path:
        try:
            with open(csv_path, "w", encoding="utf-8-sig", newline="") as fh:
                w = csv.writer(fh)
                w.writerow(["adom", "object_type", "name", "verdict",
                            "safe_to_delete", "referenced_by", "note"])
                w.writerows(rows)
            print("CSV written: %s (%d rows)" % (csv_path, len(rows)))
        except OSError as exc:
            print("WARNING: could not write %s: %s" % (csv_path, exc))
    return total


def main():
    ap = argparse.ArgumentParser(
        description="Find objects in FortiManager ADOMs that no policy "
                    "references. Read-only; never deletes anything.")
    src = ap.add_mutually_exclusive_group(required=True)
    src.add_argument("--host", help="FortiManager hostname or IP")
    src.add_argument("--offline-dir", help="Replay a directory written by --save-dir")
    ap.add_argument("--adom", action="append", default=[],
                    help="ADOM to analyse. Repeatable. Default: root plus global")
    ap.add_argument("--all-adoms", action="store_true",
                    help="Analyse every ADOM. Required for a trustworthy verdict "
                         "on global objects")
    ap.add_argument("--insecure", action="store_true",
                    help="Skip TLS verification (known 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="Max per section (default 50)")
    ap.add_argument("--csv", metavar="PATH",
                    help="Also write every object and its verdict to a CSV")
    ap.add_argument("--save-dir", help="Write every API response here for offline re-runs")
    args = ap.parse_args()

    if args.offline_dir:
        api = Offline(args.offline_dir)
    else:
        user = os.environ.get("FMG_USER")
        password = os.environ.get("FMG_PASS")
        if not user or not password:
            raise SystemExit(
                "ERROR: FMG_USER and FMG_PASS must be set in the environment.\n"
                "\n"
                "  Linux / macOS:  export FMG_USER=api_reader\n"
                "                  export FMG_PASS='...'\n"
                "  PowerShell:     $env:FMG_USER='api_reader'\n"
                "                  $env:FMG_PASS='...'\n"
                "\n"
                "Read from the environment on purpose: an argument is visible in\n"
                "`ps auxww` and lands in shell history.\n"
                "\n"
                "The account needs JSON API access enabled on its admin profile -\n"
                "read-only is enough. It is off by default.")
        try:
            import requests
        except ImportError:
            raise SystemExit("ERROR: pip install requests")
        if args.save_dir:
            os.makedirs(args.save_dir, exist_ok=True)
        try:
            api = FortiManager(args.host, user, password, not args.insecure, args.save_dir)
        except requests.exceptions.SSLError as exc:
            raise SystemExit(
                "ERROR: TLS verification failed for %s: %s\n"
                "Use --insecure only for a certificate you recognise." % (args.host, exc))
        except requests.exceptions.RequestException as exc:
            raise SystemExit("ERROR: could not reach %s: %s" % (args.host, exc))

    try:
        if args.all_adoms:
            adoms = ["global"] + [a for a in list_adoms(api) if a != "global"]
        elif args.adom:
            adoms = args.adom
        else:
            adoms = ["global", "root"]

        # Collect every ADOM's objects BEFORE resolving any reference, so a
        # policy in one ADOM can resolve against a global object.
        collected = {}
        for adom in adoms:
            collected[adom] = collect(api, adom)
        had_objects = any(objs for objs, _ in collected.values())
        global_objs = collected.get("global", ({}, {}))[0]

        per_adom = {}
        for adom in adoms:
            objs, raw = collected[adom]
            scanned = collect_refs(api, adom, objs, raw, global_objs)
            per_adom[adom] = (objs, scanned)

        # Transitive deadness is computed across everything, once.
        all_objs = {}
        for adom, (objs, _) in per_adom.items():
            for k, o in objs.items():
                all_objs[(adom,) + k] = o
        mark_transitive(all_objs)

        if not had_objects:
            print("No objects found in %s." % ", ".join(adoms))
            print("Check the ADOM names and that the account's profile has JSON")
            print("API access with read permission on Policy & Objects.")
            return 2

        consumers = [a for a in adoms if a != "global"]
        n = report(per_adom, args.limit, args.show_used, args.csv,
                   consumers if not args.all_adoms else [])
        print("API calls made: %d" % api.calls)
        return 1 if n else 0
    finally:
        api.logout()


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