NHRP Cache Inspector

#!/usr/bin/env python3
"""
NHRP Cache Inspector

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/nhrp-cache-inspector

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" "requests>=2.28"
"""

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
import time

try:
    import requests
except ImportError:
    requests = None

# ---------------------------------------------------------------------------
# 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
# ---------------------------------------------------------------------------
# NHRP is the DMVPN control plane and exists only on Cisco IOS / IOS-XE.
# Devices with any other OS are reported as skipped rather than guessed at.

HUB_COMMANDS = {
    "nhrp": "show ip nhrp",
    "nhrp_summary": "show ip nhrp summary",
    "nhrp_traffic": "show ip nhrp traffic",
    "nhrp_nhs": "show ip nhrp nhs",
    "dmvpn": "show dmvpn",
}

SPOKE_COMMANDS = {
    "nhrp": "show ip nhrp",
    "nhrp_nhs": "show ip nhrp nhs detail",
    "nhrp_traffic": "show ip nhrp traffic",
    "dmvpn": "show dmvpn",
}

SUPPORTED = {"ios", "iosxe"}


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

def parse_nhrp_cache(output: str) -> list[dict]:
    """
    Parse `show ip nhrp`.

    Entries span two lines: the mapping line carries the tunnel address, and
    the following line carries the NBMA address and flags. Parsing only the
    first line misses the NBMA address entirely, which is the field you need
    to tell a registration apart from a resolved shortcut.
    """
    entries: list[dict] = []
    cur: dict | None = None

    for line in output.splitlines():
        m = re.match(r"\s*(\d{1,3}(?:\.\d{1,3}){3})(?:/(\d+))?\s+via\s+"
                     r"(\d{1,3}(?:\.\d{1,3}){3})", line)
        if m:
            if cur:
                entries.append(cur)
            cur = {"tunnel_ip": m.group(1), "prefix_len": m.group(2) or "32",
                   "via": m.group(3), "nbma_ip": "", "interface": "",
                   "flags": [], "expires": "never", "registered": False,
                   "incomplete": False, "shortcut": False, "local": False}
            exp = re.search(r"expire\s+([\d:]+)", line, re.IGNORECASE)
            if exp:
                cur["expires"] = exp.group(1)
            iface = re.search(r"\b(Tunnel\d+)\b", line)
            if iface:
                cur["interface"] = iface.group(1)
            fl = re.search(r"<([^>]*)>", line)
            if fl:
                _apply_flags(cur, fl.group(1))
            continue

        if cur is None:
            continue

        nbma = re.search(r"NBMA address:\s*(\d{1,3}(?:\.\d{1,3}){3})", line, re.IGNORECASE)
        if nbma:
            cur["nbma_ip"] = nbma.group(1)

        # Two formats in the wild: older IOS puts flags in <angle brackets> on
        # the mapping line; newer IOS and IOS-XE put them on a following
        # "Type: dynamic, Flags: unique registered nhop" line. Both are read.
        fl = re.search(r"<([^>]*)>", line)
        if fl:
            _apply_flags(cur, fl.group(1))
        fl = re.search(r"Flags:\s*(.+)", line, re.IGNORECASE)
        if fl:
            _apply_flags(cur, fl.group(1))
        ty = re.search(r"Type:\s*(\w+)", line, re.IGNORECASE)
        if ty:
            _apply_flags(cur, ty.group(1))

        iface = re.search(r"\b(Tunnel\d+)\b", line)
        if iface and not cur["interface"]:
            cur["interface"] = iface.group(1)

    if cur:
        entries.append(cur)

    for e in entries:
        if not e["nbma_ip"]:
            e["nbma_ip"] = e["via"]
    return entries


def _apply_flags(entry: dict, flags_str: str) -> None:
    # Flags arrive comma-separated from "Flags:" lines and space-separated
    # from <angle brackets>, so split on both.
    flags = [f.strip().lower() for f in re.split(r"[,\s]+", flags_str) if f.strip()]
    entry["flags"] = sorted(set(entry["flags"] + flags))
    entry["registered"] = entry["registered"] or "registered" in flags
    entry["incomplete"] = entry["incomplete"] or "incomplete" in flags
    entry["shortcut"] = entry["shortcut"] or any(
        f in flags for f in ("router", "rib", "nho", "nhop"))
    entry["local"] = entry["local"] or "local" in flags


def parse_nhrp_summary(output: str) -> dict:
    s = {"total": 0, "static": 0, "dynamic": 0, "incomplete": 0, "negative": 0}
    for line in output.splitlines():
        for key, pat in (("total", r"Total\s+NHRP\s+Entries:\s*(\d+)"),
                         ("total", r"Total\s+Entries:\s*(\d+)"),
                         ("static", r"Static:\s*(\d+)"),
                         ("dynamic", r"Dynamic:\s*(\d+)"),
                         ("incomplete", r"Incomplete:\s*(\d+)"),
                         ("negative", r"Negative:\s*(\d+)")):
            m = re.search(pat, line, re.IGNORECASE)
            if m:
                s[key] = int(m.group(1))
    return s


def parse_nhrp_nhs(output: str) -> list[dict]:
    """Parse `show ip nhrp nhs` - the spoke's view of each hub."""
    out: list[dict] = []
    for line in output.splitlines():
        m = re.search(r"(\d{1,3}(?:\.\d{1,3}){3})\s+(?:RE|E|\w+)?\s*"
                      r"priority\s+(\d+).*?cluster\s+(\d+)", line, re.IGNORECASE)
        if m:
            state = "up" if re.search(r"\bRE\b", line) else "down"
            out.append({"nhs_ip": m.group(1), "state": state,
                        "priority": int(m.group(2)), "cluster": int(m.group(3)),
                        "req_sent": 0, "req_failed": 0,
                        "healthy": state == "up"})
            continue
        m = re.search(r"NHS\s+(\S+).*?state:?\s*(\S+)", line, re.IGNORECASE)
        if m:
            state = m.group(2).strip().lower().rstrip(",")
            sent = re.search(r"req-sent\s+(\d+)", line, re.IGNORECASE)
            failed = re.search(r"req-failed\s+(\d+)", line, re.IGNORECASE)
            out.append({"nhs_ip": m.group(1), "state": state, "priority": 0,
                        "cluster": 0,
                        "req_sent": int(sent.group(1)) if sent else 0,
                        "req_failed": int(failed.group(1)) if failed else 0,
                        "healthy": state in ("up", "re", "reachable", "e")})
    return out


def parse_nhrp_traffic(output: str) -> dict:
    stats = {"req_sent": 0, "req_rcvd": 0, "reply_sent": 0, "reply_rcvd": 0,
             "reg_sent": 0, "reg_rcvd": 0, "errors": 0, "redirects": 0}
    for line in output.splitlines():
        low = line.lower()
        m = re.search(r"sent\s+(\d+)\s+received\s+(\d+)", low)
        if m:
            s, r = int(m.group(1)), int(m.group(2))
            if "resolution request" in low:
                stats["req_sent"], stats["req_rcvd"] = s, r
            elif "resolution reply" in low:
                stats["reply_sent"], stats["reply_rcvd"] = s, r
            elif "registration request" in low:
                stats["reg_sent"], stats["reg_rcvd"] = s, r
        m = re.search(r"error\w*\s+(\d+)", low)
        if m:
            stats["errors"] += int(m.group(1))
        m = re.search(r"redirect\w*\s+sent\s+(\d+)", low)
        if m:
            stats["redirects"] += int(m.group(1))
    return stats


def parse_dmvpn(output: str) -> list[dict]:
    """`show dmvpn` gives the tunnel state the NHRP cache alone does not."""
    peers: list[dict] = []
    for line in output.splitlines():
        m = re.match(r"\s*\d+\s+(\d{1,3}(?:\.\d{1,3}){3})\s+"
                     r"(\d{1,3}(?:\.\d{1,3}){3})\s+(\S+)\s+(\S+)\s+(\S+)", line)
        if m:
            peers.append({"nbma": m.group(1), "tunnel": m.group(2),
                          "state": m.group(3), "uptime": m.group(4),
                          "attrb": m.group(5)})
    return peers


# ---------------------------------------------------------------------------
# Detail printer  (this function was referenced but never defined in the
# original script, so it crashed with NameError on the first result)
# ---------------------------------------------------------------------------

def print_nhrp_detail(r: dict, args) -> None:
    if not args.show_detail or r.get("error") or r.get("skipped"):
        return

    entries = r["entries"]
    if entries:
        print("           NHRP cache:")
        for e in sorted(entries, key=lambda x: x["tunnel_ip"]):
            kind = ("local" if e["local"] else
                    "shortcut" if e["shortcut"] else
                    "registered" if e["registered"] else
                    "INCOMPLETE" if e["incomplete"] else "dynamic")
            print(f"             {e['tunnel_ip']:<16} via NBMA {e['nbma_ip']:<16} "
                  f"{e['interface']:<9} {kind:<11} expires {e['expires']}")

    if r["nhs"]:
        print("           NHS (hubs this spoke registers with):")
        for n in r["nhs"]:
            mark = "up" if n["healthy"] else "DOWN"
            extra = (f"  req-sent {n['req_sent']} failed {n['req_failed']}"
                     if n["req_sent"] or n["req_failed"] else "")
            print(f"             {n['nhs_ip']:<16} {mark:<5} "
                  f"priority {n['priority']} cluster {n['cluster']}{extra}")

    t = r["traffic"]
    if any(t.values()):
        print(f"           traffic: resolution req {t['req_sent']}/{t['req_rcvd']} "
              f"reply {t['reply_sent']}/{t['reply_rcvd']} "
              f"registration {t['reg_sent']}/{t['reg_rcvd']} "
              f"errors {t['errors']} redirects {t['redirects']}")

    if r["tunnels"]:
        up = sum(1 for p in r["tunnels"] if p["state"].upper().startswith("UP"))
        print(f"           dmvpn: {up}/{len(r['tunnels'])} tunnel(s) UP")


# ---------------------------------------------------------------------------
# Per-device inspection
# ---------------------------------------------------------------------------

def inspect_device(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    role = str(dev.get("role", "spoke")).lower()
    r = {"host": host, "os": os_key, "role": role, "issues": [], "entries": [],
         "nhs": [], "traffic": {}, "summary": {}, "tunnels": [],
         "error": None, "skipped": False}

    if os_key not in SUPPORTED:
        r["skipped"] = True
        r["error"] = f"NHRP/DMVPN is a Cisco IOS feature; nothing to inspect on {os_key}"
        return r
    if role not in ("hub", "spoke"):
        r["error"] = f"role must be 'hub' or 'spoke', got '{role}'"
        return r

    cmds = HUB_COMMANDS if role == "hub" else SPOKE_COMMANDS
    conn = None
    try:
        conn = connect(dev)
        out: dict[str, str] = {}
        for key, cmd in cmds.items():
            try:
                out[key] = conn.send_command(cmd, read_timeout=args.read_timeout)
            except Exception:  # noqa: BLE001 - a missing command must not lose the rest
                out[key] = ""

        r["entries"] = parse_nhrp_cache(out.get("nhrp", ""))
        r["nhs"] = parse_nhrp_nhs(out.get("nhrp_nhs", ""))
        r["traffic"] = parse_nhrp_traffic(out.get("nhrp_traffic", ""))
        r["tunnels"] = parse_dmvpn(out.get("dmvpn", ""))
        if role == "hub":
            r["summary"] = parse_nhrp_summary(out.get("nhrp_summary", ""))

        for e in r["entries"]:
            if e["incomplete"]:
                r["issues"].append(
                    f"NHRP INCOMPLETE for {e['tunnel_ip']} - resolution was requested "
                    f"and never answered. The spoke is unreachable over the underlay, "
                    f"or NAT is breaking the NBMA address")

        if role == "hub":
            expected = dev.get("expected_spokes", [])
            registered = {e["tunnel_ip"] for e in r["entries"] if e["registered"]}
            for sp in expected:
                if sp not in registered:
                    r["issues"].append(
                        f"spoke {sp} has not registered - check its 'ip nhrp nhs', "
                        f"'ip nhrp map' and underlay reachability to this hub")
            if expected and len(registered) < len(expected):
                r["issues"].append(
                    f"{len(registered)}/{len(expected)} expected spoke(s) registered")

        if role == "spoke":
            if not r["nhs"]:
                r["issues"].append(
                    "no NHS configured or none responding - the spoke has no "
                    "'ip nhrp nhs' statement, or every hub is unreachable")
            for n in r["nhs"]:
                if not n["healthy"]:
                    r["issues"].append(
                        f"NHS {n['nhs_ip']} is {n['state'].upper()} - the hub is not "
                        f"answering registration. Check underlay routing and that "
                        f"the hub tunnel interface is up")
            for exp in dev.get("expected_nhs", []):
                if exp not in {n["nhs_ip"] for n in r["nhs"]}:
                    r["issues"].append(
                        f"expected NHS {exp} is not configured - add "
                        f"'ip nhrp nhs {exp} nbma <hub-wan-ip> multicast'")

        t = r["traffic"]
        if t.get("errors", 0) > args.error_threshold:
            r["issues"].append(
                f"NHRP error counter is {t['errors']} - usually a network-id "
                f"mismatch or an authentication key mismatch between hub and spoke")
        if role == "spoke" and t.get("req_sent", 0) > 0 and t.get("reply_rcvd", 0) == 0:
            r["issues"].append(
                f"{t['req_sent']} resolution request(s) sent, zero replies received - "
                f"spoke-to-spoke tunnels cannot form. Check 'ip nhrp redirect' on the "
                f"hub and 'ip nhrp shortcut' on the spokes")

        return r

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


def summarise(r: dict) -> str:
    host, role = r["host"], r["role"].upper()
    if r["skipped"]:
        return f"[SKIP]  {host:<22} {r['error']}"
    if r["error"]:
        return f"[ERROR] {host:<22} ({role}) {r['error']}"
    n = len(r["entries"])
    reg = sum(1 for e in r["entries"] if e["registered"])
    sc = sum(1 for e in r["entries"] if e["shortcut"])
    inc = sum(1 for e in r["entries"] if e["incomplete"])
    detail = f"{n} entries, {reg} registered"
    if sc:
        detail += f", {sc} shortcut"
    if inc:
        detail += f", {inc} INCOMPLETE"
    if not r["issues"]:
        return f"[OK]    {host:<22} ({role}) {detail}"
    return f"[WARN]  {host:<22} ({role}) {detail} - {len(r['issues'])} issue(s)"


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Inspect the NHRP cache on DMVPN hubs and spokes.",
        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 the DMVPN hubs and spokes")
    ap.add_argument("--daemon", action="store_true",
                    help="Run continuously instead of inspecting once")
    ap.add_argument("--interval", type=int, default=120,
                    help="Seconds between runs in daemon mode. Default: 120")
    ap.add_argument("--show-detail", action="store_true",
                    help="Print the full NHRP cache, NHS table and traffic counters")
    ap.add_argument("--error-threshold", type=int, default=10,
                    help="NHRP error count that triggers an alert. Default: 10")
    ap.add_argument("--limit", help="Only inspect hosts whose name contains this text")
    ap.add_argument("--webhook", default="", help="Slack or Teams webhook URL")
    ap.add_argument("--read-timeout", type=int, default=60,
                    help="Seconds to wait per command. Default: 60")
    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 results 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

    def run_once() -> int:
        print(f"\nInspecting NHRP cache on {len(devices)} device(s)...")
        print("=" * 82)

        results = []
        with ThreadPoolExecutor(max_workers=args.workers) as pool:
            futures = {pool.submit(inspect_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": "?", "role": "?",
                                    "issues": [], "entries": [], "nhs": [],
                                    "traffic": {}, "summary": {}, "tunnels": [],
                                    "error": f"worker crashed: {exc}", "skipped": False})

        hubs = [r for r in results if r["role"] == "hub"]
        spokes = [r for r in results if r["role"] == "spoke"]
        for r in hubs + spokes:
            print(summarise(r))
            print_nhrp_detail(r, args)
            for i in r["issues"]:
                print(f"           - {i}")

        # Cross-check the hub's view against the spokes that actually answered.
        live_hubs = [r for r in hubs if not r["error"] and not r["skipped"]]
        live_spokes = [r for r in spokes if not r["error"] and not r["skipped"]]
        cross: list[str] = []
        if live_hubs and live_spokes:
            registered = {e["tunnel_ip"] for h in live_hubs for e in h["entries"]
                          if e["registered"]}
            for s in live_spokes:
                own = {e["tunnel_ip"] for e in s["entries"] if e["local"]}
                if own and not (own & registered):
                    cross.append(f"{s['host']}: its tunnel address does not appear "
                                 f"as registered in any hub's NHRP cache")

        print("=" * 82)
        if cross:
            print(f"HUB/SPOKE MISMATCH: {len(cross)} issue(s)")
            for c in cross:
                print(f"  ! {c}")

        issues = sum(len(r["issues"]) for r in results) + len(cross)
        errs = sum(1 for r in results if r["error"] and not r["skipped"])
        if issues or errs:
            print(f"{issues} issue(s) across {len(results)} device(s), {errs} unreachable.")
        else:
            print(f"NHRP healthy on all {len(results)} device(s).")

        if (issues or errs) and args.webhook and requests is not None:
            lines = [summarise(r) for r in results if r["issues"] or r["error"]]
            try:
                requests.post(args.webhook,
                              json={"text": "*NHRP cache alert*\n" + "\n".join(lines)},
                              timeout=10)
            except Exception as exc:  # noqa: BLE001
                print(f"[WARN] webhook failed: {exc}", file=sys.stderr)

        if args.json_out:
            Path(args.json_out).write_text(json.dumps(results, indent=2), encoding="utf-8")

        return 1 if (issues or errs) else 0

    if not args.daemon:
        return run_once()

    while True:
        run_once()
        print(f"\nNext inspection in {args.interval}s. Ctrl-C to stop.")
        time.sleep(args.interval)


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