VPN Tunnel Monitor

#!/usr/bin/env python3
"""
VPN Tunnel Monitor

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/vpn-tunnel-monitor

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 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
# ---------------------------------------------------------------------------

COMMAND_MAP = {
    "ios": {"session": "show crypto session brief",
            "ike": "show crypto ikev2 sa",
            "ipsec": "show crypto ipsec sa"},
    "iosxe": {"session": "show crypto session brief",
              "ike": "show crypto ikev2 sa",
              "ipsec": "show crypto ipsec sa"},
    "iosxr": {"ike": "show crypto ikev2 sa",
              "ipsec": "show crypto ipsec sa summary"},
    "nxos": {"session": "show crypto session"},
    "junos": {"ike": "show security ike security-associations",
              "ipsec": "show security ipsec security-associations detail"},
    "panos": {"ike": "show vpn ike-sa", "ipsec": "show vpn ipsec-sa"},
    "fortios": {"ike": "get vpn ike gateway", "ipsec": "diagnose vpn tunnel list"},
}

NO_IPSEC = {
    "eos": "Arista EOS does not terminate IPsec VPN tunnels in this form",
}

IPV4 = re.compile(r"\b(\d{1,3}(?:\.\d{1,3}){3})\b")


@dataclass
class Tunnel:
    peer: str
    state: str = "down"              # up | down
    ike_state: str = ""
    sa_lifetime_sec: int | None = None
    encaps: int | None = None        # packets we sent into the tunnel
    decaps: int | None = None        # packets we received from it
    description: str = ""


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

def parse_ios_session(output: str) -> dict[str, Tunnel]:
    """`show crypto session brief` - one line per peer."""
    out: dict[str, Tunnel] = {}
    for line in output.splitlines():
        m = re.match(r"\s*(\d{1,3}(?:\.\d{1,3}){3})\s+\S+\s+(\S+)", line)
        if m:
            peer, status = m.group(1), m.group(2)
            out[peer] = Tunnel(peer=peer,
                               state="up" if status.upper().startswith("UA") or
                                     "UP" in status.upper() else "down",
                               ike_state=status)
            continue
        m = re.search(r"Peer:\s*(\d{1,3}(?:\.\d{1,3}){3}).*?Status:\s*(\S+)",
                      line, re.IGNORECASE)
        if m:
            peer, status = m.group(1), m.group(2)
            out[peer] = Tunnel(peer=peer,
                               state="up" if "up" in status.lower() else "down",
                               ike_state=status)
    return out


def enrich_ios_ipsec(output: str, tunnels: dict[str, Tunnel]) -> None:
    """
    Add SA lifetime and encap/decap counters from `show crypto ipsec sa`.

    The counters matter more than the state. A tunnel that is UP with encaps
    climbing and decaps frozen at zero means we are sending and the peer is
    not replying - the tunnel is up and useless, which a state check calls
    healthy.
    """
    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)
        t = tunnels.setdefault(peer, Tunnel(peer=peer, state="up"))

        life = re.search(r"remaining key lifetime \(k/sec\):\s*\(\d+/(\d+)\)", block)
        if life:
            secs = int(life.group(1))
            if t.sa_lifetime_sec is None or secs < t.sa_lifetime_sec:
                t.sa_lifetime_sec = secs

        enc = re.search(r"#pkts encaps:\s*(\d+)", block)
        dec = re.search(r"#pkts decaps:\s*(\d+)", block)
        if enc:
            t.encaps = (t.encaps or 0) + int(enc.group(1))
        if dec:
            t.decaps = (t.decaps or 0) + int(dec.group(1))


def parse_iosxr_ike(output: str) -> dict[str, Tunnel]:
    out: dict[str, Tunnel] = {}
    for line in output.splitlines():
        m = re.search(r"(\d{1,3}(?:\.\d{1,3}){3}).*?\b(READY|IN-NEG|DELETED|CLOSING|INIT)\b",
                      line, re.IGNORECASE)
        if m:
            peer, st = m.group(1), m.group(2).upper()
            out[peer] = Tunnel(peer=peer, state="up" if st == "READY" else "down",
                               ike_state=st)
    return out


def parse_junos_ike(output: str) -> dict[str, Tunnel]:
    out: dict[str, Tunnel] = {}
    for line in output.splitlines():
        m = re.match(r"\s*\d+\s+(\S+)\s+(\d{1,3}(?:\.\d{1,3}){3})", line)
        if m:
            st, peer = m.group(1), m.group(2)
            out[peer] = Tunnel(peer=peer,
                               state="up" if st.upper() in ("UP", "ESTABLISHED") else "down",
                               ike_state=st)
    return out


def enrich_junos_ipsec(output: str, tunnels: dict[str, Tunnel]) -> None:
    peer = None
    for line in output.splitlines():
        m = re.search(r"Remote gateway:\s*(\d{1,3}(?:\.\d{1,3}){3})", line)
        if m:
            peer = m.group(1)
            tunnels.setdefault(peer, Tunnel(peer=peer, state="up"))
            continue
        if not peer:
            continue
        m = re.search(r"Hard lifetime:\s*Expires in (\d+) seconds", line)
        if m:
            t = tunnels[peer]
            secs = int(m.group(1))
            if t.sa_lifetime_sec is None or secs < t.sa_lifetime_sec:
                t.sa_lifetime_sec = secs


def parse_panos_ike(output: str) -> dict[str, Tunnel]:
    out: dict[str, Tunnel] = {}
    for line in output.splitlines():
        if not IPV4.search(line):
            continue
        m = IPV4.search(line)
        peer = m.group(1)
        up = bool(re.search(r"\b(active|established)\b", line, re.IGNORECASE))
        state_m = re.search(r"\b(active|init|expired|down|established)\b", line, re.IGNORECASE)
        out[peer] = Tunnel(peer=peer, state="up" if up else "down",
                           ike_state=state_m.group(1) if state_m else "")
    return out


def parse_fortios_ike(output: str) -> dict[str, Tunnel]:
    out: dict[str, Tunnel] = {}
    name = ""
    for line in output.splitlines():
        m = re.match(r"name:\s*(\S+)", line.strip())
        if m:
            name = m.group(1)
        m = re.search(r"remote-gateway:\s*(\d{1,3}(?:\.\d{1,3}){3})", line)
        if m:
            peer = m.group(1)
            out[peer] = Tunnel(peer=peer, description=name)
        m = re.search(r"status:\s*(\S+)", line)
        if m and out:
            last = list(out.values())[-1]
            last.ike_state = m.group(1)
            last.state = "up" if "up" in m.group(1).lower() else "down"
    return out


def enrich_fortios_tunnel(output: str, tunnels: dict[str, Tunnel]) -> None:
    peer = None
    for line in output.splitlines():
        m = re.search(r"gw\s+(\d{1,3}(?:\.\d{1,3}){3})", line)
        if m:
            peer = m.group(1)
            tunnels.setdefault(peer, Tunnel(peer=peer, state="up"))
        if not peer:
            continue
        m = re.search(r"encrypted\s+(\d+)\s+decrypted\s+(\d+)", line, re.IGNORECASE)
        if m:
            tunnels[peer].encaps = int(m.group(1))
            tunnels[peer].decaps = int(m.group(2))


def collect(out: dict[str, str], os_key: str) -> dict[str, Tunnel]:
    if os_key in ("ios", "iosxe"):
        t = parse_ios_session(out.get("session", ""))
        if out.get("ipsec"):
            enrich_ios_ipsec(out["ipsec"], t)
        return t
    if os_key == "iosxr":
        return parse_iosxr_ike(out.get("ike", ""))
    if os_key == "nxos":
        return parse_ios_session(out.get("session", ""))
    if os_key == "junos":
        t = parse_junos_ike(out.get("ike", ""))
        if out.get("ipsec"):
            enrich_junos_ipsec(out["ipsec"], t)
        return t
    if os_key == "panos":
        return parse_panos_ike(out.get("ike", ""))
    if os_key == "fortios":
        t = parse_fortios_ike(out.get("ike", ""))
        if out.get("ipsec"):
            enrich_fortios_tunnel(out["ipsec"], t)
        return t
    return {}


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

def check_device(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    r = {"host": host, "os": os_key, "tunnels": [], "issues": [],
         "error": None, "skipped": False}

    if os_key in NO_IPSEC:
        r["skipped"] = True
        r["error"] = NO_IPSEC[os_key]
        return r
    if os_key not in COMMAND_MAP:
        r["skipped"] = True
        r["error"] = f"IPsec VPN monitoring not implemented for {os_key}"
        return r

    conn = None
    try:
        conn = connect(dev)
        out: dict[str, str] = {}
        for key, cmd in COMMAND_MAP[os_key].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 = collect(out, os_key)
        r["tunnels"] = [asdict(t) for t in tunnels.values()]

        for t in tunnels.values():
            if t.state != "up":
                r["issues"].append(
                    f"peer {t.peer} is DOWN (IKE state {t.ike_state or 'unknown'}) - "
                    f"check the peer, the pre-shared key, and that UDP 500/4500 "
                    f"and ESP reach it")
                continue

            # Up but one-way: the single most useful IPsec finding.
            if t.encaps is not None and t.decaps is not None:
                if t.encaps > 0 and t.decaps == 0:
                    r["issues"].append(
                        f"peer {t.peer} is UP but ONE-WAY: {t.encaps:,} packets "
                        f"encrypted, 0 decrypted. We are sending and the peer is "
                        f"not replying - check the far end's interesting-traffic "
                        f"ACL and its return routing")
                elif t.decaps > 0 and t.encaps == 0:
                    r["issues"].append(
                        f"peer {t.peer} is UP but ONE-WAY: {t.decaps:,} packets "
                        f"decrypted, 0 encrypted. The peer is sending and we are "
                        f"not - check our interesting-traffic ACL")

            if (t.sa_lifetime_sec is not None
                    and t.sa_lifetime_sec < args.expiry_warn):
                r["issues"].append(
                    f"peer {t.peer}: SA rekeys in {t.sa_lifetime_sec}s "
                    f"(below the {args.expiry_warn}s threshold)")

        for expected in dev.get("expected_peers", []):
            if expected not in tunnels:
                r["issues"].append(
                    f"expected peer {expected} has no SA at all - the tunnel was "
                    f"never negotiated, not merely down")

        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, os_key = r["host"], r["os"]
    if r["skipped"]:
        return f"[SKIP]  {host:<24} {r['error']}"
    if r["error"]:
        return f"[ERROR] {host:<24} ({os_key}) {r['error']}"
    total = len(r["tunnels"])
    up = sum(1 for t in r["tunnels"] if t["state"] == "up")
    if not r["issues"]:
        return f"[OK]    {host:<24} ({os_key}) {up}/{total} tunnel(s) up"
    return f"[WARN]  {host:<24} ({os_key}) {up}/{total} up - {len(r['issues'])} issue(s)"


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Monitor IPsec/IKE VPN tunnel state across a mixed-vendor estate.",
        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 VPN endpoints to check")
    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("--expiry-warn", type=int, default=3600,
                    help="Warn when an SA rekeys in fewer seconds than this. "
                         "Default: 3600")
    ap.add_argument("--show-tunnels", action="store_true",
                    help="List every tunnel with its counters and SA lifetime")
    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 VPN tunnels on {len(devices)} device(s)...")
        print("=" * 82)

        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": "?", "tunnels": [], "issues": [],
                                    "error": f"worker crashed: {exc}", "skipped": False})

        for r in sorted(results, key=lambda x: x["host"]):
            print(summarise(r))
            if args.show_tunnels and r["tunnels"]:
                for t in sorted(r["tunnels"], key=lambda x: x["peer"]):
                    life = (f"rekey in {t['sa_lifetime_sec']}s"
                            if t["sa_lifetime_sec"] is not None else "lifetime unknown")
                    ctr = ""
                    if t["encaps"] is not None or t["decaps"] is not None:
                        ctr = f"  enc {t['encaps'] or 0:,} / dec {t['decaps'] or 0:,}"
                    name = f"  {t['description']}" if t["description"] else ""
                    print(f"           {t['peer']:<16} {t['state']:<5} {life}{ctr}{name}")
            for i in r["issues"]:
                print(f"           - {i}")

        issues = sum(len(r["issues"]) for r in results)
        errs = sum(1 for r in results if r["error"] and not r["skipped"])
        print("=" * 82)
        if issues or errs:
            print(f"{issues} tunnel issue(s), {errs} device(s) unreachable.")
        else:
            total = sum(len(r["tunnels"]) for r in results)
            print(f"All {total} tunnel(s) healthy across {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": "*VPN tunnel 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)