BGP Route & Path Selection Analyzer

#!/usr/bin/env python3
"""
BGP Route & Path Selection Analyzer

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/bgp-route-analyzer

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
from collections import defaultdict

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

PREFIX_CMDS = {
    "ios": "show ip bgp {prefix}",
    "iosxe": "show ip bgp {prefix}",
    "iosxr": "show bgp ipv4 unicast {prefix}",
    "nxos": "show bgp ipv4 unicast {prefix}",
    "junos": "show route {prefix} detail protocol bgp",
    "eos": "show ip bgp {prefix}",
    "fortios": "get router info bgp network {prefix}",
}

VRF_PREFIX_CMDS = {
    "ios": "show ip bgp vpnv4 vrf {vrf} {prefix}",
    "iosxe": "show ip bgp vpnv4 vrf {vrf} {prefix}",
    "iosxr": "show bgp vrf {vrf} ipv4 unicast {prefix}",
    "nxos": "show bgp vrf {vrf} ipv4 unicast {prefix}",
    "junos": "show route table {vrf}.inet.0 {prefix} detail protocol bgp",
    "eos": "show ip bgp {prefix} vrf {vrf}",
    "fortios": "get router info bgp network {prefix}",
}

NO_BGP = {"panos": "PAN-OS exposes BGP paths only through its XML API"}

# The BGP decision process, in the order the RFC and every vendor applies it.
# Each entry returns a sort key where LOWER wins, so one comparison works for
# attributes where higher is better (weight, local-pref) and where lower is
# better (AS-path length, MED).
DECISION_STEPS = [
    ("WEIGHT",      lambda p: -p.get("weight", 0),
     "highest weight (Cisco-proprietary, local to this router)"),
    ("LOCAL_PREF",  lambda p: -p.get("locpref", 100),
     "highest local-preference"),
    ("LOCAL_ORIG",  lambda p: 0 if p.get("local_origin") else 1,
     "locally originated route preferred"),
    ("AS_PATH",     lambda p: _aspath_len(p),
     "shortest AS-PATH"),
    ("ORIGIN",      lambda p: {"i": 0, "igp": 0, "e": 1, "egp": 1}.get(
         str(p.get("origin", "?")).lower(), 2),
     "best origin code (IGP < EGP < Incomplete)"),
    ("MED",         lambda p: p.get("metric", 0),
     "lowest MED"),
    ("EBGP_IBGP",   lambda p: 1 if p.get("ibgp") else 0,
     "eBGP preferred over iBGP"),
    ("IGP_METRIC",  lambda p: p.get("igp_metric", 0),
     "lowest IGP metric to the next-hop"),
    ("ROUTER_ID",   lambda p: _ip_key(p.get("router_id", "255.255.255.255")),
     "lowest BGP router-id"),
    ("NEIGHBOR_IP", lambda p: _ip_key(p.get("nexthop", "255.255.255.255")),
     "lowest neighbor address"),
]


def _aspath_len(p: dict) -> int:
    ap = (p.get("aspath") or "").strip()
    if not ap or ap.lower() in ("local", "i", "?"):
        return 0
    return len([t for t in ap.split() if t.isdigit()])


def _ip_key(ip: str) -> tuple:
    try:
        return tuple(int(o) for o in ip.split("."))
    except (ValueError, AttributeError):
        return (255, 255, 255, 255)


# ---------------------------------------------------------------------------
# Parsing
# ---------------------------------------------------------------------------

def parse_cisco(output: str) -> list[dict]:
    """Parse IOS / IOS-XE / NX-OS / EOS `show ip bgp <prefix>` output."""
    paths: list[dict] = []
    cur: dict | None = None

    for line in output.splitlines():
        # The AS-path line starts a new path block; it is the first line after
        # "  Refresh Epoch N" or follows the previous block's attributes.
        m = re.match(r"^\s{2}([\d\s]*|Local)\s*$", line)
        if m and m.group(1).strip():
            if cur:
                paths.append(cur)
            ap = m.group(1).strip()
            cur = {"aspath": "" if ap == "Local" else ap,
                   "local_origin": ap == "Local",
                   "weight": 0, "locpref": 100, "metric": 0,
                   "origin": "?", "best": False, "ibgp": False,
                   "communities": [], "nexthop": "", "router_id": ""}
            continue

        if cur is None:
            continue

        m = re.search(r"^\s+(\d{1,3}(?:\.\d{1,3}){3})\s+from\s+"
                      r"(\d{1,3}(?:\.\d{1,3}){3})\s+\(([\d.]+)\)", line)
        if m:
            cur["nexthop"] = m.group(1)
            cur["peer"] = m.group(2)
            cur["router_id"] = m.group(3)
            continue

        m = re.search(r"Origin (\S+?),", line)
        if m:
            cur["origin"] = {"IGP": "i", "EGP": "e"}.get(m.group(1), "?")
        m = re.search(r"metric (\d+)", line)
        if m:
            cur["metric"] = int(m.group(1))
        m = re.search(r"localpref (\d+)", line)
        if m:
            cur["locpref"] = int(m.group(1))
        m = re.search(r"weight (\d+)", line)
        if m:
            cur["weight"] = int(m.group(1))
        if "internal" in line:
            cur["ibgp"] = True
        if re.search(r"\bbest\b", line):
            cur["best"] = True
        m = re.search(r"Community:\s*(.+)", line)
        if m:
            cur["communities"].extend(
                re.findall(r"\d+:\d+|no-export|no-advertise|local-AS", m.group(1)))

    if cur:
        paths.append(cur)
    return paths


def parse_iosxr(output: str) -> list[dict]:
    paths: list[dict] = []
    cur: dict | None = None

    for line in output.splitlines():
        m = re.match(r"\s+Next hop:\s+(\S+)", line)
        if m:
            if cur:
                paths.append(cur)
            cur = {"nexthop": m.group(1), "locpref": 100, "weight": 0, "metric": 0,
                   "aspath": "", "origin": "?", "best": False, "ibgp": False,
                   "local_origin": False, "communities": [], "router_id": ""}
            continue
        if cur is None:
            continue
        for pat, key, cast in (
            (r"Local pref:\s+(\d+)", "locpref", int),
            (r"MED:\s+(\d+)", "metric", int),
            (r"Weight:\s+(\d+)", "weight", int),
            (r"AS-path:\s+(.*)", "aspath", str),
            (r"Origin:\s+(\S+)", "origin", str),
            (r"Router ID:\s+(\S+)", "router_id", str),
        ):
            m = re.search(pat, line)
            if m:
                cur[key] = cast(m.group(1).strip())
        if "best path" in line.lower() or re.search(r"\bbest\b", line):
            cur["best"] = True
        if "internal" in line.lower():
            cur["ibgp"] = True
        m = re.search(r"Community:\s*(.+)", line)
        if m:
            cur["communities"].extend(
                re.findall(r"\d+:\d+|no-export|no-advertise", m.group(1)))

    if cur:
        paths.append(cur)
    return paths


def parse_junos(output: str) -> list[dict]:
    paths: list[dict] = []
    cur: dict | None = None

    for line in output.splitlines():
        m = re.search(r"Next hop:\s+(\d{1,3}(?:\.\d{1,3}){3})", line)
        if m:
            if cur:
                paths.append(cur)
            cur = {"nexthop": m.group(1), "locpref": 100, "weight": 0, "metric": 0,
                   "aspath": "", "origin": "?", "best": False, "ibgp": False,
                   "local_origin": False, "communities": [], "router_id": ""}
            continue
        if cur is None:
            continue
        m = re.search(r"Localpref:\s+(\d+)", line)
        if m:
            cur["locpref"] = int(m.group(1))
        m = re.search(r"Metric:\s+(\d+)", line)
        if m:
            cur["metric"] = int(m.group(1))
        m = re.search(r"AS path:\s+(.*?)(?:\s+I|\s+E|\s+\?|$)", line)
        if m:
            cur["aspath"] = m.group(1).strip()
        if re.search(r"AS path:.*\bI\b", line):
            cur["origin"] = "i"
        elif re.search(r"AS path:.*\bE\b", line):
            cur["origin"] = "e"
        m = re.search(r"Communities:\s*(.+)", line)
        if m:
            cur["communities"].extend(re.findall(r"\d+:\d+|no-export", m.group(1)))
        if "Active" in line and "Inactive" not in line:
            cur["best"] = True
        m = re.search(r"Peer ID:\s+(\S+)", line)
        if m:
            cur["router_id"] = m.group(1)

    if cur:
        paths.append(cur)
    return paths


# ---------------------------------------------------------------------------
# Decision process
# ---------------------------------------------------------------------------

def rank_paths(paths: list[dict]) -> tuple[list[dict], str]:
    """
    Walk the BGP decision process and record which step eliminated each path.

    Returns (paths, deciding_step). The deciding step is the LAST comparison
    that narrowed the field - the one that actually chose the winner. Reporting
    the first elimination instead would name a step that merely removed an
    also-ran.
    """
    if len(paths) <= 1:
        if paths:
            paths[0]["best"] = True
        return paths, "only one path available"

    for p in paths:
        p["best"] = False
        p.pop("eliminated_by", None)

    survivors = list(range(len(paths)))
    deciding = "no attribute differed - selection fell through to a tiebreaker"

    for name, key_fn, reason in DECISION_STEPS:
        if len(survivors) <= 1:
            break
        try:
            scores = {i: key_fn(paths[i]) for i in survivors}
        except Exception:  # noqa: BLE001 - a missing attribute must not abort the walk
            continue
        best_val = min(scores.values())
        kept = [i for i in survivors if scores[i] == best_val]
        if len(kept) < len(survivors):
            for i in survivors:
                if i not in kept:
                    paths[i]["eliminated_by"] = f"{name}: {reason}"
            deciding = f"{name}: {reason}"
            survivors = kept

    if survivors:
        paths[survivors[0]]["best"] = True
    return paths, deciding


# ---------------------------------------------------------------------------
# Per-device analysis
# ---------------------------------------------------------------------------

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

    if os_key in NO_BGP:
        return {"host": host, "os": os_key, "status": "skip", "note": NO_BGP[os_key]}

    try:
        conn = connect(dev)
        table = VRF_PREFIX_CMDS if args.vrf else PREFIX_CMDS
        cmd = table[os_key].format(prefix=args.prefix, vrf=args.vrf or "")
        raw = conn.send_command(cmd, read_timeout=args.read_timeout)

        if os_key == "iosxr":
            paths = parse_iosxr(raw)
        elif os_key == "junos":
            paths = parse_junos(raw)
        else:
            paths = parse_cisco(raw)

        if not paths:
            return {"host": host, "os": os_key, "status": "nopath",
                    "note": f"no BGP path for {args.prefix}", "raw": raw[:600]}

        paths, deciding = rank_paths(paths)
        best = next((p for p in paths if p.get("best")), None)
        return {"host": host, "os": os_key, "status": "ok", "paths": paths,
                "deciding": deciding,
                "best_nexthop": best.get("nexthop") if best else "",
                "best_aspath": (best.get("aspath") or "local") if best else ""}

    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


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Explain which BGP path wins for a prefix, and why.",
        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 routers to query")
    ap.add_argument("--prefix", required=True,
                    help="The prefix to analyse, e.g. 203.0.113.0/24")
    ap.add_argument("--vrf", help="Analyse inside this VRF instead of the global table")
    ap.add_argument("--limit", help="Only query hosts whose name contains this text")
    ap.add_argument("--show-all-paths", action="store_true",
                    help="Print every path with its attributes, not just the winner")
    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 routers to query 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

    scope = f" in VRF {args.vrf}" if args.vrf else ""
    print(f"Analysing {args.prefix}{scope} on {len(devices)} router(s)...")
    print("=" * 84)

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

    for r in sorted(results, key=lambda x: x["host"]):
        label = f"{r['host']} ({r['os']})"
        if r["status"] == "error":
            print(f"[ERROR] {label:<26} {r['error']}")
            continue
        if r["status"] == "skip":
            print(f"[SKIP]  {label:<26} {r['note']}")
            continue
        if r["status"] == "nopath":
            print(f"[NONE]  {label:<26} {r['note']}")
            continue

        n = len(r["paths"])
        print(f"\n{label}  -  {n} path(s)")
        print(f"   WINNER  next-hop {r['best_nexthop']}  via AS-path "
              f"{r['best_aspath'] or 'local'}")
        print(f"   DECIDED BY  {r['deciding']}")

        if args.show_all_paths:
            for i, p in enumerate(r["paths"], 1):
                mark = "BEST" if p.get("best") else "    "
                print(f"   [{mark}] path {i}: nh {p.get('nexthop', '?'):<16} "
                      f"w={p.get('weight', 0)} lp={p.get('locpref', 100)} "
                      f"as={p.get('aspath') or 'local'} "
                      f"origin={p.get('origin', '?')} med={p.get('metric', 0)}"
                      + ("  iBGP" if p.get("ibgp") else ""))
                if p.get("communities"):
                    print(f"            communities: {' '.join(p['communities'])}")
                if p.get("eliminated_by"):
                    print(f"            eliminated by {p['eliminated_by']}")

    # Different routers choosing different exits is the finding that matters
    # when you are asking why traffic from site A takes a different path.
    live = [r for r in results if r["status"] == "ok"]
    exits = defaultdict(list)
    for r in live:
        exits[r["best_nexthop"]].append(r["host"])

    print("\n" + "=" * 84)
    if len(exits) > 1:
        print(f"ROUTERS DISAGREE: {len(exits)} different best next-hops for {args.prefix}")
        for nh, hosts in sorted(exits.items()):
            print(f"  {nh:<18} chosen by {', '.join(sorted(hosts))}")
        print("\nThat is expected when the routers sit at different points in the\n"
              "topology, and a problem when they should be symmetric.")
    elif live:
        nh = next(iter(exits))
        print(f"All {len(live)} router(s) agree: best next-hop is {nh}")

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

    errs = sum(1 for r in results if r["status"] == "error")
    return 1 if errs else 0


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