BGP Neighbor State Monitor

#!/usr/bin/env python3
"""
BGP Neighbor State Monitor

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/bgp-neighbor-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" "tabulate>=0.9"
"""

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 logging
import logging.handlers
import re
import time

try:
    import requests
except ImportError:
    requests = None
try:
    from tabulate import tabulate
except ImportError:
    tabulate = 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
# ---------------------------------------------------------------------------

BGP_CMDS = {
    "ios":     {"global": "show ip bgp summary",
                "vrf": "show ip bgp vpnv4 vrf {vrf} summary"},
    "iosxe":   {"global": "show ip bgp summary",
                "vrf": "show ip bgp vpnv4 vrf {vrf} summary"},
    "iosxr":   {"global": "show bgp summary",
                "vrf": "show bgp vrf {vrf} summary"},
    "nxos":    {"global": "show bgp summary",
                "vrf": "show bgp vrf {vrf} summary"},
    "junos":   {"global": "show bgp summary",
                "vrf": "show bgp summary instance {vrf}"},
    "eos":     {"global": "show ip bgp summary",
                "vrf": "show ip bgp vrf {vrf} summary"},
    "panos":   {"global": "show routing protocol bgp peer",
                "vrf": "show routing protocol bgp peer virtual-router {vrf}"},
    "fortios": {"global": "get router info bgp summary",
                "vrf": "get router info bgp summary vrf {vrf}"},
}

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

# Every FSM state that is not Established, spelled as each platform prints it.
FSM_STATES = {"idle", "active", "connect", "opensent", "openconfirm",
              "established", "estab", "establ"}


def parse_bgp_summary(output: str, os_key: str, vrf: str) -> list[dict]:
    """
    Return [{peer, state, prefixes, vrf}].

    The last column of a BGP summary is State/PfxRcd: a number when the session
    is Established, an FSM state word otherwise. `prefixes` is None for a peer
    that is not Established, so "down" is never confused with "up and sending
    zero prefixes" - those mean completely different things.
    """
    peers: list[dict] = []

    if os_key == "junos":
        # Peer AS InPkt OutPkt OutQ Flaps Last-Up/Dwn State|#Active/Received/...
        for line in output.splitlines():
            parts = line.split()
            if len(parts) < 7 or not IPV4.match(parts[0]):
                continue
            tail = parts[-1]
            m = re.match(r"(\d+)/(\d+)/(\d+)/(\d+)", tail)
            if m:
                peers.append({"peer": parts[0], "state": "Established",
                              "prefixes": int(m.group(2)), "vrf": vrf,
                              "flaps_device": _int_or_none(parts, 5)})
            else:
                peers.append({"peer": parts[0], "state": tail, "prefixes": None,
                              "vrf": vrf, "flaps_device": _int_or_none(parts, 5)})
        return peers

    if os_key == "panos":
        for line in output.splitlines():
            parts = line.split()
            if len(parts) < 3 or not IPV4.match(parts[0]):
                continue
            state = next((p for p in parts if p.lower() in FSM_STATES), "unknown")
            nums = [int(p) for p in parts if p.isdigit()]
            established = state.lower() == "established"
            peers.append({"peer": parts[0],
                          "state": "Established" if established else state,
                          "prefixes": (nums[-1] if nums else None) if established else None,
                          "vrf": vrf, "flaps_device": None})
        return peers

    if os_key == "eos":
        for line in output.splitlines():
            parts = line.split()
            if len(parts) < 8 or not IPV4.match(parts[0]):
                continue
            state = next((p for p in parts if p.lower() in FSM_STATES), None)
            if state and state.lower().startswith("estab"):
                pfx = int(parts[-1]) if parts[-1].isdigit() else None
                peers.append({"peer": parts[0], "state": "Established",
                              "prefixes": pfx, "vrf": vrf, "flaps_device": None})
            elif state:
                peers.append({"peer": parts[0], "state": state, "prefixes": None,
                              "vrf": vrf, "flaps_device": None})
        return peers

    # IOS / IOS-XE / IOS-XR / NX-OS / FortiOS
    for line in output.splitlines():
        parts = line.split()
        if len(parts) < 8 or not IPV4.match(parts[0]):
            continue
        last = parts[-1]
        if last.isdigit():
            peers.append({"peer": parts[0], "state": "Established",
                          "prefixes": int(last), "vrf": vrf, "flaps_device": None})
        else:
            peers.append({"peer": parts[0], "state": last, "prefixes": None,
                          "vrf": vrf, "flaps_device": None})
    return peers


def _int_or_none(parts: list[str], idx: int) -> int | None:
    try:
        return int(parts[idx])
    except (IndexError, ValueError):
        return None


# ---------------------------------------------------------------------------
# State
# ---------------------------------------------------------------------------

def load_state(path: str) -> dict:
    try:
        return json.loads(Path(path).read_text(encoding="utf-8"))
    except FileNotFoundError:
        return {"poll_count": 0, "peers": {}}
    except json.JSONDecodeError as exc:
        raise SystemExit(f"State file {path} is corrupt: {exc}\n"
                         f"Delete it to start tracking again from scratch.")


def save_state(state: dict, path: str) -> None:
    tmp = Path(path).with_suffix(".tmp")
    tmp.write_text(json.dumps(state, indent=2), encoding="utf-8")
    tmp.replace(path)          # atomic, so a crash cannot truncate the file


def update_peer(state: dict, host: str, p: dict, args) -> list[str]:
    """
    Fold one observation into the tracked history and return any alerts.

    Called single-threaded after every worker has finished. Mutating this dict
    from inside the workers would be a data race on a shared structure.
    """
    key = f"{host}|{p['vrf']}|{p['peer']}"
    now = time.time()
    up = p["state"] == "Established"

    e = state["peers"].setdefault(key, {
        "host": host, "peer": p["peer"], "vrf": p["vrf"],
        "total_polls": 0, "established_polls": 0,
        "flaps": [], "last_state": None, "last_prefixes": None,
        "first_seen": now, "ever_established": False,
    })

    alerts: list[str] = []
    scope = "global" if p["vrf"] == "global" else f"vrf:{p['vrf']}"

    if e["last_state"] == "Established" and not up:
        e["flaps"].append(now)
        alerts.append(f"FLAP {host} [{scope}] peer {p['peer']} went "
                      f"Established -> {p['state']}")
    elif not up:
        if e["ever_established"]:
            alerts.append(f"DOWN {host} [{scope}] peer {p['peer']} is {p['state']}")
        else:
            alerts.append(f"NEVERUP {host} [{scope}] peer {p['peer']} is {p['state']} "
                          f"and has never reached Established since monitoring began")

    if up and e["last_prefixes"] and p["prefixes"] is not None and e["last_prefixes"] > 0:
        drop = (e["last_prefixes"] - p["prefixes"]) / e["last_prefixes"]
        if drop > args.drop_pct:
            alerts.append(f"DROP {host} [{scope}] peer {p['peer']} prefixes "
                          f"{e['last_prefixes']:,} -> {p['prefixes']:,} "
                          f"({drop*100:.1f}% down)")

    # Flap rate over a window beats a lifetime counter: a peer that flapped
    # twice a year ago is healthy; twice in the last hour is not.
    window = now - args.flap_window
    e["flaps"] = [t for t in e["flaps"] if t >= window]
    if len(e["flaps"]) >= args.flap_threshold:
        alerts.append(f"UNSTABLE {host} [{scope}] peer {p['peer']} flapped "
                      f"{len(e['flaps'])} time(s) in the last "
                      f"{int(args.flap_window/60)} minutes")

    e["total_polls"] += 1
    if up:
        e["established_polls"] += 1
        e["ever_established"] = True
    e["last_state"] = p["state"]
    if p["prefixes"] is not None:
        e["last_prefixes"] = p["prefixes"]
    return alerts


# ---------------------------------------------------------------------------
# Polling
# ---------------------------------------------------------------------------

def poll_device(dev: dict, args) -> dict:
    """Collect only. State is updated single-threaded by the caller."""
    host = dev["host"]
    os_key = dev["os"]
    cmds = BGP_CMDS[os_key]
    conn = None
    peers: list[dict] = []

    try:
        conn = connect(dev)
        for vrf in [None] + list(dev.get("vrfs", [])):
            scope = vrf or "global"
            cmd = cmds["global"] if vrf is None else cmds["vrf"].format(vrf=vrf)
            out = conn.send_command(cmd, read_timeout=args.read_timeout)
            peers.extend(parse_bgp_summary(out, os_key, scope))
        return {"host": host, "os": os_key, "status": "ok", "peers": peers}

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


def send_webhook(alerts: list[str], url: str) -> None:
    if not url or not alerts or requests is None:
        return
    try:
        requests.post(url, json={"text": "*BGP alerts*\n" + "\n".join(alerts)}, timeout=5)
    except Exception as exc:  # noqa: BLE001 - a failed webhook must not stop monitoring
        print(f"[WARN] webhook failed: {exc}")


def run_once(devices: list[dict], state: dict, args, syslog_logger) -> int:
    state["poll_count"] = state.get("poll_count", 0) + 1
    print(f"\nPolling BGP on {len(devices)} device(s)  [poll #{state['poll_count']}]")
    print("=" * 88)

    results = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(poll_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": "?", "status": "error", "peers": [],
                                "error": f"worker crashed: {exc}"})

    # Single-threaded from here: shared state and ordered output.
    all_alerts: list[str] = []
    rows = []
    for r in sorted(results, key=lambda x: x["host"]):
        if r["status"] == "error":
            print(f"[ERROR ] {r['host']:<22} {r['error']}")
            continue
        for p in sorted(r["peers"], key=lambda x: (x["vrf"], x["peer"])):
            alerts = update_peer(state, r["host"], p, args)
            all_alerts.extend(alerts)
            key = f"{r['host']}|{p['vrf']}|{p['peer']}"
            e = state["peers"][key]
            uptime = (e["established_polls"] / e["total_polls"] * 100
                      if e["total_polls"] else 0.0)
            scope = "[global]" if p["vrf"] == "global" else f"[vrf:{p['vrf']}]"
            pfx = f"{p['prefixes']:,}" if p["prefixes"] is not None else "-"

            if alerts:
                for a in alerts:
                    print(f"[{a.split()[0]:<7}] {r['host']:<22} {scope:<14} "
                          f"peer {p['peer']:<16} {a.split(' ', 1)[1]}")
            else:
                print(f"[OK     ] {r['host']:<22} {scope:<14} peer {p['peer']:<16} "
                      f"{p['state']}  {pfx:>10} prefixes  uptime {uptime:.1f}%")

            rows.append([p["peer"], r["host"], p["vrf"], p["state"], pfx,
                         f"{uptime:.1f}%", len(e["flaps"])])

    if rows and args.summary:
        hdrs = ["Peer", "Device", "VRF", "State", "Prefixes", "Uptime", "Flaps"]
        print("\nPeer summary:")
        if tabulate:
            print(tabulate(rows, headers=hdrs, tablefmt="simple"))
        else:
            print("  ".join(h.ljust(14) for h in hdrs))
            for row in rows:
                print("  ".join(str(c).ljust(14) for c in row))

    save_state(state, args.state_file)

    print("=" * 88)
    kinds = {}
    for a in all_alerts:
        kinds[a.split()[0]] = kinds.get(a.split()[0], 0) + 1
    errs = sum(1 for r in results if r["status"] == "error")
    if kinds or errs:
        print("  " + "   ".join(f"{k}={v}" for k, v in sorted(kinds.items()))
              + (f"   unreachable={errs}" if errs else ""))
    else:
        total = sum(len(r["peers"]) for r in results)
        print(f"  All {total} peer(s) Established across {len(results)} device(s).")

    if all_alerts:
        send_webhook(all_alerts, args.webhook)
        if syslog_logger:
            for a in all_alerts:
                syslog_logger.warning(a)

    return 1 if (all_alerts or errs) else 0


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Monitor BGP neighbor state, flaps and prefix counts.",
        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 devices to poll")
    ap.add_argument("--daemon", action="store_true",
                    help="Run continuously instead of polling once and exiting")
    ap.add_argument("--interval", type=int, default=60,
                    help="Seconds between polls in daemon mode. Default: 60")
    ap.add_argument("--drop-pct", type=float, default=0.10,
                    help="Alert when prefixes fall this fraction between polls. "
                         "Default: 0.10 (10 percent)")
    ap.add_argument("--flap-window", type=int, default=3600,
                    help="Seconds over which flaps are counted. Default: 3600")
    ap.add_argument("--flap-threshold", type=int, default=3,
                    help="Flaps within the window before a peer is called unstable. "
                         "Default: 3")
    ap.add_argument("--state-file", default="bgp_monitor_state.json",
                    help="Where peer history persists between runs")
    ap.add_argument("--webhook", default="",
                    help="Slack or Teams incoming webhook URL for alerts")
    ap.add_argument("--syslog", action="store_true", help="Also write alerts to syslog")
    ap.add_argument("--summary", action="store_true",
                    help="Print a table of every peer after the per-peer lines")
    ap.add_argument("--limit", help="Only poll hosts whose name contains this text")
    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 poll at once. Default: 10")
    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

    syslog_logger = None
    if args.syslog:
        syslog_logger = logging.getLogger("bgp_monitor")
        syslog_logger.setLevel(logging.WARNING)
        try:
            syslog_logger.addHandler(logging.handlers.SysLogHandler(address="/dev/log"))
        except OSError:
            syslog_logger.addHandler(logging.StreamHandler())

    state = load_state(args.state_file)
    print(f"Tracking {len(state.get('peers', {}))} peer(s) from {args.state_file}")

    if not args.daemon:
        return run_once(devices, state, args, syslog_logger)

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


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nStopped. State was saved after the last completed poll.", file=sys.stderr)
        sys.exit(130)