ACL / Policy Security Analyzer

#!/usr/bin/env python3
"""
ACL / Policy Security Analyzer

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/acl-analyzer

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 "netmiko>=4.3" "PyYAML>=6.0"
"""

from __future__ import annotations

import argparse
import json
import os
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from pathlib import Path

try:
    import yaml
except ImportError:
    yaml = None

from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException
import re
from collections import defaultdict
from dataclasses import dataclass, asdict
from typing import Any

# ---------------------------------------------------------------------------
# Inventory loading and connection
# ---------------------------------------------------------------------------

NETMIKO_DRIVER = {
    "ios": "cisco_ios",
    "iosxe": "cisco_ios",
    "iosxr": "cisco_xr",
    "nxos": "cisco_nxos",
    "junos": "juniper_junos",
    "eos": "arista_eos",
    "panos": "paloalto_panos",
    "fortios": "fortinet",
}


def load_inventory(path: str) -> list[dict]:
    """
    Read devices.yaml (or .json) and return a list of device dicts.

    Applies the `defaults:` block to every device, validates the `os` field,
    and fails with a readable message rather than a stack trace.
    """
    p = Path(path)
    if not p.exists():
        raise SystemExit(
            f"Inventory file not found: {path}\n"
            f"Create it with:\n\n"
            f"  devices:\n"
            f"    - host: 192.168.1.1\n"
            f"      os: ios\n"
        )

    raw = p.read_text(encoding="utf-8")
    try:
        if p.suffix.lower() in (".yaml", ".yml"):
            if yaml is None:
                raise SystemExit("PyYAML is not installed. Run:  pip install PyYAML")
            data = yaml.safe_load(raw)
        else:
            data = json.loads(raw)
    except SystemExit:
        raise
    except Exception as exc:
        raise SystemExit(f"Could not parse {path}: {exc}\n"
                         f"Check for tabs (use spaces) and missing spaces after colons.")

    if isinstance(data, dict):
        defaults = data.get("defaults", {}) or {}
        devices = data.get("devices", []) or []
    else:
        defaults, devices = {}, data

    if not devices:
        raise SystemExit(f"No devices found in {path}. The file needs a `devices:` list.")

    out = []
    for i, d in enumerate(devices, 1):
        if not isinstance(d, dict):
            raise SystemExit(f"Device #{i} in {path} is not a mapping. "
                             f"Each entry needs at least `host:` and `os:`.")
        merged = {**defaults, **d}
        if "host" not in merged:
            raise SystemExit(f"Device #{i} in {path} has no `host:` field.")
        os_key = merged.get("os")
        if not os_key:
            raise SystemExit(f"Device '{merged['host']}' has no `os:` field. "
                             f"Valid values: {', '.join(sorted(NETMIKO_DRIVER))}")
        if os_key not in NETMIKO_DRIVER:
            raise SystemExit(
                f"Device '{merged['host']}' has os '{os_key}', which is not supported.\n"
                f"Valid values: {', '.join(sorted(NETMIKO_DRIVER))}\n"
                f"Note: 'cisco' is a vendor, not an OS - use ios, iosxe, iosxr, or nxos."
            )
        merged.setdefault("vrfs", [])
        out.append(merged)
    return out


def credentials() -> tuple[str, str, str]:
    """Read credentials from the environment and fail clearly if missing."""
    user = os.environ.get("NET_USER", "")
    pwd = os.environ.get("NET_PASS", "")
    enable = os.environ.get("NET_ENABLE", "")
    if not user or not pwd:
        raise SystemExit(
            "Set NET_USER and NET_PASS in your environment before running.\n\n"
            "  Linux/macOS:   export NET_USER='admin'\n"
            "                 export NET_PASS='your-password'\n\n"
            "  PowerShell:    $env:NET_USER = 'admin'\n"
            "                 $env:NET_PASS = 'your-password'\n"
        )
    return user, pwd, enable


def connect(dev: dict):
    """Open an SSH session to one device using netmiko."""
    user, pwd, enable = credentials()
    params = {
        "device_type": NETMIKO_DRIVER[dev["os"]],
        "host": dev["host"],
        "username": dev.get("username") or user,
        "password": dev.get("password") or pwd,
        "port": dev.get("port", 22),
        "conn_timeout": 20,
        "fast_cli": False,
    }
    secret = dev.get("secret") or enable
    if secret:
        params["secret"] = secret

    conn = ConnectHandler(**params)
    if secret and dev["os"] in ("ios", "iosxe", "nxos"):
        conn.enable()
    return conn


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

ACL_CMD = {
    "ios": "show ip access-lists",
    "iosxe": "show ip access-lists",
    "iosxr": "show access-lists",
    "nxos": "show ip access-lists",
    "junos": "show configuration firewall | display set",
    "eos": "show ip access-lists",
    "fortios": "show firewall policy",
    "panos": "show running security-policy",
}

# Where the ACL is actually applied. A permissive ACL bound to nothing is a
# latent risk; the same ACL applied inbound on the internet edge is an open
# door. The two deserve different severities, so both are checked.
APPLIED_CMD = {
    "ios": "show running-config | include ip access-group|access-class",
    "iosxe": "show running-config | include ip access-group|access-class",
    "nxos": "show running-config | include ip access-group|access-class",
    "eos": "show running-config | include ip access-group|access-class",
    "iosxr": "show running-config | include ipv4 access-group",
    "junos": "show configuration | display set | match filter",
}


@dataclass
class Finding:
    severity: str          # HIGH | MEDIUM | LOW
    acl: str
    rule: str              # line number or rule name
    message: str
    applied: bool = False


# ---------------------------------------------------------------------------
# Parsers
# ---------------------------------------------------------------------------

def parse_applied(output: str) -> set[str]:
    """Return the set of ACL names bound to an interface or VTY line."""
    names: set[str] = set()
    for line in output.splitlines():
        for pat in (r"ip access-group (\S+)", r"access-class (\S+)",
                    r"ipv4 access-group (\S+)", r"filter input (\S+)",
                    r"filter output (\S+)"):
            m = re.search(pat, line)
            if m:
                names.add(m.group(1))
    return names


def parse_cisco_acls(output: str, applied: set[str]) -> list[Finding]:
    """
    Parse IOS / IOS-XE / IOS-XR / NX-OS / EOS access lists.

    Beyond spotting permit-any, this tracks whether an earlier rule already
    covers a later one. A rule that can never match is dead config, and dead
    config is where stale permissions hide.
    """
    findings: list[Finding] = []
    acl: str | None = None
    seq = 0
    broad_permit_at: int | None = None

    for line in output.splitlines():
        m = re.match(r"^(?:Extended|Standard|IPv4)\s+(?:IP\s+)?access list (\S+)",
                     line, re.IGNORECASE)
        if m:
            acl = m.group(1)
            seq = 0
            broad_permit_at = None
            continue
        if not acl or not line.strip():
            continue

        seq += 1
        s = line.strip()
        is_applied = acl in applied

        if re.search(r"permit\s+ip\s+any\s+any", s, re.IGNORECASE):
            findings.append(Finding(
                "HIGH" if is_applied else "MEDIUM", acl, str(seq),
                "permit ip any any - every protocol, every source, every destination",
                is_applied))
            if broad_permit_at is None:
                broad_permit_at = seq

        elif re.search(r"permit\s+\S+\s+any\s+any", s, re.IGNORECASE):
            findings.append(Finding(
                "MEDIUM" if is_applied else "LOW", acl, str(seq),
                "permit <proto> any any - any source to any destination", is_applied))
            if broad_permit_at is None:
                broad_permit_at = seq

        elif re.search(r"permit\s+.*\bany\b", s, re.IGNORECASE):
            findings.append(Finding(
                "LOW", acl, str(seq),
                "permit with an 'any' in source or destination", is_applied))

        # Anything after an unconditional permit can never be reached.
        if broad_permit_at is not None and seq > broad_permit_at:
            findings.append(Finding(
                "LOW", acl, str(seq),
                f"unreachable - line {broad_permit_at} already permits this traffic",
                is_applied))

    return findings


def parse_junos_filters(output: str, applied: set[str]) -> list[Finding]:
    """Flag Junos terms that accept without a from clause (match-all)."""
    findings: list[Finding] = []
    terms: dict[tuple[str, str], dict] = defaultdict(lambda: {"from": False, "accept": False})

    for line in output.splitlines():
        m = re.search(r"firewall family \S+ filter (\S+) term (\S+) (from|then) (.+)", line)
        if not m:
            continue
        key = (m.group(1), m.group(2))
        if m.group(3) == "from":
            terms[key]["from"] = True
        elif "accept" in m.group(4):
            terms[key]["accept"] = True

    for (filt, term), v in terms.items():
        if v["accept"] and not v["from"]:
            findings.append(Finding(
                "HIGH" if filt in applied else "MEDIUM", filt, term,
                "term accepts with no from clause - matches all traffic",
                filt in applied))
    return findings


def parse_fortios_policy(output: str) -> list[Finding]:
    """Flag FortiOS policies that are any/any/accept, and note disabled ones."""
    findings: list[Finding] = []
    pid: str | None = None
    state: dict[str, Any] = {}

    for line in output.splitlines():
        m = re.match(r"\s*edit (\d+)", line)
        if m:
            pid = m.group(1)
            state = {"src_all": False, "dst_all": False, "accept": False,
                     "svc_all": False, "disabled": False, "name": ""}
            continue
        if pid is None:
            continue

        if re.search(r'set name "([^"]*)"', line):
            state["name"] = re.search(r'set name "([^"]*)"', line).group(1)
        if re.search(r'set srcaddr "all"', line):
            state["src_all"] = True
        if re.search(r'set dstaddr "all"', line):
            state["dst_all"] = True
        if re.search(r'set service "ALL"', line):
            state["svc_all"] = True
        if re.search(r"set action accept", line):
            state["accept"] = True
        if re.search(r"set status disable", line):
            state["disabled"] = True

        if re.match(r"\s*next\s*$", line):
            label = f"policy {pid}" + (f" ({state['name']})" if state["name"] else "")
            if state["src_all"] and state["dst_all"] and state["accept"]:
                sev = "LOW" if state["disabled"] else "HIGH"
                extra = " [policy is disabled]" if state["disabled"] else ""
                svc = " and service ALL" if state["svc_all"] else ""
                findings.append(Finding(
                    sev, label, pid,
                    f"srcaddr all + dstaddr all{svc} + action accept{extra}",
                    not state["disabled"]))
            elif state["svc_all"] and state["accept"] and not state["disabled"]:
                findings.append(Finding(
                    "LOW", label, pid,
                    "service ALL with action accept - every port permitted", True))
            pid = None
    return findings


def parse_panos_policy(output: str) -> list[Finding]:
    """
    Flag PAN-OS rules that are any/any.

    `show running security-policy` prints each rule as a block of
    'key: value;' pairs, so the rule is accumulated until the next header.
    """
    findings: list[Finding] = []
    rule: str | None = None
    attrs: dict[str, str] = {}

    def flush() -> None:
        if not rule:
            return
        src = attrs.get("source", "")
        dst = attrs.get("destination", "")
        app = attrs.get("application", "")
        svc = attrs.get("service", "")
        action = attrs.get("action", "")
        if "allow" not in action.lower():
            return
        if "any" in src.lower() and "any" in dst.lower():
            detail = "source any + destination any"
            if "any" in app.lower():
                detail += " + application any"
            if "any" in svc.lower():
                detail += " + service any"
            findings.append(Finding("HIGH", rule, "-", detail + " with action allow", True))
        elif "any" in app.lower() and "any" in svc.lower():
            findings.append(Finding(
                "MEDIUM", rule, "-",
                "application any + service any - App-ID gives no protection here", True))

    for line in output.splitlines():
        m = re.match(r'^"?([^"\s:]+)"?\s*\{?\s*$', line.strip())
        if m and not re.search(r"[:;]", line):
            flush()
            rule = m.group(1)
            attrs = {}
            continue
        for k in ("source", "destination", "application", "service", "action"):
            g = re.search(rf"{k}:\s*([^;]+);", line, re.IGNORECASE)
            if g:
                attrs[k] = g.group(1).strip()
    flush()
    return findings


# ---------------------------------------------------------------------------
# Per-device analysis
# ---------------------------------------------------------------------------

def analyse_device(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    conn = None
    try:
        conn = connect(dev)

        applied: set[str] = set()
        if APPLIED_CMD.get(os_key) and not args.skip_applied_check:
            try:
                applied = parse_applied(
                    conn.send_command(APPLIED_CMD[os_key], read_timeout=90))
            except Exception:  # noqa: BLE001 - binding check is an enrichment
                applied = set()

        raw = conn.send_command(ACL_CMD[os_key], read_timeout=args.read_timeout)

        if os_key in ("ios", "iosxe", "iosxr", "nxos", "eos"):
            findings = parse_cisco_acls(raw, applied)
        elif os_key == "junos":
            findings = parse_junos_filters(raw, applied)
        elif os_key == "fortios":
            findings = parse_fortios_policy(raw)
        elif os_key == "panos":
            findings = parse_panos_policy(raw)
        else:
            findings = []

        if args.applied_only:
            findings = [f for f in findings if f.applied]

        rank = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
        findings.sort(key=lambda f: (rank[f.severity], f.acl, f.rule))
        return {"host": host, "os": os_key, "status": "ok",
                "findings": findings, "applied_acls": sorted(applied)}

    except NetmikoAuthenticationException as exc:
        return {"host": host, "os": os_key, "status": "error", "findings": [],
                "error": f"authentication failed - check NET_USER/NET_PASS ({exc})"}
    except NetmikoTimeoutException as exc:
        return {"host": host, "os": os_key, "status": "error", "findings": [],
                "error": f"unreachable or SSH timed out ({exc})"}
    except Exception as exc:  # noqa: BLE001 - one device must not abort the run
        return {"host": host, "os": os_key, "status": "error", "findings": [],
                "error": f"{type(exc).__name__}: {exc}"}
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Find overly permissive and unreachable rules in ACLs and "
                    "firewall policies.",
        epilog="Credentials come from NET_USER and NET_PASS. See the docstring at "
               "the top of this file for the full inventory format.",
    )
    ap.add_argument("--inventory", required=True,
                    help="Path to devices.yaml listing which devices to analyse")
    ap.add_argument("--limit", help="Only analyse hosts whose name contains this text")
    ap.add_argument("--applied-only", action="store_true",
                    help="Report only rules in an ACL that is actually bound to an "
                         "interface or VTY line. Cuts the noise from unused ACLs")
    ap.add_argument("--skip-applied-check", action="store_true",
                    help="Do not read the running config to see where ACLs are "
                         "applied. One fewer command, but severity is less accurate")
    ap.add_argument("--min-severity", choices=["HIGH", "MEDIUM", "LOW"], default="LOW",
                    help="Only report findings at or above this severity. Default: LOW")
    ap.add_argument("--read-timeout", type=int, default=120,
                    help="Seconds to wait per command. Default: 120")
    ap.add_argument("--workers", type=int, default=10,
                    help="How many devices to contact at once. Default: 10")
    ap.add_argument("--json-out", help="Write machine-readable findings here")
    args = ap.parse_args()

    devices = load_inventory(args.inventory)
    if args.limit:
        devices = [d for d in devices if args.limit in d["host"]]
        if not devices:
            raise SystemExit(f"No device name contains '{args.limit}'.")

    credentials()  # fail fast before opening any connection

    print(f"Analysing ACLs and firewall policies on {len(devices)} device(s)...")
    print("=" * 84)

    results = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(analyse_device, d, args): d["host"] for d in devices}
        for fut in as_completed(futures):
            host = futures[fut]
            try:
                results.append(fut.result())
            except Exception as exc:  # noqa: BLE001
                results.append({"host": host, "os": "?", "status": "error",
                                "findings": [], "error": f"worker crashed: {exc}"})

    rank = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
    floor = rank[args.min_severity]
    tally = {"HIGH": 0, "MEDIUM": 0, "LOW": 0}

    for r in sorted(results, key=lambda x: x["host"]):
        label = f"{r['host']} ({r['os']})"
        if r["status"] == "error":
            print(f"[ERROR] {label:<26} {r['error']}")
            continue

        shown = [f for f in r["findings"] if rank[f.severity] <= floor]
        if not shown:
            n_applied = len(r.get("applied_acls", []))
            print(f"[OK]    {label:<26} no permissive rules "
                  f"({n_applied} ACL(s) applied to interfaces)")
            continue

        print(f"\n{label}  -  {len(shown)} finding(s)")
        for f in shown:
            tally[f.severity] += 1
            mark = "applied" if f.applied else "NOT applied anywhere"
            print(f"  [{f.severity:<6}] {f.acl} line {f.rule}  ({mark})")
            print(f"            {f.message}")

    print("\n" + "=" * 84)
    print(f"{tally['HIGH']} HIGH, {tally['MEDIUM']} MEDIUM, {tally['LOW']} LOW "
          f"across {len(devices)} device(s).")
    if tally["HIGH"]:
        print("\nHIGH means the rule is permissive AND the ACL is bound to an "
              "interface - it is live.")

    if args.json_out:
        payload = [{"host": r["host"], "os": r["os"], "status": r["status"],
                    "error": r.get("error"),
                    "applied_acls": r.get("applied_acls", []),
                    "findings": [asdict(f) for f in r["findings"]]}
                   for r in results]
        Path(args.json_out).write_text(json.dumps(payload, indent=2), encoding="utf-8")

    err = sum(1 for r in results if r["status"] == "error")
    return 1 if (tally["HIGH"] or err) else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted.", file=sys.stderr)
        sys.exit(130)