FMC Unused Object Finder

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

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/fmc-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
import time

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

# FMC accepts 120 GET requests per minute from one IP. Two per second with a
# little headroom keeps a large estate inside that without thinking about it.
RATE_PER_SEC = 1.8

# Object classes to inventory: (api path, human label, covered by the native
# unusedOnly filter?). The third column is the whole point of this tool - the
# native filter is documented for network addresses only, so everything with
# False is a gap FMC does not fill.
OBJECT_CLASSES = [
    ("networks",             "network",            True),
    ("hosts",                "host",               True),
    ("ranges",               "range",              True),
    ("fqdns",                "fqdn",               True),
    ("networkgroups",        "network group",      False),
    ("protocolportobjects",  "port",               False),
    ("icmpv4objects",        "icmpv4",             False),
    ("portobjectgroups",     "port group",         False),
    ("urls",                 "url",                False),
    ("urlgroups",            "url group",          False),
    ("vlantags",             "vlan tag",           False),
    ("vlangrouptags",        "vlan group",         False),
    ("securityzones",        "security zone",      False),
]

# Fields inside a rule that can name an object. FMC nests them as
# {"objects": [...], "literals": [...]}.
RULE_REF_FIELDS = [
    "sourceNetworks", "destinationNetworks", "sourcePorts", "destinationPorts",
    "sourceZones", "destinationZones", "urls", "vlanTags", "originalSource",
    "originalDestination", "translatedSource", "translatedDestination",
    "originalPort", "translatedPort", "sourceDynamicObjects",
    "destinationDynamicObjects", "interfaceInOriginalDestination",
    "interfaceInTranslatedSource",
]

NOT_CHECKED = [
    "FlexConfig objects and FlexConfig policies",
    "Remote access and site-to-site VPN topologies",
    "Platform settings, health policies and variable sets",
    "Objects referenced only by a device's local configuration outside FMC",
    "Identity policies, SSL policies and DNS policies",
    "Anything referenced by an external system through the API",
]


class Throttle:
    def __init__(self, per_sec):
        self.interval = 1.0 / per_sec
        self.last = 0.0

    def wait(self):
        delta = time.time() - self.last
        if delta < self.interval:
            time.sleep(self.interval - delta)
        self.last = time.time()


class FMC:
    """FMC REST client: token auth, throttling, pagination.

    Token lifetime is 30 minutes. On a large estate a full run can exceed
    that, so a 401 triggers one re-authentication rather than failing.
    """

    def __init__(self, host, user, password, verify=True, save_dir=None):
        import requests
        import urllib3
        self.requests = requests
        self.host = host
        self.base = "https://%s" % host
        self.user = user
        self.password = password
        self.verify = verify
        self.save_dir = save_dir
        self.throttle = Throttle(RATE_PER_SEC)
        self.calls = 0
        if not verify:
            urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
        self.s = requests.Session()
        self.s.verify = verify
        self._auth()

    def _auth(self):
        r = self.s.post(
            self.base + "/api/fmc_platform/v1/auth/generatetoken",
            auth=(self.user, self.password), timeout=60,
        )
        if r.status_code == 401:
            raise SystemExit("ERROR: %s rejected the credentials (401)." % self.host)
        if r.status_code == 429:
            raise SystemExit(
                "ERROR: too many active tokens for this user. FMC allows a "
                "limited number at once; wait for them to expire or use a "
                "dedicated API account.")
        r.raise_for_status()
        tok = r.headers.get("X-auth-access-token")
        if not tok:
            raise SystemExit("ERROR: FMC did not return an access token.")
        self.s.headers.update({"X-auth-access-token": tok})
        self.default_domain = r.headers.get("DOMAIN_UUID")

    def get(self, path, params=None, retry=True):
        self.throttle.wait()
        r = self.s.get(self.base + path, params=params or {}, timeout=120)
        self.calls += 1
        if r.status_code == 401 and retry:
            self._auth()
            return self.get(path, params, retry=False)
        if r.status_code == 429:
            time.sleep(20)
            return self.get(path, params, retry=False)
        if r.status_code == 404:
            return None
        r.raise_for_status()
        return r.json()

    def get_all(self, path, params=None):
        """Walk FMC pagination. expanded=true so objects arrive whole."""
        out, offset, limit = [], 0, 1000
        while True:
            p = dict(params or {})
            p.update({"offset": offset, "limit": limit, "expanded": "true"})
            data = self.get(path, p)
            if data is None:
                return out
            items = data.get("items") or []
            out.extend(items)
            paging = data.get("paging") or {}
            count = paging.get("count", len(out))
            offset += limit
            if offset >= count or not items:
                break
        if self.save_dir:
            self._save(path, params, out)
        return out

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


class Offline:
    """Replay a directory produced by --save-dir. No network, no credentials.

    This exists so the logic can be tested and reviewed without an FMC, and
    so somebody can collect once and analyse repeatedly.
    """

    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 not f.endswith(".json"):
                continue
            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))
            key = (blob.get("path"), json.dumps(blob.get("params") or {}, sort_keys=True))
            self.store[key] = blob.get("items") or []
        if not self.store:
            raise SystemExit("ERROR: no saved responses found in %s" % directory)
        self.default_domain = None

    def get(self, path, params=None):
        return {"items": self.get_all(path, params)}

    def get_all(self, path, params=None):
        self.calls += 1
        return self.store.get((path, json.dumps(params or {}, sort_keys=True)), [])


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

    def __init__(self, name, kind, uuid):
        self.name = name
        self.kind = kind
        self.uuid = uuid
        self.used_by = []
        self.native_unused = None     # True / False / None = not applicable
        self.transitively_dead = False


def walk_refs(node, out, where):
    """Collect every object name/id a rule field names, at any nesting depth."""
    if isinstance(node, dict):
        if "name" in node and ("id" in node or "type" in node):
            out.append((node.get("name"), node.get("id"), where))
        for v in node.values():
            walk_refs(v, out, where)
    elif isinstance(node, list):
        for v in node:
            walk_refs(v, out, where)


def collect(api, domain_uuid, use_native):
    base = "/api/fmc_config/v1/domain/%s" % domain_uuid
    objs = {}

    for path, kind, native_covered in OBJECT_CLASSES:
        items = api.get_all(base + "/object/" + path)
        for it in items:
            # Skip FMC's own read-only system objects: they cannot be deleted
            # and listing them as unused is noise.
            if it.get("overridable") is None and it.get("metadata", {}).get("readOnly"):
                continue
            name, uid = it.get("name"), it.get("id")
            if not name:
                continue
            o = Obj(name, kind, uid)
            # Only claim a class is covered when the filter is actually being
            # consulted. Marking it covered without asking made those objects
            # fall into a section that is not printed in --no-native mode, so
            # they vanished from the report entirely.
            if native_covered and use_native:
                o.native_unused = False
            objs[uid or (kind, name)] = o

        # Nested group membership - the class home-grown scripts miss.
        for it in items:
            for member in (it.get("objects") or []):
                refs = [(member.get("name"), member.get("id"),
                         "%s '%s'" % (kind, it.get("name")))]
                for nm, uid2, where in refs:
                    key = uid2 or (kind, nm)
                    if key in objs:
                        objs[key].used_by.append(where)

    # The native filter, used rather than reimplemented.
    if use_native:
        for path, kind, native_covered in OBJECT_CLASSES:
            if not native_covered:
                continue
            items = api.get_all(base + "/object/" + path,
                                {"filter": "unusedOnly:true"})
            flagged = {it.get("id") for it in items}
            for key, o in objs.items():
                if o.kind == kind and o.uuid in flagged:
                    o.native_unused = True
    return objs


def collect_policy_refs(api, domain_uuid, objs):
    base = "/api/fmc_config/v1/domain/%s" % domain_uuid
    by_uuid = {o.uuid: o for o in objs.values() if o.uuid}
    by_name = {}
    for o in objs.values():
        by_name.setdefault(o.name, []).append(o)

    def record(name, uid, where):
        if uid and uid in by_uuid:
            by_uuid[uid].used_by.append(where)
        elif name:
            for o in by_name.get(name, []):
                o.used_by.append(where)

    scanned = []

    for pol in api.get_all(base + "/policy/accesspolicies"):
        pid, pname = pol.get("id"), pol.get("name")
        rules = api.get_all(base + "/policy/accesspolicies/%s/accessrules" % pid)
        scanned.append("access policy '%s' (%d rules)" % (pname, len(rules)))
        for rule in rules:
            where = "access rule '%s' in '%s'" % (rule.get("name"), pname)
            found = []
            for field in RULE_REF_FIELDS:
                if field in rule:
                    walk_refs(rule[field], found, where)
            for nm, uid, w in found:
                record(nm, uid, w)

    for pol in api.get_all(base + "/policy/prefilterpolicies"):
        pid, pname = pol.get("id"), pol.get("name")
        rules = api.get_all(base + "/policy/prefilterpolicies/%s/prefilterrules" % pid)
        scanned.append("prefilter policy '%s' (%d rules)" % (pname, len(rules)))
        for rule in rules:
            where = "prefilter rule '%s' in '%s'" % (rule.get("name"), pname)
            found = []
            for field in RULE_REF_FIELDS:
                if field in rule:
                    walk_refs(rule[field], found, where)
            for nm, uid, w in found:
                record(nm, uid, w)

    for pol in api.get_all(base + "/policy/ftdnatpolicies"):
        pid, pname = pol.get("id"), pol.get("name")
        for sub, label in (("autonatrules", "auto NAT"), ("manualnatrules", "manual NAT")):
            rules = api.get_all(base + "/policy/ftdnatpolicies/%s/%s" % (pid, sub))
            if rules:
                scanned.append("%s in '%s' (%d rules)" % (label, pname, len(rules)))
            for rule in rules:
                where = "%s rule in '%s'" % (label, pname)
                found = []
                for field in RULE_REF_FIELDS:
                    if field in rule:
                        walk_refs(rule[field], found, where)
                for nm, uid, w in found:
                    record(nm, uid, w)

    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


CSV_HEADER = ["object_type", "name", "uuid", "verdict", "safe_to_delete",
              "fmc_native_filter", "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 native_col(o):
    if o.native_unused is True:
        return "says unused"
    if o.native_unused is False:
        return "did not flag"
    return "not covered"


def report(objs, scanned, use_native, limit, show_used, csv_path=None):
    unused = [o for o in objs.values() if not o.used_by or o.transitively_dead]
    # The dangerous disagreement: FMC's own filter says unused, but this tool
    # found a reference. Somebody acting on the GUI list would delete it.
    contested = [o for o in objs.values()
                 if o.native_unused is True and o.used_by and not o.transitively_dead]
    confirmed, native_only, script_only, transitive = [], [], [], []

    for o in unused:
        if o.transitively_dead:
            transitive.append(o)
        elif o.native_unused is True:
            confirmed.append(o)          # FMC agrees and so do we
        elif o.native_unused is False:
            script_only.append(o)        # we say unused, FMC's filter did not
        else:
            native_only.append(o)        # no native filter for this class

    print("=" * 78)
    print("fmc-unused-objects")
    print("=" * 78)
    print("Objects inventoried: %d" % len(objs))
    print()
    print("Policies scanned:")
    for s in scanned or ["  (none found)"]:
        print("  - %s" % s)
    print()

    def block(title, items, note=None):
        if not items:
            return 0
        print("-" * 78)
        print(title)
        if note:
            for line in note:
                print("  %s" % line)
        print("-" * 78)
        for o in sorted(items, key=lambda x: (x.kind, x.name))[:limit]:
            print("  %-16s %s" % (o.kind, o.name))
        if len(items) > limit:
            print("  ... and %d more (raise --limit)" % (len(items) - limit))
        print()
        return len(items)

    native_header = ("UNUSED - object classes FMC's filter does not cover"
                     if use_native else
                     "UNUSED - no reference found (native filter not consulted)")
    native_note = (["Groups, ports, URLs, zones and VLAN tags. The native filter is",
                    "documented for network addresses only, so for these there is no",
                    "second opinion to compare against."]
                   if use_native else
                   ["--no-native was passed, so there is no cross-check at all.",
                    "Every result here rests on this tool's reference walk alone."])

    total = 0
    if contested:
        print("!" * 78)
        print("DO NOT DELETE - FMC's filter calls these unused, but this tool")
        print("found a reference to each one.")
        print("!" * 78)
        print("  Anyone acting on the GUI's Show Unused Objects list would remove")
        print("  these. Confirm by hand before touching them, and treat it as a")
        print("  reason to distrust the list rather than this tool.")
        for o in sorted(contested, key=lambda x: (x.kind, x.name))[:limit]:
            print("  %-16s %-30s referenced by: %s" % (o.kind, o.name, o.used_by[0]))
        print()

    if use_native:
        total += block(
            "UNUSED - confirmed by BOTH this tool and FMC's own unusedOnly filter",
            confirmed,
            ["The strongest result available. Two independent methods agree."])
        total += block(
            "UNUSED - found by this tool, NOT flagged by FMC's filter",
            script_only,
            ["Investigate these first. Either the filter scopes differently, or",
             "something references it that this tool did not read. Treat the",
             "disagreement as a question, not as a finding."])
    total += block(native_header, native_only, native_note)
    total += block(
        "UNUSED TRANSITIVELY - referenced only by groups that are themselves unused",
        transitive,
        ["Kept alive by a corpse. Delete the container first, then re-run."])

    if show_used:
        used = [o for o in objs.values() if o.used_by and not o.transitively_dead]
        print("-" * 78)
        print("IN USE (%d)" % len(used))
        print("-" * 78)
        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("  %-16s %-34s %s%s" % (o.kind, o.name, o.used_by[0], extra))
        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()
    if csv_path:
        rows = []
        for o in contested:
            rows.append([o.kind, o.name, o.uuid or "", "DO_NOT_DELETE", "NO",
                         native_col(o), o.used_by[0],
                         "FMC's filter calls this unused but a policy "
                         "references it - resolve before touching it"])
        for o in confirmed:
            rows.append([o.kind, o.name, o.uuid or "", "UNUSED", "candidate",
                         native_col(o), "",
                         "both this tool and FMC's filter agree"])
        for o in script_only:
            rows.append([o.kind, o.name, o.uuid or "", "UNUSED", "candidate",
                         native_col(o), "",
                         "this tool found no reference; FMC's filter did not "
                         "flag it - investigate the disagreement"])
        for o in native_only:
            rows.append([o.kind, o.name, o.uuid or "", "UNUSED", "candidate",
                         native_col(o), "",
                         "no native second opinion exists for this class"])
        for o in transitive:
            rows.append([o.kind, o.name, o.uuid or "", "UNUSED_TRANSITIVE",
                         "candidate", native_col(o),
                         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 and o not in contested:
                rows.append([o.kind, o.name, o.uuid or "", "IN_USE", "no",
                             native_col(o), o.used_by[0]
                             + ("" if len(o.used_by) == 1
                                else " (+%d more)" % (len(o.used_by) - 1)), ""])
        write_csv(csv_path, rows)

    print("Total candidates: %d" % total)
    return total


def main():
    ap = argparse.ArgumentParser(
        description="Find objects in Cisco Secure Firewall Management Center "
                    "that no policy references. Read-only; never deletes.")
    src = ap.add_mutually_exclusive_group(required=True)
    src.add_argument("--host", help="FMC hostname or IP")
    src.add_argument("--offline-dir",
                     help="Replay a directory written by --save-dir. No network "
                          "and no credentials at all.")
    ap.add_argument("--domain", help="Domain name or UUID (default: the account's)")
    ap.add_argument("--insecure", action="store_true",
                    help="Skip TLS verification (known self-signed certificate only)")
    ap.add_argument("--no-native", action="store_true",
                    help="Do not use FMC's own unusedOnly filter, so you lose "
                         "the cross-check")
    ap.add_argument("--show-used", action="store_true",
                    help="Also list the 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, "
                         "for the spreadsheet the cleanup actually lives in")
    ap.add_argument("--save-dir", help="Write every API response here for offline re-runs")
    args = ap.parse_args()

    use_native = not args.no_native

    if args.offline_dir:
        api = Offline(args.offline_dir)
        domain_uuid = args.domain or "OFFLINE"
    else:
        user = os.environ.get("FMC_USER")
        password = os.environ.get("FMC_PASS")
        if not user or not password:
            raise SystemExit(
                "ERROR: FMC_USER and FMC_PASS must be set in the environment.\n"
                "\n"
                "  Linux / macOS:  export FMC_USER=api_reader\n"
                "                  export FMC_PASS='...'\n"
                "  PowerShell:     $env:FMC_USER='api_reader'\n"
                "                  $env:FMC_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 REST access. A read-only role is enough.")
        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 = FMC(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))

        domain_uuid = args.domain or api.default_domain
        if not domain_uuid:
            domains = (api.get("/api/fmc_platform/v1/info/domain") or {}).get("items", [])
            if not domains:
                raise SystemExit("ERROR: could not determine the domain UUID.")
            domain_uuid = domains[0]["uuid"]
        if args.domain and not re.match(r"^[0-9a-fA-F-]{36}$", args.domain):
            domains = (api.get("/api/fmc_platform/v1/info/domain") or {}).get("items", [])
            match = [d for d in domains if d.get("name") == args.domain]
            if not match:
                raise SystemExit(
                    "ERROR: no domain named %s. Available: %s"
                    % (args.domain, ", ".join(d.get("name", "?") for d in domains)))
            domain_uuid = match[0]["uuid"]

    objs = collect(api, domain_uuid, use_native)
    if not objs:
        print("No objects found. Check the domain and that the account's role")
        print("allows reading object management.")
        return 2
    scanned = collect_policy_refs(api, domain_uuid, objs)
    mark_transitive(objs)
    n = report(objs, scanned, use_native, args.limit, args.show_used, args.csv)
    print("API calls made: %d" % api.calls)
    return 1 if n else 0


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