Running vs Startup Config Diff

#!/usr/bin/env python3
"""
Running vs Startup Config Diff

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/config-diff

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 difflib
from datetime import datetime, timezone

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

# Running config, per OS.
RUNNING = {
    "ios": "show running-config",
    "iosxe": "show running-config",
    "iosxr": "show running-config",
    "nxos": "show running-config",
    "junos": "show configuration | display set",
    "eos": "show running-config",
    "panos": "show config running",
    "fortios": "show full-configuration",
}

# Saved/startup config, per OS. Platforms with a commit model have no separate
# startup file - what is committed IS what boots - so they are compared a
# different way (see STARTUP_NOTE).
STARTUP = {
    "ios": "show startup-config",
    "iosxe": "show startup-config",
    "nxos": "show startup-config",
    "eos": "show startup-config",
    "iosxr": None,
    "junos": None,
    "panos": "show config saved",
    "fortios": None,
}

STARTUP_NOTE = {
    "iosxr": "IOS-XR commits to persistent storage - there is no unsaved state",
    "junos": "Junos commits to persistent storage - there is no unsaved state",
    "fortios": "FortiOS saves immediately - there is no unsaved state",
}

# Lines that differ on every read even when nothing changed.
VOLATILE = (
    "Building configuration",
    "Current configuration",
    "! Last configuration change",
    "! NVRAM config last updated",
    "ntp clock-period",
    "!Time:",
    "## Last commit:",
    "Time: ",
)


def normalise(text: str) -> list[str]:
    return [
        ln.rstrip() for ln in text.splitlines()
        if ln.strip() and not any(ln.strip().startswith(v) for v in VOLATILE)
    ]


def newest_backup(backup_dir: str, host: str) -> Path | None:
    d = Path(backup_dir) / host
    if not d.exists():
        return None
    files = sorted(d.glob("*.cfg"), key=lambda p: p.stat().st_mtime, reverse=True)
    return files[0] if files else None


# ---------------------------------------------------------------------------
# Per-device comparison
# ---------------------------------------------------------------------------

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

    try:
        conn = connect(dev)
        running_raw = conn.send_command(RUNNING[os_key], read_timeout=args.read_timeout)
        if not running_raw.strip():
            return {"host": host, "os": os_key, "status": "error",
                    "error": "running-config came back empty - check account privilege"}
        running = normalise(running_raw)

        if args.mode == "backup":
            prev = newest_backup(args.backup_dir, host)
            if not prev:
                return {"host": host, "os": os_key, "status": "skip",
                        "note": f"no previous backup under {args.backup_dir}/{host}/"}
            baseline = normalise(prev.read_text(encoding="utf-8"))
            from_label, to_label = f"backup ({prev.name})", "running"
        else:
            startup_cmd = STARTUP.get(os_key)
            if not startup_cmd:
                return {"host": host, "os": os_key, "status": "skip",
                        "note": STARTUP_NOTE.get(os_key, f"no startup config on {os_key}")}
            startup_raw = conn.send_command(startup_cmd, read_timeout=args.read_timeout)
            baseline = normalise(startup_raw)
            from_label, to_label = "startup", "running"

        diff = list(difflib.unified_diff(
            baseline, running, fromfile=from_label, tofile=to_label,
            lineterm="", n=args.context,
        ))
        changed = [d for d in diff
                   if d.startswith(("+", "-")) and not d.startswith(("---", "+++"))]
        added = sum(1 for d in changed if d.startswith("+"))
        removed = sum(1 for d in changed if d.startswith("-"))

        if changed and args.save_diffs:
            d = Path(args.diff_dir)
            d.mkdir(parents=True, exist_ok=True)
            ts = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
            (d / f"{host}_{ts}.diff").write_text("\n".join(diff) + "\n", encoding="utf-8")

        return {"host": host, "os": os_key, "status": "ok",
                "added": added, "removed": removed,
                "from": from_label, "to": to_label,
                "diff": diff if changed 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="Find configuration that is running but not saved, or that "
                    "drifted since the last backup.",
        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 compare")
    ap.add_argument("--mode", choices=["startup", "backup"], default="startup",
                    help="startup: running vs startup-config, finds unsaved changes. "
                         "backup: running vs the newest file from config-backup, "
                         "finds drift since that backup. Default: startup")
    ap.add_argument("--backup-dir", default="./backups",
                    help="Where config-backup wrote its files. Used by --mode backup")
    ap.add_argument("--limit", help="Only compare hosts whose name contains this text")
    ap.add_argument("--show-diff", action="store_true",
                    help="Print the actual changed lines, not just the counts")
    ap.add_argument("--context", type=int, default=2,
                    help="Lines of unchanged context around each change. Default: 2")
    ap.add_argument("--save-diffs", action="store_true",
                    help="Write each diff to a file under --diff-dir")
    ap.add_argument("--diff-dir", default="./diffs")
    ap.add_argument("--read-timeout", type=int, default=180,
                    help="Seconds to wait per config download. Default: 180")
    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

    what = ("running vs startup (unsaved changes)" if args.mode == "startup"
            else f"running vs newest backup in {args.backup_dir}/")
    print(f"Comparing {what} on {len(devices)} device(s)...")
    print("=" * 74)

    results = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(diff_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}"})

    drifted = 0
    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 not r["diff"]:
            print(f"[SAME]  {label:<28} no difference")
        else:
            drifted += 1
            print(f"[DIFF]  {label:<28} +{r['added']} -{r['removed']} "
                  f"({r['from']} -> {r['to']})")
            if args.show_diff:
                for line in r["diff"]:
                    print(f"           {line}")

    print("=" * 74)
    same = sum(1 for r in results if r["status"] == "ok" and not r["diff"])
    err = sum(1 for r in results if r["status"] == "error")
    skip = sum(1 for r in results if r["status"] == "skip")
    print(f"{same} identical, {drifted} with differences, {skip} skipped, "
          f"{err} unreachable (of {len(devices)} device(s)).")

    if drifted and args.mode == "startup":
        print("\nDevices with differences have configuration that will be LOST on reload.")
        print("Save it with 'write memory' (IOS/EOS) or 'copy run start' (NX-OS).")

    if args.json_out:
        slim = [{k: v for k, v in r.items() if k != "diff"} for r in results]
        Path(args.json_out).write_text(json.dumps(slim, indent=2), encoding="utf-8")

    return 1 if (drifted or err) else 0


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