VXLAN VTEP & EVPN Fabric Monitor

#!/usr/bin/env python3
"""
VXLAN VTEP & EVPN Fabric Monitor

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/vxlan-vtep-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"
"""

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, field, asdict

# ---------------------------------------------------------------------------
# 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
# ---------------------------------------------------------------------------
# VXLAN EVPN fabrics run on NX-OS, IOS-XE, Arista EOS and Junos. IOS-XR,
# PAN-OS and FortiOS do not build VXLAN EVPN fabrics, so devices with those
# OS values are reported as skipped rather than guessed at.

FABRIC_CMDS = {
    "nxos": {
        "peers": "show nve peers",
        "vni": "show nve vni",
        "bgp": "show bgp l2vpn evpn summary",
        "nve": "show nve interface nve1 detail",
    },
    "iosxe": {
        "peers": "show l2vpn evpn peers vxlan",
        "vni": "show l2vpn evpn evi",
        "bgp": "show bgp l2vpn evpn summary",
        "nve": "",
    },
    "eos": {
        "peers": "show vxlan vtep",
        "vni": "show vxlan vni",
        "bgp": "show bgp evpn summary",
        "nve": "show interfaces vxlan 1",
    },
    "junos": {
        "peers": "show ethernet-switching vxlan-tunnel-end-point remote",
        "vni": "show evpn instance",
        "bgp": "show bgp summary",
        "nve": "",
    },
}

NOT_FABRIC = {
    "ios": "classic IOS does not build VXLAN EVPN fabrics",
    "iosxr": "IOS-XR uses EVPN but not the NVE/VTEP model this script checks",
    "panos": "PAN-OS is a firewall, not a VXLAN fabric node",
    "fortios": "FortiOS is a firewall, not a VXLAN fabric node",
}

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


@dataclass
class Fabric:
    host: str
    os: str
    peers: dict[str, str] = field(default_factory=dict)      # peer_ip -> state
    vnis: dict[str, str] = field(default_factory=dict)       # vni -> state
    bgp: dict[str, str] = field(default_factory=dict)        # neighbor -> state
    source_ip: str = ""
    issues: list[str] = field(default_factory=list)
    status: str = "ok"
    error: str = ""


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

def parse_peers(output: str, os_key: str) -> dict[str, str]:
    peers: dict[str, str] = {}

    if os_key == "nxos":
        # nve1  10.1.0.2  Up  CP  1d02h  5001.0000.1b08
        for line in output.splitlines():
            m = re.match(r"nve\d+\s+(\d+\.\d+\.\d+\.\d+)\s+(\S+)", line.strip())
            if m:
                peers[m.group(1)] = m.group(2)
        return peers

    if os_key == "eos":
        # Remote VTEPS for Vxlan1: 10.1.0.2, 10.1.0.3  (or a per-line table)
        for line in output.splitlines():
            for ip in re.findall(r"\b(\d{1,3}(?:\.\d{1,3}){3})\b", line):
                state = "Up"
                if re.search(r"\b(down|unreachable)\b", line, re.IGNORECASE):
                    state = "Down"
                peers[ip] = state
        return peers

    if os_key == "junos":
        for line in output.splitlines():
            m = re.search(r"\b(\d{1,3}(?:\.\d{1,3}){3})\b.*?\b(RNVE|up|down)\b",
                          line, re.IGNORECASE)
            if m:
                peers[m.group(1)] = "Up" if m.group(2).lower() != "down" else "Down"
            elif IPV4.match(line.strip().split()[0] if line.strip() else ""):
                peers[line.split()[0]] = "Up"
        return peers

    # iosxe
    for line in output.splitlines():
        parts = line.split()
        if parts and IPV4.match(parts[0]):
            up = any(p.lower().startswith(("up", "estab")) for p in parts[1:])
            peers[parts[0]] = "Up" if up else (parts[1] if len(parts) > 1 else "Unknown")
    return peers


def parse_vnis(output: str, os_key: str) -> dict[str, str]:
    vnis: dict[str, str] = {}

    if os_key == "nxos":
        # nve1  10010  UnicastBGP  Up  CP  L2 [10]
        for line in output.splitlines():
            m = re.match(r"nve\d+\s+(\d+)\s+\S+\s+(\S+)", line.strip())
            if m:
                vnis[m.group(1)] = m.group(2)
        return vnis

    if os_key == "eos":
        for line in output.splitlines():
            m = re.search(r"\bVNI\s+(\d+)\b.*?\b(\w+)\s*$", line) or \
                re.match(r"\s*(\d{4,})\s+\S+\s+(\S+)", line)
            if m:
                vnis[m.group(1)] = m.group(2)
        return vnis

    if os_key == "junos":
        for line in output.splitlines():
            m = re.search(r"VNI[:\s]+(\d+)", line)
            if m:
                vnis[m.group(1)] = "Up"
        return vnis

    # iosxe: EVI  VLAN  state
    for line in output.splitlines():
        m = re.match(r"\s*(\d+)\s+(\d+)\s+(\S+)", line)
        if m:
            vnis[m.group(1)] = m.group(3)
    return vnis


def parse_bgp_evpn(output: str) -> dict[str, str]:
    """BGP EVPN summary shares the standard layout: a count means Established."""
    sessions: dict[str, str] = {}
    for line in output.splitlines():
        parts = line.split()
        if len(parts) < 8 or not IPV4.match(parts[0]):
            continue
        sessions[parts[0]] = "Established" if parts[-1].isdigit() else parts[-1]
    return sessions


def parse_source_ip(output: str, os_key: str) -> str:
    """
    The VTEP source address other leaves peer with.

    On NX-OS with vPC the NVE loopback carries a primary and a secondary; the
    SECONDARY is the anycast VTEP address remote leaves see. Reporting the
    primary here would make a correct fabric look broken.
    """
    if os_key == "nxos":
        m = re.search(r"secondary:\s*(\d{1,3}(?:\.\d{1,3}){3})", output, re.IGNORECASE)
        if m:
            return m.group(1)
        m = re.search(r"primary:\s*(\d{1,3}(?:\.\d{1,3}){3})", output, re.IGNORECASE)
        return m.group(1) if m else ""
    if os_key == "eos":
        m = re.search(r"Source interface is \S+ and is active with "
                      r"(\d{1,3}(?:\.\d{1,3}){3})", output)
        return m.group(1) if m else ""
    return ""


# ---------------------------------------------------------------------------
# Per-device collection
# ---------------------------------------------------------------------------

def check_device(dev: dict, args) -> Fabric:
    host = dev["host"]
    os_key = dev["os"]
    f = Fabric(host=host, os=os_key)
    conn = None

    if os_key not in FABRIC_CMDS:
        f.status = "skip"
        f.error = NOT_FABRIC.get(os_key, f"VXLAN EVPN not supported on {os_key}")
        return f

    cmds = FABRIC_CMDS[os_key]
    try:
        conn = connect(dev)

        f.peers = parse_peers(conn.send_command(cmds["peers"], read_timeout=args.read_timeout),
                              os_key)
        f.vnis = parse_vnis(conn.send_command(cmds["vni"], read_timeout=args.read_timeout),
                            os_key)
        f.bgp = parse_bgp_evpn(conn.send_command(cmds["bgp"], read_timeout=args.read_timeout))
        if cmds["nve"]:
            try:
                f.source_ip = parse_source_ip(
                    conn.send_command(cmds["nve"], read_timeout=60), os_key)
            except Exception:  # noqa: BLE001 - source IP is an enrichment
                pass

        for ip, state in f.peers.items():
            if state.lower() != "up":
                f.issues.append(f"VTEP peer {ip} is {state}")
        for vni, state in f.vnis.items():
            if state.lower() not in ("up", "established"):
                f.issues.append(f"VNI {vni} is {state}")
        for nbr, state in f.bgp.items():
            if state != "Established":
                f.issues.append(f"BGP EVPN neighbor {nbr} is {state}")

        expected = dev.get("expected_vnis")
        if expected is not None and len(f.vnis) < int(expected):
            f.issues.append(f"only {len(f.vnis)} VNI(s) present, expected {expected}")

        return f

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


# ---------------------------------------------------------------------------
# Fabric-wide analysis
# ---------------------------------------------------------------------------

def analyse_fabric(fabrics: list[Fabric]) -> list[str]:
    """
    Compare leaves against each other.

    A leaf reporting its own peers and VNIs as healthy tells you nothing about
    whether the FABRIC is healthy. The failures that matter are asymmetric:
    leaf-01 sees leaf-04 but leaf-04 does not see leaf-01; a VNI configured on
    three leaves and missing from a fourth. Neither is visible from any single
    device.
    """
    issues: list[str] = []
    live = [f for f in fabrics if f.status == "ok"]
    if len(live) < 2:
        return issues

    # Peer symmetry
    by_source = {f.source_ip: f for f in live if f.source_ip}
    for f in live:
        for peer_ip, state in f.peers.items():
            if state.lower() != "up":
                continue
            other = by_source.get(peer_ip)
            if other and f.source_ip and f.source_ip not in other.peers:
                issues.append(
                    f"asymmetric adjacency: {f.host} sees VTEP {peer_ip} ({other.host}), "
                    f"but {other.host} does not list {f.source_ip} as a peer")

    # Peer-count outliers
    counts = {f.host: len([s for s in f.peers.values() if s.lower() == "up"]) for f in live}
    if counts:
        expected = max(set(counts.values()), key=list(counts.values()).count)
        for host, n in sorted(counts.items()):
            if n < expected:
                issues.append(f"{host}: {n} VTEP peer(s) up, most leaves report "
                              f"{expected} - {expected - n} tunnel(s) missing")

    # VNI presence across the fabric
    everywhere: dict[str, set[str]] = defaultdict(set)
    for f in live:
        for vni in f.vnis:
            everywhere[vni].add(f.host)
    hosts = {f.host for f in live}
    for vni, present in sorted(everywhere.items()):
        if len(present) < len(hosts):
            missing = sorted(hosts - present)
            issues.append(f"VNI {vni}: on {len(present)}/{len(hosts)} leaves, "
                          f"missing from {', '.join(missing[:4])}"
                          + (f" +{len(missing)-4}" if len(missing) > 4 else ""))

    # BGP EVPN session counts
    bgp_counts = {f.host: sum(1 for s in f.bgp.values() if s == "Established") for f in live}
    if bgp_counts:
        norm = max(set(bgp_counts.values()), key=list(bgp_counts.values()).count)
        for host, n in sorted(bgp_counts.items()):
            if n < norm:
                issues.append(f"{host}: {n} BGP EVPN session(s) established, "
                              f"most leaves have {norm}")
    return issues


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Monitor VXLAN VTEP, VNI and BGP EVPN health across a fabric.",
        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 fabric leaves and spines")
    ap.add_argument("--daemon", action="store_true",
                    help="Run continuously instead of checking once and exiting")
    ap.add_argument("--interval", type=int, default=60,
                    help="Seconds between polls in daemon mode. Default: 60")
    ap.add_argument("--limit", help="Only check hosts whose name contains this text")
    ap.add_argument("--show-detail", action="store_true",
                    help="List every VTEP peer and VNI per device")
    ap.add_argument("--read-timeout", type=int, default=90,
                    help="Seconds to wait per command. Default: 90")
    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 VXLAN fabric health on {len(devices)} device(s)...")
        print("=" * 86)

        fabrics: list[Fabric] = []
        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:
                    fabrics.append(fut.result())
                except Exception as exc:  # noqa: BLE001
                    f = Fabric(host=host, os="?")
                    f.status, f.error = "error", f"worker crashed: {exc}"
                    fabrics.append(f)

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

            up = sum(1 for s in f.peers.values() if s.lower() == "up")
            vni_up = sum(1 for s in f.vnis.values()
                         if s.lower() in ("up", "established"))
            bgp_up = sum(1 for s in f.bgp.values() if s == "Established")
            tag = "[OK]   " if not f.issues else "[WARN] "
            src = f"  src {f.source_ip}" if f.source_ip else ""
            print(f"{tag} {label:<26} VTEP {up}/{len(f.peers)}  "
                  f"VNI {vni_up}/{len(f.vnis)}  BGP EVPN {bgp_up}/{len(f.bgp)}{src}")
            for i in f.issues:
                print(f"           - {i}")
            if args.show_detail:
                for ip, st in sorted(f.peers.items()):
                    print(f"           peer {ip:<16} {st}")
                for vni, st in sorted(f.vnis.items(), key=lambda x: int(x[0]) if x[0].isdigit() else 0):
                    print(f"           vni  {vni:<16} {st}")

        fabric_issues = analyse_fabric(fabrics)
        print("=" * 86)
        if fabric_issues:
            print(f"FABRIC CONSISTENCY: {len(fabric_issues)} issue(s)\n")
            for i in fabric_issues:
                print(f"  ! {i}")
        else:
            live = sum(1 for f in fabrics if f.status == "ok")
            print(f"FABRIC CONSISTENCY: symmetric across {live} node(s)")

        device_issues = sum(len(f.issues) for f in fabrics)
        errs = sum(1 for f in fabrics if f.status == "error")
        print(f"\n{device_issues} device-level issue(s), {len(fabric_issues)} "
              f"fabric-level issue(s), {errs} unreachable.")

        if args.json_out:
            Path(args.json_out).write_text(json.dumps(
                {"devices": [asdict(f) for f in fabrics], "fabric_issues": fabric_issues},
                indent=2), encoding="utf-8")

        return 1 if (device_issues or fabric_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)