OSPF Neighbor State Checker

#!/usr/bin/env python3
"""
OSPF Neighbor State Checker

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

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

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


# ---------------------------------------------------------------------------
# OSPF neighbor commands and parsing
# ---------------------------------------------------------------------------

COMMAND_MAP = {
    "ios":     {"global": "show ip ospf neighbor"},
    "iosxe":   {"global": "show ip ospf neighbor"},
    "nxos":    {"global": "show ip ospf neighbors",
                "vrf_fmt": "show ip ospf neighbors vrf {vrf}"},
    "iosxr":   {"global": "show ospf neighbor",
                "vrf_fmt": "show ospf vrf {vrf} neighbor"},
    "junos":   {"global": "show ospf neighbor",
                "vrf_fmt": "show ospf neighbor instance {vrf}"},
    "eos":     {"global": "show ip ospf neighbor",
                "vrf_fmt": "show ip ospf neighbor vrf {vrf}"},
    "fortios": {"global": "get router info ospf neighbor"},
    # PAN-OS runs OSPF but exposes it only through the XML API, not a parseable
    # CLI table. Those devices are reported as skipped rather than guessed at.
}

# A healthy adjacency is FULL. On a broadcast segment, DR-OTHER routers settle
# at 2WAY with each other by design - that is not a fault, so it is only
# flagged when --strict-2way is used.
HEALTHY = {"FULL"}
BENIGN = {"2WAY"}

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


def parse_neighbors(output: str, os_key: str) -> list[tuple[str, str, str]]:
    """
    Return a list of (neighbor_id, state, interface).

    Column layouts differ per platform:
      IOS/XE/XR/NX-OS/EOS   Neighbor-ID  Pri  State        Dead  Address  Interface
      JunOS                 Address      Interface  State  ID    Pri      Dead
      FortiOS               Neighbor-ID  Pri  State        Dead  Address  Interface
    """
    out: list[tuple[str, str, str]] = []

    for line in output.splitlines():
        parts = line.split()
        if len(parts) < 3 or not IPV4.match(parts[0]):
            continue

        if os_key == "junos":
            # Address Interface State ID Pri Dead  -> state is column 3
            state = parts[2].upper()
            iface = parts[1]
            nbr = parts[3] if len(parts) > 3 and IPV4.match(parts[3]) else parts[0]
        else:
            # Neighbor-ID Pri State Dead Address Interface -> state is column 3,
            # often written FULL/DR or FULL/BDR; the role after the slash is not
            # a health signal.
            state = parts[2].split("/")[0].upper()
            iface = parts[-1]
            nbr = parts[0]

        out.append((nbr, state, iface))
    return out


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

def check_device(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    cmds = COMMAND_MAP.get(os_key)
    conn = None

    if not cmds:
        return {"host": host, "os": os_key, "status": "skip",
                "note": f"OSPF neighbor table is not readable over CLI on {os_key}"}

    expected = dev.get("expected_neighbors")
    issues: list[str] = []
    neighbors: list[tuple[str, str, str]] = []

    try:
        conn = connect(dev)

        contexts = [("global", cmds["global"])]
        vrf_fmt = cmds.get("vrf_fmt")
        for vrf in dev.get("vrfs", []):
            if vrf_fmt:
                contexts.append((vrf, vrf_fmt.format(vrf=vrf)))

        for ctx, cmd in contexts:
            raw = conn.send_command(cmd, read_timeout=args.read_timeout)
            found = parse_neighbors(raw, os_key)
            neighbors.extend(found)
            where = "" if ctx == "global" else f" (VRF {ctx})"

            for nbr, state, iface in found:
                if state in HEALTHY:
                    continue
                if state in BENIGN and not args.strict_2way:
                    continue
                issues.append(f"{nbr} on {iface} is {state}{where}")

        # A neighbor stuck in EXSTART is visible. A neighbor that disappeared
        # entirely is not - unless you declare how many you expect.
        if expected is not None and len(neighbors) < int(expected):
            issues.append(
                f"only {len(neighbors)} neighbor(s) present, expected {expected} "
                f"- {int(expected) - len(neighbors)} adjacency is missing entirely"
            )

        return {"host": host, "os": os_key, "status": "ok",
                "total": len(neighbors),
                "full": sum(1 for _, s, _ in neighbors if s in HEALTHY),
                "expected": expected, "issues": issues,
                "neighbors": [{"id": n, "state": s, "interface": i}
                              for n, s, i in neighbors]}

    except NetmikoAuthenticationException as exc:
        return {"host": host, "os": os_key, "status": "error",
                "error": f"authentication failed - check NET_USER/NET_PASS ({exc})"}
    except NetmikoTimeoutException as exc:
        return {"host": host, "os": os_key, "status": "error",
                "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",
                "error": f"{type(exc).__name__}: {exc}"}
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass


# ---------------------------------------------------------------------------
# What each stuck state actually means
# ---------------------------------------------------------------------------

STATE_HELP = {
    "DOWN":     "no hellos received - check the interface, VLAN, and that OSPF is enabled on it",
    "ATTEMPT":  "NBMA only - hellos sent to a configured neighbor with no reply",
    "INIT":     "hellos received but ours are not reaching the peer - one-way. Check an ACL or an MTU mismatch",
    "2WAY":     "normal between DR-OTHER routers on a broadcast segment; a fault only if this pair should be adjacent",
    "EXSTART":  "classic MTU mismatch between the two interfaces, or a duplicate router-id",
    "EXCHANGE": "DBD exchange stalled - almost always MTU, sometimes an ACL dropping large packets",
    "LOADING":  "LSA request stalled - usually a corrupt or oversized LSA",
}


def explain(issues: list[str]) -> list[str]:
    """Attach the likely cause to each stuck state, so the output is actionable."""
    notes = []
    for state, help_text in STATE_HELP.items():
        if any(f"is {state}" in i for i in issues):
            notes.append(f"{state}: {help_text}")
    return notes


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Check OSPF neighbor adjacencies across many devices.",
        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 check")
    ap.add_argument("--limit", help="Only check hosts whose name contains this text")
    ap.add_argument("--strict-2way", action="store_true",
                    help="Treat 2WAY as a fault. By default it is accepted, because "
                         "DR-OTHER routers on a broadcast segment sit at 2WAY by design")
    ap.add_argument("--show-all", action="store_true",
                    help="List every neighbor, not just the ones with problems")
    ap.add_argument("--read-timeout", type=int, default=60,
                    help="Seconds to wait for each 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

    print(f"Checking OSPF neighbors on {len(devices)} device(s)...")
    print("=" * 74)

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

    for r in sorted(results, key=lambda x: x["host"]):
        label = f"{r['host']} ({r['os']})"
        if r["status"] == "error":
            print(f"[ERROR] {label:<28} {r['error']}")
        elif r["status"] == "skip":
            print(f"[SKIP]  {label:<28} {r['note']}")
        elif r["issues"]:
            exp = f"/{r['expected']} expected" if r.get("expected") else ""
            print(f"[WARN]  {label:<28} {r['full']}/{r['total']}{exp} FULL")
            for i in r["issues"]:
                print(f"           - {i}")
        else:
            print(f"[OK]    {label:<28} {r['total']} neighbor(s), all FULL")

        if args.show_all and r.get("neighbors"):
            for n in r["neighbors"]:
                print(f"           {n['id']:<17} {n['state']:<10} {n['interface']}")

    all_issues = [i for r in results for i in r.get("issues", [])]
    notes = explain(all_issues)

    print("=" * 74)
    ok = sum(1 for r in results if r["status"] == "ok" and not r["issues"])
    warn = sum(1 for r in results if r["status"] == "ok" and r["issues"])
    err = sum(1 for r in results if r["status"] == "error")
    print(f"{ok} healthy, {warn} with issues, {err} unreachable "
          f"(of {len(devices)} device(s)).")

    if notes:
        print("\nWhat these states mean:")
        for n in notes:
            print(f"  {n}")

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

    return 1 if (warn or err) else 0


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