Network Device Config Backup

#!/usr/bin/env python3
"""
Network Device Config Backup

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

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


# ---------------------------------------------------------------------------
# Backup
# ---------------------------------------------------------------------------

# The command that prints the full running configuration, per OS.
COMMAND_MAP = {
    "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",
}

# Lines that change on every read (timestamps, uptime counters). Stripping them
# means two backups of an unchanged device produce identical files, so the
# --only-changed check and any git diff show real change only.
VOLATILE = (
    "Building configuration",
    "Current configuration",
    "! Last configuration change",
    "! NVRAM config last updated",
    "!Time:",
    "## Last commit:",
    "ntp clock-period",
)


def strip_volatile(text: str) -> str:
    return "\n".join(
        ln.rstrip() for ln in text.splitlines()
        if ln.strip() and not any(ln.strip().startswith(v) for v in VOLATILE)
    )


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


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

    try:
        conn = connect(dev)
        output = conn.send_command(command, read_timeout=args.read_timeout)

        if not output.strip():
            return {"host": host, "status": "error",
                    "error": f"'{command}' returned nothing - check account permissions"}

        # A truncated config is worse than no config: it looks like a valid
        # backup but cannot restore the device. Refuse anything implausibly short.
        if len(output.splitlines()) < args.min_lines:
            return {"host": host, "status": "error",
                    "error": f"only {len(output.splitlines())} lines returned "
                             f"(minimum {args.min_lines}) - output looks truncated"}

        normalised = strip_volatile(output)
        d = Path(args.backup_dir) / host
        d.mkdir(parents=True, exist_ok=True)

        if args.only_changed:
            prev = latest_backup(d)
            if prev and strip_volatile(prev.read_text(encoding="utf-8")) == normalised:
                return {"host": host, "status": "unchanged", "file": str(prev),
                        "lines": len(output.splitlines())}

        ts = datetime.now(timezone.utc).strftime("%Y-%m-%d_%H-%M-%S")
        path = d / f"{ts}.cfg"
        path.write_text(
            f"! Backup of {host} ({os_key}) taken {ts} UTC\n"
            f"! Command: {command}\n!\n{output}\n",
            encoding="utf-8",
        )
        return {"host": host, "status": "ok", "file": str(path),
                "lines": len(output.splitlines())}

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


def prune(backup_dir: str, host: str, keep: int) -> int:
    """Delete all but the `keep` most recent backups for one host."""
    d = Path(backup_dir) / host
    if not d.exists():
        return 0
    files = sorted(d.glob("*.cfg"), key=lambda p: p.stat().st_mtime, reverse=True)
    removed = 0
    for old in files[keep:]:
        try:
            old.unlink()
            removed += 1
        except OSError:
            pass
    return removed


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Back up running configuration from network 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 back up")
    ap.add_argument("--backup-dir", default="./backups",
                    help="Where to write the files. Default: ./backups")
    ap.add_argument("--limit",
                    help="Only back up hosts whose name contains this text")
    ap.add_argument("--only-changed", action="store_true",
                    help="Skip writing a new file when the config is identical to "
                         "the previous backup (ignoring timestamps)")
    ap.add_argument("--keep", type=int, default=0,
                    help="Keep only the N most recent backups per device. 0 = keep all")
    ap.add_argument("--min-lines", type=int, default=20,
                    help="Reject a backup shorter than this many lines, which almost "
                         "always means truncated output. Default: 20")
    ap.add_argument("--read-timeout", type=int, default=180,
                    help="Seconds to wait for the config to download. Raise it for "
                         "very large configs. 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

    print(f"Backing up {len(devices)} device(s) to {args.backup_dir}/")
    results = []

    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(backup_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, "status": "error", "error": f"worker crashed: {exc}"}
            results.append(r)
            if r["status"] == "ok":
                print(f"[OK]    {r['host']:<24} {r['lines']:>6} lines -> {r['file']}")
            elif r["status"] == "unchanged":
                print(f"[SAME]  {r['host']:<24} {r['lines']:>6} lines    no change since last backup")
            else:
                print(f"[FAIL]  {r['host']:<24} {r['error']}")

    if args.keep > 0:
        total = sum(prune(args.backup_dir, r["host"], args.keep)
                    for r in results if r["status"] in ("ok", "unchanged"))
        if total:
            print(f"\nPruned {total} old backup file(s), keeping {args.keep} per device.")

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

    ok = sum(1 for r in results if r["status"] == "ok")
    same = sum(1 for r in results if r["status"] == "unchanged")
    failed = sum(1 for r in results if r["status"] == "error")
    print(f"\nDone. {ok} saved, {same} unchanged, {failed} failed "
          f"(of {len(devices)} device(s)).")
    return 1 if failed else 0


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