IPv6 Security Audit

#!/usr/bin/env python3
"""
IPv6 Security Audit

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/ipv6-security-audit

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.0" pyyaml
"""

# SAFE: read-only. Every command it sends is a show/display. It never enters
# configuration mode and has no code path that writes to a device.
#
# EXIT CODES  0 nothing found   1 findings   2 the run failed
#
# CREDENTIALS come from the environment, never from an argument:
#   NET_USER / NET_PASS / NET_ENABLE   (not needed at all with --config-dir)

import argparse
import csv
import os
import re
import sys

try:
    sys.stdout.reconfigure(encoding="utf-8", errors="replace")
except Exception:
    pass

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

# The one command per platform that carries everything this script reads.
CONFIG_CMD = {
    "ios": "show running-config", "iosxe": "show running-config",
    "iosxr": "show running-config", "nxos": "show running-config",
    "junos": "show configuration | display set", "eos": "show running-config",
    "panos": "show config running", "fortios": "show full-configuration",
}

SEV_ORDER = {"HIGH": 0, "MEDIUM": 1, "LOW": 2, "COULD_NOT_VERIFY": 3}


# ----------------------------------------------------------------- inventory
def load_inventory(path):
    try:
        import yaml
    except ImportError:
        raise SystemExit("PyYAML is not installed. Run: pip install pyyaml")
    try:
        with open(path, encoding="utf-8") as fh:
            doc = yaml.safe_load(fh) or {}
    except OSError as exc:
        raise SystemExit("could not read the inventory: %s" % exc)

    defaults = doc.get("defaults") or {}
    devices = doc.get("devices")
    if not devices:
        raise SystemExit("the inventory has no 'devices:' list")

    out = []
    for i, raw in enumerate(devices, 1):
        d = dict(defaults)
        d.update(raw or {})
        if not d.get("host"):
            raise SystemExit("device %d has no 'host'" % i)
        osn = str(d.get("os", "")).lower()
        if osn not in NETMIKO_DRIVER:
            raise SystemExit(
                "device %s has os '%s'. Valid values: %s"
                % (d["host"], osn or "(missing)", ", ".join(sorted(NETMIKO_DRIVER))))
        d["os"] = osn
        d["port"] = int(d.get("port", 22))
        d["device_name"] = d.get("device_name") or d["host"]
        out.append(d)
    return out


def credentials():
    user, pwd = os.environ.get("NET_USER"), os.environ.get("NET_PASS")
    if not user or not pwd:
        raise SystemExit(
            "NET_USER and NET_PASS must be set in the environment.\n"
            "  Linux/macOS : export NET_USER=admin; read -s NET_PASS; export NET_PASS\n"
            "  PowerShell  : $env:NET_USER='admin'; $env:NET_PASS='...'\n"
            "They are read from the environment on purpose: an argument is visible "
            "in 'ps auxww' and lands in shell history.")
    return user, pwd, os.environ.get("NET_ENABLE") or pwd


def connect(device):
    try:
        from netmiko import ConnectHandler
    except ImportError:
        raise SystemExit("netmiko is not installed. Run: pip install 'netmiko>=4.0'")
    user, pwd, secret = credentials()
    return ConnectHandler(
        device_type=NETMIKO_DRIVER[device["os"]], host=device["host"],
        port=device["port"], username=user, password=pwd, secret=secret,
        fast_cli=False, conn_timeout=20)


# ----------------------------------------------------------------- findings
class Finding(object):
    def __init__(self, device, severity, check, where, detail, fix=""):
        self.device = device
        self.severity = severity
        self.check = check
        self.where = where
        self.detail = detail
        self.fix = fix


# ----------------------------------------------------------------- parsing
def split_interfaces_ios(cfg):
    """Return {name: [lines]} for IOS-like configs (IOS, IOS-XE, NX-OS, EOS)."""
    out, name, buf = {}, None, []
    for line in cfg.split("\n"):
        m = re.match(r"^interface\s+(\S.*?)\s*$", line)
        if m:
            if name:
                out[name] = buf
            name, buf = m.group(1), []
            continue
        if name is not None:
            if line and not line.startswith((" ", "\t")):
                out[name] = buf
                name, buf = None, []
            else:
                buf.append(line.strip())
    if name:
        out[name] = buf
    return out


def split_interfaces_junos(cfg):
    """Junos `display set`: one flat line per statement."""
    out = {}
    for line in cfg.split("\n"):
        m = re.match(r"^set interfaces (\S+)\s+(.*)$", line.strip())
        if m:
            out.setdefault(m.group(1), []).append(m.group(2))
    return out


def is_point_to_point(lines):
    text = " ".join(lines).lower()
    return "point-to-point" in text or "/127" in text or "/31" in text


# ----------------------------------------------------------------- checks
def audit_ios_like(name, osn, cfg, out):
    """IOS, IOS-XE, NX-OS, EOS and (partly) IOS-XR share enough syntax."""
    low = cfg.lower()
    ifaces = split_interfaces_ios(cfg)

    routing_on = bool(re.search(r"^ipv6 unicast-routing", cfg, re.M)) or \
        bool(re.search(r"^feature\s+.*\n|^ipv6 unicast-routing", cfg, re.M))

    # ---- 1. the finding that matters most: v4 filtered, v6 not
    for ifname, lines in ifaces.items():
        raw = " ".join(lines)
        joined = raw.lower()
        has_v6_addr = bool(re.search(r"\bipv6 address\b", joined)) or "ipv6 enable" in joined
        # Matched case-insensitively but reported from the ORIGINAL text. An ACL
        # name is case-sensitive on the device, and printing "users_in" when the
        # config says "USERS_IN" sends someone looking for an ACL that does not
        # exist.
        v4_acl = re.search(r"\bip access-group\s+(\S+)\s+(in|out)", raw, re.I)
        v6_acl = re.search(r"\bipv6 traffic-filter\s+(\S+)\s+(in|out)", raw, re.I)
        if v4_acl and not v6_acl:
            out.append(Finding(
                name, "HIGH", "acl-parity", ifname,
                "an IPv4 ACL (%s %s) is applied and there is no IPv6 traffic-filter. "
                "The host behind this port already has IPv6 whether anyone configured "
                "it or not, so this interface is filtered on one family and open on "
                "the other." % (v4_acl.group(1), v4_acl.group(2)),
                "ipv6 traffic-filter <name> %s" % v4_acl.group(2)))
        elif has_v6_addr and not v6_acl and not v4_acl:
            out.append(Finding(
                name, "LOW", "no-filter", ifname,
                "IPv6 is configured and no filter is applied in either family.",
                "intentional on a transit link; a finding on a user segment"))

        # ---- 2. first-hop security on access ports
        if "switchport mode access" in joined or "switchport access vlan" in joined:
            if "ipv6 nd raguard" not in joined and "nd raguard" not in low:
                out.append(Finding(
                    name, "HIGH", "ra-guard", ifname,
                    "access port with no RA Guard. Any host on it can become the "
                    "default gateway for the whole segment by sending a Router "
                    "Advertisement, and most rogue RAs are accidents rather than "
                    "attacks: a laptop with connection sharing, a hypervisor bridged "
                    "the wrong way, a phone tethering.",
                    "ipv6 nd raguard attach-policy <policy>"))
            if "ipv6 dhcp guard" not in joined and "dhcp guard" not in low:
                out.append(Finding(
                    name, "MEDIUM", "dhcpv6-guard", ifname,
                    "access port with no DHCPv6 Guard.",
                    "ipv6 dhcp guard attach-policy <policy>"))

        # ---- 3. /64 where a /127 belongs
        for ln in lines:
            m = re.search(r"ipv6 address\s+\S+/(\d+)", ln, re.I)
            if m and int(m.group(1)) == 64 and is_point_to_point(lines):
                out.append(Finding(
                    name, "MEDIUM", "p2p-prefix", ifname,
                    "a /64 on what looks like a point-to-point link. A /64 holds 2^64 "
                    "addresses, so sweeping it fills the neighbour cache and takes the "
                    "device down. There is no IPv4 equivalent at this scale.",
                    "use a /127 (RFC 6164)"))

        # ---- 4. a router that accepts RAs
        if has_v6_addr and routing_on and "ipv6 nd ra suppress" not in joined:
            if re.search(r"^(vlan|loopback)", ifname, re.I) is None:
                out.append(Finding(
                    name, "LOW", "ra-accept", ifname,
                    "a routed interface that does not suppress RAs. A router that "
                    "processes a Router Advertisement can install a default route it "
                    "was never meant to have.",
                    "ipv6 nd ra suppress all, where no hosts need the RA"))

    # ---- 5. the RA Guard bypass
    if "nd raguard" in low and "undetermined-transport" not in low:
        out.append(Finding(
            name, "HIGH", "rfc6980", "global",
            "RA Guard is configured but nothing drops IPv6 fragments carrying "
            "Neighbour Discovery. RA Guard inspects the first fragment and the ICMPv6 "
            "type can be placed in a later one, so a fragmented RA walks straight "
            "through (RFC 7113). Without the RFC 6980 drop, RA Guard is theatre.",
            "deny ipv6 fragments carrying ND, or the platform's undetermined-transport knob"))

    # ---- 6. ICMPv6 filtered to death.
    #
    # Only ACLs that are actually APPLIED as a traffic-filter count. An ACL
    # sitting in the configuration filters nothing, and the first version of
    # this check flagged a purpose-built ND-fragment drop list as if it were
    # blocking all of ICMPv6.
    applied = {m.group(1).lower() for m in
               re.finditer(r"\bipv6 traffic-filter\s+(\S+)\s+(?:in|out)", cfg, re.I)}
    for m in re.finditer(r"^ipv6 access-list\s+(\S+)(.*?)(?=^\S|\Z)", cfg, re.M | re.S):
        aclname, body = m.group(1), m.group(2).lower()
        if aclname.lower() not in applied:
            continue
        if "icmp" not in body and re.search(r"\bdeny\s+ipv6\s+any\s+any", body):
            out.append(Finding(
                name, "HIGH", "icmpv6-blocked", "ipv6 access-list " + aclname,
                "an IPv6 ACL with an explicit deny-all and no ICMPv6 permits. ICMPv6 "
                "is not optional: Neighbour Discovery, Router Advertisements, MLD and "
                "Packet Too Big all run on it. Blocking type 2 is the worst of them, "
                "because IPv6 routers never fragment, so PMTUD has no fallback and "
                "large packets vanish.",
                "permit ICMPv6 per RFC 4890: types 1, 2, 3, 4, 128, 129 end to end; "
                "133-137 on the link"))

    # ---- 7. the binding table everything else depends on
    if "nd raguard" in low or "ipv6 source-guard" in low:
        if "ipv6 snooping" not in low and "ipv6 nd inspection" not in low:
            out.append(Finding(
                name, "MEDIUM", "no-binding-table", "global",
                "first-hop security features are configured without IPv6 snooping. "
                "Source Guard and Destination Guard are only as good as the binding "
                "table, and without snooping there is not one.",
                "ipv6 snooping policy, attached to the access ports"))

    if routing_on and "nd raguard" not in low:
        out.append(Finding(
            name, "HIGH", "no-fhs", "global",
            "IPv6 routing is enabled and no first-hop security appears anywhere in "
            "the configuration.",
            "start with IPv6 snooping, then RA Guard and DHCPv6 Guard"))


def audit_junos(name, cfg, out):
    ifaces = split_interfaces_junos(cfg)
    low = cfg.lower()
    for ifname, lines in ifaces.items():
        joined = " ".join(lines).lower()
        has_v6 = "family inet6" in joined
        v4_filter = "family inet filter" in joined
        v6_filter = "family inet6 filter" in joined
        if v4_filter and not v6_filter:
            out.append(Finding(
                name, "HIGH", "acl-parity", ifname,
                "family inet has a filter and family inet6 does not. Filtered on one "
                "family, open on the other.",
                "set interfaces %s unit 0 family inet6 filter input <name>" % ifname))
        if has_v6 and "/64" in joined and is_point_to_point(lines):
            out.append(Finding(
                name, "MEDIUM", "p2p-prefix", ifname,
                "a /64 on a point-to-point link exposes the neighbour cache.",
                "use a /127 (RFC 6164)"))
    if "family inet6" in low and "router-advertisement-guard" not in low:
        out.append(Finding(
            name, "MEDIUM", "ra-guard", "global",
            "IPv6 is configured and no RA Guard appears in the configuration.",
            "set forwarding-options access-security router-advertisement-guard"))


# ---- what each platform's rules can and cannot see, stated rather than implied
COVERAGE = {
    "ios":    "ACL parity, RA Guard, DHCPv6 Guard, snooping, /127, ICMPv6, RFC 6980",
    "iosxe":  "ACL parity, RA Guard, DHCPv6 Guard, snooping, /127, ICMPv6, RFC 6980",
    "nxos":   "ACL parity, /127, ICMPv6. First-hop security naming differs; verify by hand",
    "eos":    "ACL parity, /127, ICMPv6. First-hop security naming differs; verify by hand",
    "iosxr":  "ACL parity, /127, ICMPv6 only",
    "junos":  "filter parity per family, /127, RA Guard presence",
    "panos":  "NOT PARSED: policy is XML and the parity question needs the merged "
              "effective policy, which the device alone does not give",
    "fortios": "NOT PARSED: see panos",
}


def audit(name, osn, cfg):
    out = []
    if osn in ("ios", "iosxe", "nxos", "eos", "iosxr"):
        audit_ios_like(name, osn, cfg, out)
    elif osn == "junos":
        audit_junos(name, cfg, out)
    else:
        out.append(Finding(
            name, "COULD_NOT_VERIFY", "platform", osn,
            COVERAGE.get(osn, "no rules for this platform"),
            "audit this one by hand, or use firewall-policy-auditor"))
    return out


# ----------------------------------------------------------------- report
def report(findings, devices_seen, args):
    print("=" * 74)
    print("ipv6-security-audit")
    print("=" * 74)
    print("Devices: %s" % ", ".join(devices_seen) if devices_seen else "none")
    print("")

    if not findings:
        print("No findings. That is not the same as secure -- read NOT CHECKED below.")
    order = sorted(findings, key=lambda f: (SEV_ORDER[f.severity], f.device, f.check))
    cur = None
    for f in order:
        if f.severity != cur:
            cur = f.severity
            n = len([x for x in order if x.severity == cur])
            print("\n%s (%d)" % (cur, n))
            print("-" * 74)
        print("  [%s] %s  %s" % (f.device, f.check, f.where))
        print("      %s" % f.detail)
        if f.fix:
            print("      fix: %s" % f.fix)

    print("\nWHAT THIS DID NOT CHECK")
    print("-" * 74)
    for t in [
        "Anything that is not in the running configuration. A RA Guard policy",
        "  applied by a template the device does not print is invisible here.",
        "Whether a filter that exists is actually correct. Parity is a weaker",
        "  claim than equivalence, and this tool only checks parity.",
        "Host firewalls. A device can be perfect and the server behind it still",
        "  listening on :: with an empty ip6tables.",
        "Palo Alto and Fortinet policy: the parity question needs the merged",
        "  effective policy from Panorama or FortiManager, not the device.",
        "Whether IPv6 is reachable at all. This reads configuration, not traffic.",
    ]:
        print("  - %s" % t if not t.startswith("  ") else "   %s" % t)


def write_csv(path, findings):
    with open(path, "w", encoding="utf-8-sig", newline="") as fh:
        w = csv.writer(fh)
        w.writerow(["severity", "device", "check", "where", "detail", "fix"])
        for f in sorted(findings, key=lambda x: (SEV_ORDER[x.severity], x.device)):
            w.writerow([f.severity, f.device, f.check, f.where, f.detail, f.fix])


# ----------------------------------------------------------------- entry
def parse_args(argv):
    p = argparse.ArgumentParser(
        description="Find the IPv6 holes in a device configuration: the ACL that "
                    "only filters IPv4, missing first-hop security, ICMPv6 filtered "
                    "to death, and the /64 on a point-to-point link. Read-only.")
    p.add_argument("--inventory", help="YAML inventory. Needs NET_USER and NET_PASS")
    p.add_argument("--config-dir", help="Audit saved configs instead. No device access. "
                                        "File name decides the platform: name.ios.cfg")
    p.add_argument("--csv", help="Also write every finding to a CSV")
    p.add_argument("--save-dir", help="Save each config pulled, for later offline runs")
    args = p.parse_args(argv)
    if bool(args.inventory) == bool(args.config_dir):
        p.error("pass exactly one of --inventory or --config-dir")
    return args


def main(argv=None):
    args = parse_args(argv)
    findings, seen = [], []

    if args.config_dir:
        if not os.path.isdir(args.config_dir):
            raise SystemExit("--config-dir does not exist: %s" % args.config_dir)
        for fn in sorted(os.listdir(args.config_dir)):
            parts = fn.split(".")
            osn = parts[-2].lower() if len(parts) >= 3 else ""
            if osn not in NETMIKO_DRIVER:
                print("skipping %s: the name must be <device>.<os>.cfg" % fn, file=sys.stderr)
                continue
            name = parts[0]
            with open(os.path.join(args.config_dir, fn), encoding="utf-8", errors="replace") as fh:
                cfg = fh.read()
            seen.append("%s (%s)" % (name, osn))
            findings.extend(audit(name, osn, cfg))
    else:
        for dev in load_inventory(args.inventory):
            try:
                conn = connect(dev)
            except SystemExit:
                raise
            except Exception as exc:
                findings.append(Finding(dev["device_name"], "COULD_NOT_VERIFY",
                                        "unreachable", dev["host"], str(exc)[:160]))
                continue
            try:
                cfg = conn.send_command(CONFIG_CMD[dev["os"]], read_timeout=120)
            finally:
                conn.disconnect()
            seen.append("%s (%s)" % (dev["device_name"], dev["os"]))
            if args.save_dir:
                if not os.path.isdir(args.save_dir):
                    os.makedirs(args.save_dir)
                with open(os.path.join(args.save_dir, "%s.%s.cfg" % (dev["device_name"], dev["os"])),
                          "w", encoding="utf-8") as fh:
                    fh.write(cfg)
            findings.extend(audit(dev["device_name"], dev["os"], cfg))

    report(findings, seen, args)
    if args.csv:
        write_csv(args.csv, findings)
        print("\nCSV written: %s" % args.csv)
    return 1 if any(f.severity in ("HIGH", "MEDIUM", "LOW") for f in findings) else 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        sys.stderr.write("\ninterrupted\n")
        sys.exit(2)
    except SystemExit:
        raise
    except OSError as exc:
        sys.stderr.write("failed: %s\n" % exc)
        sys.exit(2)