DMVPN Health Check

#!/usr/bin/env python3
"""
DMVPN Health Check

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/dmvpn-health-check

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
from collections import defaultdict
from dataclasses import dataclass, asdict

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
# ---------------------------------------------------------------------------
# DMVPN is a Cisco IOS / IOS-XE technology. Any other OS is reported as
# skipped rather than guessed at.

COMMANDS = {
    "dmvpn": "show dmvpn",
    "tunnel": "show ip interface brief | include Tunnel",
    "nhrp_summary": "show ip nhrp summary",
    "crypto": "show crypto session brief",
    "ipsec": "show crypto ipsec sa",
}

SUPPORTED = {"ios", "iosxe"}

# DMVPN fails in three independent layers. Reporting WHICH layer broke is the
# whole point: the fix for a GRE problem has nothing in common with the fix
# for an NHRP or an IPsec problem.
LAYER_GRE = "GRE"       # tunnel interface and underlay reachability
LAYER_NHRP = "NHRP"     # spoke registration and resolution
LAYER_IPSEC = "IPSEC"   # the crypto session protecting the tunnel


@dataclass
class Peer:
    tunnel_ip: str
    nbma_ip: str
    state: str            # UP | NHRP | IKE | down
    uptime: str = ""
    attrb: str = ""


@dataclass
class Finding:
    layer: str
    message: str
    fix: str = ""


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

def parse_dmvpn(output: str) -> dict[str, list[Peer]]:
    """
    Parse `show dmvpn`.

    The peer line is:  # Ent  Peer-NBMA  Peer-Tunnel  State  UpDn-Time  Attrb
    State is the DMVPN session state and carries the diagnosis directly:
      UP    tunnel is fully established
      NHRP  NHRP resolution is in progress or stuck
      IKE   IKE negotiation is in progress or stuck
    """
    out: dict[str, list[Peer]] = defaultdict(list)
    tunnel = None

    for line in output.splitlines():
        m = re.search(r"Interface:\s*(Tunnel\d+)", line, re.IGNORECASE)
        if m:
            tunnel = m.group(1)
            out.setdefault(tunnel, [])
            continue
        m = re.match(r"^(Tunnel\d+)\b", line)
        if m:
            tunnel = m.group(1)
            out.setdefault(tunnel, [])
            continue
        if tunnel is None:
            continue

        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:
            out[tunnel].append(Peer(nbma_ip=m.group(1), tunnel_ip=m.group(2),
                                    state=m.group(3).upper(), uptime=m.group(4),
                                    attrb=m.group(5)))
    return dict(out)


def parse_tunnel_state(output: str) -> dict[str, str]:
    """`show ip interface brief | include Tunnel` -> {Tunnel0: 'up'|'down'}."""
    out: dict[str, str] = {}
    for line in output.splitlines():
        m = re.match(r"(Tunnel\d+)\s+\S+\s+\S+\s+\S+\s+(\S+)\s+(\S+)", line)
        if m:
            admin, proto = m.group(2).lower(), m.group(3).lower()
            out[m.group(1)] = "up" if admin == "up" and proto == "up" else "down"
    return out


def parse_nhrp_summary(output: str) -> dict[str, int]:
    s = {"total": 0, "incomplete": 0, "negative": 0}
    for line in output.splitlines():
        for key, pat in (("total", r"Total\s+(?:NHRP\s+)?Entries:\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_crypto(output: str) -> dict[str, str]:
    out: dict[str, str] = {}
    for line in output.splitlines():
        m = re.search(r"Peer:\s*(\d{1,3}(?:\.\d{1,3}){3}).*?Status:\s*([\w-]+)",
                      line, re.IGNORECASE)
        if m:
            out[m.group(1)] = m.group(2).upper()
            continue
        m = re.match(r"\s*(\d{1,3}(?:\.\d{1,3}){3})\s+\S+\s+([\w-]+)", line)
        if m:
            out[m.group(1)] = m.group(2).upper()
    return out


def parse_ipsec_counters(output: str) -> dict[str, tuple[int, int]]:
    """{peer: (encaps, decaps)} - the one-way check."""
    out: dict[str, tuple[int, int]] = {}
    for block in re.split(r"(?=local\s+ident)", output)[1:]:
        m = re.search(r"current_peer\s+(\d{1,3}(?:\.\d{1,3}){3})", block)
        if not m:
            continue
        peer = m.group(1)
        enc = re.search(r"#pkts encaps:\s*(\d+)", block)
        dec = re.search(r"#pkts decaps:\s*(\d+)", block)
        e, d = out.get(peer, (0, 0))
        out[peer] = (e + (int(enc.group(1)) if enc else 0),
                     d + (int(dec.group(1)) if dec else 0))
    return out


# ---------------------------------------------------------------------------
# Per-device check
# ---------------------------------------------------------------------------

def check_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, "findings": [], "tunnels": {},
         "peers": {}, "nhrp": {}, "error": None, "skipped": False}

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

    conn = None
    try:
        conn = connect(dev)
        out: dict[str, str] = {}
        for key, cmd in COMMANDS.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] = ""

        tunnels = parse_tunnel_state(out.get("tunnel", ""))
        peers = parse_dmvpn(out.get("dmvpn", ""))
        nhrp = parse_nhrp_summary(out.get("nhrp_summary", ""))
        crypto = parse_crypto(out.get("crypto", ""))
        counters = parse_ipsec_counters(out.get("ipsec", ""))

        r["tunnels"] = tunnels
        r["peers"] = {t: [asdict(p) for p in ps] for t, ps in peers.items()}
        r["nhrp"] = nhrp
        f: list[Finding] = []

        # --- Layer 1: GRE -------------------------------------------------
        for iface, state in tunnels.items():
            if state != "up":
                f.append(Finding(LAYER_GRE,
                                 f"{iface} is {state.upper()}",
                                 "check the tunnel source interface, the underlay "
                                 "route to the destination, and 'no shutdown'"))

        # --- Layer 2: NHRP ------------------------------------------------
        if nhrp.get("incomplete", 0) > 0:
            f.append(Finding(LAYER_NHRP,
                             f"{nhrp['incomplete']} incomplete NHRP entry(ies)",
                             "resolution was requested and never answered - the far "
                             "spoke is unreachable over the underlay, or NAT is "
                             "rewriting the NBMA address"))

        all_peers = [p for ps in peers.values() for p in ps]
        for p in all_peers:
            if p.state == "UP":
                continue
            if p.state == "NHRP":
                f.append(Finding(LAYER_NHRP,
                                 f"peer {p.tunnel_ip} (NBMA {p.nbma_ip}) is stuck in NHRP",
                                 "registration or resolution is not completing - check "
                                 "'ip nhrp network-id' and 'ip nhrp authentication' "
                                 "match on both ends"))
            elif p.state == "IKE":
                f.append(Finding(LAYER_IPSEC,
                                 f"peer {p.tunnel_ip} (NBMA {p.nbma_ip}) is stuck in IKE",
                                 "the tunnel is up at GRE level but IPsec will not "
                                 "negotiate - check the pre-shared key, the IKE "
                                 "proposal, and that UDP 500/4500 and ESP reach the peer"))
            else:
                f.append(Finding(LAYER_NHRP,
                                 f"peer {p.tunnel_ip} (NBMA {p.nbma_ip}) is {p.state}",
                                 "session is not established"))

        if role == "hub":
            expected = dev.get("expected_spoke_count")
            up = sum(1 for p in all_peers if p.state == "UP")
            if expected is not None and up < int(expected):
                f.append(Finding(LAYER_NHRP,
                                 f"{up}/{expected} expected spoke(s) are UP",
                                 "a spoke that never registered leaves no entry at "
                                 "all - check its 'ip nhrp nhs' and underlay path "
                                 "to this hub"))
        else:
            if not all_peers:
                f.append(Finding(LAYER_NHRP,
                                 "no DMVPN peer at all - the spoke has not registered",
                                 "check 'ip nhrp nhs <hub-tunnel-ip>', "
                                 "'ip nhrp map <hub-tunnel-ip> <hub-wan-ip>', and "
                                 "underlay reachability to the hub"))

        # --- Layer 3: IPsec -----------------------------------------------
        for peer_ip, status in crypto.items():
            if "DOWN" in status or "IDLE" in status:
                f.append(Finding(LAYER_IPSEC,
                                 f"crypto session to {peer_ip} is {status}",
                                 "check the IKE policy, the pre-shared key, and the "
                                 "crypto ACL"))

        for peer_ip, (enc, dec) in counters.items():
            if enc > 0 and dec == 0:
                f.append(Finding(LAYER_IPSEC,
                                 f"peer {peer_ip} is encrypting {enc:,} packets and "
                                 f"decrypting 0 - the tunnel is one-way",
                                 "we are sending and the peer is not replying - check "
                                 "the far end's crypto ACL and its return route"))
            elif dec > 0 and enc == 0:
                f.append(Finding(LAYER_IPSEC,
                                 f"peer {peer_ip} is decrypting {dec:,} packets and "
                                 f"encrypting 0 - the tunnel is one-way",
                                 "the peer is sending and we are not - check our "
                                 "crypto ACL and route to the remote subnet"))

        r["findings"] = [asdict(x) for x in f]
        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']}"

    peers = [p for ps in r["peers"].values() for p in ps]
    up = sum(1 for p in peers if p["state"] == "UP")
    if not r["findings"]:
        return f"[OK]    {host:<22} ({role}) {up}/{len(peers)} peer(s) UP, all layers healthy"

    layers = sorted({f["layer"] for f in r["findings"]})
    return (f"[WARN]  {host:<22} ({role}) {up}/{len(peers)} peer(s) UP - "
            f"{len(r['findings'])} issue(s) in {'/'.join(layers)}")


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Check DMVPN health across its three layers: GRE, NHRP and IPsec.",
        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 checking once")
    ap.add_argument("--interval", type=int, default=300,
                    help="Seconds between checks in daemon mode. Default: 300")
    ap.add_argument("--layer", choices=["GRE", "NHRP", "IPSEC"],
                    help="Only report findings in this layer")
    ap.add_argument("--show-peers", action="store_true",
                    help="List every DMVPN peer with its state and uptime")
    ap.add_argument("--limit", help="Only check 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"\nChecking DMVPN health on {len(devices)} device(s)...")
        print("=" * 84)

        results = []
        with ThreadPoolExecutor(max_workers=args.workers) as pool:
            futures = {pool.submit(check_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": "?",
                                    "findings": [], "tunnels": {}, "peers": {},
                                    "nhrp": {}, "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))
            if args.show_peers:
                for tun, ps in r["peers"].items():
                    for p in sorted(ps, key=lambda x: x["tunnel_ip"]):
                        print(f"           {tun} {p['tunnel_ip']:<16} "
                              f"NBMA {p['nbma_ip']:<16} {p['state']:<5} {p['uptime']}")
            for f in r["findings"]:
                if args.layer and f["layer"] != args.layer:
                    continue
                print(f"           [{f['layer']:<5}] {f['message']}")
                if f["fix"]:
                    print(f"                   -> {f['fix']}")

        # Which layer is actually failing, across the whole DMVPN
        by_layer = defaultdict(int)
        for r in results:
            for f in r["findings"]:
                by_layer[f["layer"]] += 1

        print("=" * 84)
        if by_layer:
            parts = "  ".join(f"{k}={v}" for k, v in sorted(by_layer.items()))
            print(f"Findings by layer:  {parts}")
            if by_layer.get(LAYER_GRE):
                print("GRE findings come first - NHRP and IPsec cannot work over a "
                      "tunnel that is down. Fix the GRE layer before the rest.")
        else:
            print(f"DMVPN healthy across all {len(results)} device(s).")

        issues = sum(len(r["findings"]) for r in results)
        errs = sum(1 for r in results if r["error"] and not r["skipped"])
        if errs:
            print(f"{errs} device(s) unreachable.")

        if (issues or errs) and args.webhook and requests is not None:
            lines = [summarise(r) for r in results if r["findings"] or r["error"]]
            try:
                requests.post(args.webhook,
                              json={"text": "*DMVPN health 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 check 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)