Bulk Interface Configuration

#!/usr/bin/env python3
"""
Bulk Interface Configuration

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

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 logging
import os
import re
import sys
import time
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

try:
    import yaml
except ImportError:
    yaml = None

from netmiko import ConnectHandler
from netmiko.exceptions import NetmikoAuthenticationException, NetmikoTimeoutException

LOG = logging.getLogger("bulkintf")

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",
}

# Commands used to discover the current interface inventory and state.
DISCOVER = {
    "ios":    {"brief": "show ip interface brief", "status": "show interfaces status",
               "switch": "show interfaces switchport", "po": "show etherchannel summary",
               "run_intf": "show running-config | section ^interface"},
    "iosxe":  {"brief": "show ip interface brief", "status": "show interfaces status",
               "switch": "show interfaces switchport", "po": "show etherchannel summary",
               "run_intf": "show running-config | section ^interface"},
    "nxos":   {"brief": "show ip interface brief", "status": "show interface status",
               "switch": "show interface switchport", "po": "show port-channel summary",
               "run_intf": "show running-config interface"},
    "eos":    {"brief": "show ip interface brief", "status": "show interfaces status",
               "switch": "show interfaces switchport", "po": "show port-channel summary",
               "run_intf": "show running-config | section interface"},
    "iosxr":  {"brief": "show ipv4 interface brief", "status": "show interfaces brief",
               "switch": "", "po": "show bundle brief",
               "run_intf": "show running-config interface"},
    "junos":  {"brief": "show interfaces terse", "status": "show interfaces descriptions",
               "switch": "show ethernet-switching interface", "po": "show interfaces ae0 terse",
               "run_intf": "show configuration interfaces | display set"},
    "fortios": {"brief": "get system interface physical", "status": "get system interface",
                "switch": "", "po": "", "run_intf": "show system interface"},
    "panos":  {"brief": "show interface all", "status": "show interface all",
               "switch": "", "po": "", "run_intf": "show config running | match interface"},
}

# Interfaces that are structurally unsafe to bulk-edit.
UPLINK_HINTS = re.compile(
    r"(uplink|trunk|core|spine|wan|mpls|peer|mlag|vpc|port-?channel|bundle|ae\d|po\d)",
    re.IGNORECASE,
)


# --------------------------------------------------------------------------
# Model
# --------------------------------------------------------------------------

@dataclass
class Interface:
    name: str
    description: str = ""
    status: str = ""          # connected | notconnect | disabled | up | down
    vlan: str = ""
    mode: str = ""            # access | trunk | routed
    in_portchannel: bool = False
    current_cfg: list[str] = field(default_factory=list)

    @property
    def is_uplink(self) -> bool:
        return (
            self.in_portchannel
            or self.mode == "trunk"
            or bool(UPLINK_HINTS.search(self.name))
            or bool(UPLINK_HINTS.search(self.description))
        )


@dataclass
class DeviceResult:
    host: str
    os: str
    status: str = "pending"       # ok | dry_run | failed | skipped | aborted
    discovered: int = 0
    selected: list[str] = field(default_factory=list)
    excluded: dict[str, str] = field(default_factory=dict)
    verified: int = 0
    message: str = ""
    undo_file: str = ""
    elapsed_s: float = 0.0


# --------------------------------------------------------------------------
# Discovery and parsing
# --------------------------------------------------------------------------

def parse_status_table(raw: str, os_key: str) -> dict[str, Interface]:
    """Parse `show interfaces status` style output into Interface objects."""
    out: dict[str, Interface] = {}

    if os_key in ("ios", "iosxe", "nxos", "eos"):
        for ln in raw.splitlines():
            if not ln.strip() or ln.lstrip().startswith(("Port", "---", "Name")):
                continue
            # Name  Description(may contain spaces)  Status  Vlan  Duplex Speed Type
            m = re.match(
                r"^(\S+)\s+(.*?)\s+(connected|notconnect|disabled|err-disabled|monitoring|inactive)"
                r"\s+(\S+)\s+", ln)
            if not m:
                continue
            name, desc, status, vlan = m.groups()
            out[name] = Interface(name=name, description=desc.strip(),
                                  status=status, vlan=vlan)
        return out

    if os_key == "junos":
        # show interfaces descriptions:  Interface  Admin  Link  Description
        for ln in raw.splitlines():
            m = re.match(r"^(\S+)\s+(up|down)\s+(up|down)\s*(.*)$", ln)
            if m:
                name, admin, link, desc = m.groups()
                out[name] = Interface(name=name, description=desc.strip(),
                                      status="connected" if link == "up" else "notconnect")
        return out

    if os_key == "iosxr":
        for ln in raw.splitlines():
            m = re.match(r"^(\S+)\s+(up|down|admin-down)\s+(up|down|admin-down)", ln)
            if m:
                name, admin, _ = m.groups()
                out[name] = Interface(name=name,
                                      status="connected" if admin == "up" else "notconnect")
        return out

    if os_key == "fortios":
        cur = None
        for ln in raw.splitlines():
            m = re.match(r"^==\s*\[\s*(\S+)\s*\]", ln)
            if m:
                cur = m.group(1)
                out[cur] = Interface(name=cur)
            elif cur and "status:" in ln:
                out[cur].status = "connected" if "up" in ln else "notconnect"
        return out

    return out


def enrich_switchport(ifaces: dict[str, Interface], raw: str, os_key: str) -> None:
    """Fill mode and access VLAN from `show interfaces switchport`."""
    if os_key not in ("ios", "iosxe", "nxos", "eos") or not raw:
        return
    cur = None
    for ln in raw.splitlines():
        m = re.match(r"^Name:\s*(\S+)", ln)
        if m:
            cur = m.group(1)
            continue
        if not cur or cur not in ifaces:
            continue
        m = re.search(r"Operational Mode:\s*(\S+)", ln)
        if m:
            mode = m.group(1).lower()
            ifaces[cur].mode = "trunk" if "trunk" in mode else (
                "access" if "access" in mode else "routed")
        m = re.search(r"Access Mode VLAN:\s*(\d+)", ln)
        if m:
            ifaces[cur].vlan = m.group(1)


def enrich_portchannel(ifaces: dict[str, Interface], raw: str, os_key: str) -> None:
    if not raw:
        return
    for i in ifaces.values():
        short = i.name.replace("GigabitEthernet", "Gi").replace("TenGigabitEthernet", "Te")
        if re.search(rf"\b{re.escape(short)}\b", raw) or re.search(rf"\b{re.escape(i.name)}\b", raw):
            i.in_portchannel = True


def capture_current(ifaces: dict[str, Interface], raw: str, os_key: str) -> None:
    """Record each interface's existing config block, used to build the undo file."""
    if not raw:
        return
    cur = None
    for ln in raw.splitlines():
        m = re.match(r"^interface\s+(\S+)", ln, re.IGNORECASE)
        if m:
            cur = m.group(1)
            continue
        if ln.strip() in ("!", "exit"):
            cur = None
            continue
        if cur and cur in ifaces and ln.startswith((" ", "\t")):
            ifaces[cur].current_cfg.append(ln.rstrip())


def discover(conn, os_key: str) -> dict[str, Interface]:
    cmds = DISCOVER.get(os_key, {})
    status_raw = conn.send_command(cmds["status"], read_timeout=90) if cmds.get("status") else ""
    ifaces = parse_status_table(status_raw, os_key)

    if cmds.get("switch"):
        enrich_switchport(ifaces, conn.send_command(cmds["switch"], read_timeout=120), os_key)
    if cmds.get("po"):
        enrich_portchannel(ifaces, conn.send_command(cmds["po"], read_timeout=60), os_key)
    if cmds.get("run_intf"):
        capture_current(ifaces, conn.send_command(cmds["run_intf"], read_timeout=180), os_key)
    return ifaces


# --------------------------------------------------------------------------
# Selection
# --------------------------------------------------------------------------

def select(ifaces: dict[str, Interface], args) -> tuple[list[Interface], dict[str, str]]:
    chosen: list[Interface] = []
    rejected: dict[str, str] = {}

    for name, i in sorted(ifaces.items()):
        if args.match_name and not re.search(args.match_name, name):
            continue
        if args.exclude_name and re.search(args.exclude_name, name):
            rejected[name] = "exclude-name"
            continue
        if args.match_desc and not re.search(args.match_desc, i.description, re.IGNORECASE):
            continue
        if args.exclude_desc and re.search(args.exclude_desc, i.description, re.IGNORECASE):
            rejected[name] = "exclude-desc"
            continue
        if args.match_vlan and i.vlan != str(args.match_vlan):
            continue
        if args.match_status:
            want = args.match_status
            if want == "up" and i.status != "connected":
                continue
            if want == "down" and i.status != "notconnect":
                continue
            if want == "admin-down" and i.status not in ("disabled", "admin-down"):
                continue
        if i.is_uplink and not args.include_uplinks:
            rejected[name] = "uplink/trunk/po-member"
            continue
        chosen.append(i)

    return chosen, rejected


# --------------------------------------------------------------------------
# Profile rendering
# --------------------------------------------------------------------------

def render_profile(profile: dict, os_key: str, iface: Interface,
                   overrides: dict[str, Any]) -> list[str]:
    body = profile.get("config", {}).get(os_key)
    if not body:
        raise ValueError(f"profile '{profile.get('name')}' has no body for os '{os_key}'")

    ctx = dict(profile.get("vars", {}))
    ctx.update(overrides)
    ctx.setdefault("interface", iface.name)
    ctx.setdefault("description", iface.description)

    lines: list[str] = []
    for raw in body:
        try:
            lines.append(raw.format(**ctx))
        except KeyError as exc:
            raise ValueError(
                f"profile line '{raw}' references undefined variable {exc}") from exc
    return lines


def build_commands(os_key: str, iface: Interface, body: list[str]) -> list[str]:
    if os_key in ("ios", "iosxe", "nxos", "eos", "iosxr"):
        return [f"interface {iface.name}", *body]
    if os_key == "junos":
        return [b if b.startswith(("set ", "delete ")) else
                f"set interfaces {iface.name} {b}" for b in body]
    if os_key == "fortios":
        return ["config system interface", f"    edit {iface.name}", *[f"    {b}" for b in body],
                "    next", "end"]
    if os_key == "panos":
        return [b if b.startswith("set ") else f"set network interface {iface.name} {b}"
                for b in body]
    raise ValueError(f"unsupported os '{os_key}'")


def build_undo(os_key: str, iface: Interface) -> list[str]:
    """Reconstruct the interface's prior config so the change can be reversed."""
    if os_key in ("ios", "iosxe", "nxos", "eos", "iosxr"):
        if not iface.current_cfg:
            return [f"! {iface.name}: prior config not captured, restore from snapshot"]
        return [f"default interface {iface.name}", f"interface {iface.name}",
                *iface.current_cfg]
    if os_key == "junos":
        return [f"! {iface.name}: use 'rollback 1' on the device"]
    return [f"! {iface.name}: restore from the pre-change snapshot"]


# --------------------------------------------------------------------------
# Verification
# --------------------------------------------------------------------------

def verify(conn, os_key: str, selected: list[Interface], profile: dict) -> tuple[int, list[str]]:
    """Confirm each port reached the state the profile intended."""
    expect = profile.get("verify", {})
    if not expect:
        return len(selected), []

    cmd = expect.get("command")
    pattern = expect.get("expect")
    if not cmd or not pattern:
        return len(selected), []

    ok, bad = 0, []
    for i in selected:
        out = conn.send_command(cmd.format(interface=i.name), read_timeout=60)
        if re.search(pattern.format(interface=i.name), out, re.IGNORECASE):
            ok += 1
        else:
            bad.append(i.name)
    return ok, bad


# --------------------------------------------------------------------------
# Device workflow
# --------------------------------------------------------------------------

def connect(dev: dict):
    params = {
        "device_type": NETMIKO_DRIVER[dev["os"]],
        "host": dev["host"],
        "username": os.environ.get("NET_USER", ""),
        "password": os.environ.get("NET_PASS", ""),
        "port": dev.get("port", 22),
        "conn_timeout": 20,
        "fast_cli": False,
    }
    if os.environ.get("NET_ENABLE"):
        params["secret"] = os.environ["NET_ENABLE"]
    conn = ConnectHandler(**params)
    if params.get("secret") and dev["os"] in ("ios", "iosxe", "nxos"):
        conn.enable()
    return conn


def process(dev: dict, profile: dict, args, stamp: str) -> DeviceResult:
    res = DeviceResult(host=dev["host"], os=dev["os"])
    t0 = time.time()
    conn = None
    try:
        conn = connect(dev)
        ifaces = discover(conn, dev["os"])
        res.discovered = len(ifaces)

        chosen, rejected = select(ifaces, args)
        res.selected = [i.name for i in chosen]
        res.excluded = rejected

        if not chosen:
            res.status = "skipped"
            res.message = f"no interface matched (of {len(ifaces)} discovered)"
            return res

        if len(chosen) > args.max_change:
            res.status = "aborted"
            res.message = (f"selection of {len(chosen)} exceeds --max-change {args.max_change}; "
                           f"narrow the filters or raise the cap deliberately")
            return res

        # Build the full command set and the matching undo fragment
        cmds: list[str] = []
        undo: list[str] = []
        overrides = dev.get("vars", {})
        for i in chosen:
            body = render_profile(profile, dev["os"], i, overrides)
            cmds.extend(build_commands(dev["os"], i, body))
            undo.extend(build_undo(dev["os"], i))

        if args.undo_dir:
            d = Path(args.undo_dir)
            d.mkdir(parents=True, exist_ok=True)
            p = d / f"{dev['host']}-{stamp}.cfg"
            p.write_text("\n".join([
                f"! Undo for {dev['host']} ({dev['os']}) generated {stamp}",
                f"! Reverts profile '{profile.get('name')}' on {len(chosen)} interface(s)",
                "!", *undo, ""]), encoding="utf-8")
            res.undo_file = str(p)

        if not args.apply:
            res.status = "dry_run"
            res.message = f"{len(chosen)} interface(s) would change, {len(rejected)} excluded"
            return res

        out = conn.send_config_set(cmds, read_timeout=240)
        for marker in ("% Invalid", "% Incomplete", "% Ambiguous", "ERROR:", "Invalid input"):
            if marker in out:
                res.status = "failed"
                res.message = f"device rejected config ({marker}); undo at {res.undo_file}"
                return res

        time.sleep(args.settle_seconds)
        ok, bad = verify(conn, dev["os"], chosen, profile)
        res.verified = ok

        if bad:
            res.status = "failed"
            res.message = (f"{len(bad)} interface(s) failed verification: "
                           f"{', '.join(bad[:6])}{'...' if len(bad) > 6 else ''}")
            return res

        if dev["os"] in ("ios", "iosxe", "eos"):
            conn.send_command_timing("write memory")
        elif dev["os"] == "nxos":
            conn.send_command_timing("copy running-config startup-config")

        res.status = "ok"
        res.message = f"{len(chosen)} interface(s) configured and verified"

    except NetmikoAuthenticationException as exc:
        res.status, res.message = "failed", f"authentication failed: {exc}"
    except NetmikoTimeoutException as exc:
        res.status, res.message = "failed", f"timeout: {exc} - VERIFY DEVICE STATE MANUALLY"
    except ValueError as exc:
        res.status, res.message = "failed", f"profile error: {exc}"
    except Exception as exc:  # noqa: BLE001
        res.status, res.message = "failed", f"{type(exc).__name__}: {exc}"
    finally:
        res.elapsed_s = round(time.time() - t0, 1)
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass
    return res


# --------------------------------------------------------------------------
# Reporting
# --------------------------------------------------------------------------

def render(results: list[DeviceResult], args) -> int:
    icon = {"ok": "[OK]", "dry_run": "[DRY]", "failed": "[FAIL]",
            "skipped": "[SKIP]", "aborted": "[ABORT]"}
    total_if = 0

    for r in sorted(results, key=lambda x: x.host):
        total_if += len(r.selected)
        print(f"\n{icon.get(r.status, '[?]'):<8} {r.host:<20} {r.os:<7} "
              f"{len(r.selected):>3}/{r.discovered:<3} intf  {r.elapsed_s:>5.1f}s  {r.message}")
        if r.selected and args.show_interfaces:
            for chunk in (r.selected[i:i + 6] for i in range(0, len(r.selected), 6)):
                print(f"           {'  '.join(c.ljust(14) for c in chunk)}")
        if r.excluded and args.show_excluded:
            for name, why in list(r.excluded.items())[:10]:
                print(f"           excluded {name:<16} ({why})")
        if r.undo_file:
            print(f"           undo: {r.undo_file}")

    tally: dict[str, int] = {}
    for r in results:
        tally[r.status] = tally.get(r.status, 0) + 1
    print(f"\n{'=' * 78}")
    print(f"  {len(results)} device(s)   {total_if} interface(s) in scope")
    print("  " + "   ".join(f"{k}={v}" for k, v in sorted(tally.items())))

    if tally.get("aborted"):
        print("\n  Aborted devices exceeded --max-change. Review the selection filters.")
    bad = tally.get("failed", 0) + tally.get("aborted", 0)
    if not args.apply and tally.get("dry_run"):
        print("\n  Dry-run. Re-run with --apply to write these changes.")
    return 1 if bad else 0


def main() -> int:
    ap = argparse.ArgumentParser(description="Apply an interface profile in bulk, safely.")
    ap.add_argument("--inventory", required=True)
    ap.add_argument("--profile", required=True, help="YAML file of named interface profiles")
    ap.add_argument("--profile-name", required=True, help="Which profile to apply")

    sel = ap.add_argument_group("selection")
    sel.add_argument("--match-name", help="Regex the interface name must match")
    sel.add_argument("--match-desc", help="Regex the description must match")
    sel.add_argument("--match-vlan", help="Only ports in this access VLAN")
    sel.add_argument("--match-status", choices=["up", "down", "admin-down"])
    sel.add_argument("--exclude-name", help="Regex of names to skip")
    sel.add_argument("--exclude-desc", help="Regex of descriptions to skip")
    sel.add_argument("--include-uplinks", action="store_true",
                     help="Allow trunks, port-channel members and uplink-named ports")

    safe = ap.add_argument_group("safety")
    safe.add_argument("--apply", action="store_true", help="Write changes (default is dry-run)")
    safe.add_argument("--max-change", type=int, default=24,
                      help="Abort a device if more interfaces than this are selected. Default 24")
    safe.add_argument("--settle-seconds", type=int, default=5)
    safe.add_argument("--undo-dir", default="./undo", help="Where undo fragments are written")

    ap.add_argument("--workers", type=int, default=6)
    ap.add_argument("--show-interfaces", action="store_true")
    ap.add_argument("--show-excluded", action="store_true")
    ap.add_argument("--json-out")
    ap.add_argument("-v", "--verbose", action="store_true")
    args = ap.parse_args()

    logging.basicConfig(
        level=logging.DEBUG if args.verbose else logging.INFO,
        format="%(asctime)s %(levelname)-7s %(message)s", datefmt="%H:%M:%S")

    if not any([args.match_name, args.match_desc, args.match_vlan, args.match_status]):
        LOG.error("At least one --match-* filter is required. "
                  "Refusing to select every interface on every device.")
        return 2
    if not os.environ.get("NET_USER") or not os.environ.get("NET_PASS"):
        LOG.error("Set NET_USER and NET_PASS in the environment.")
        return 2

    raw = Path(args.inventory).read_text(encoding="utf-8")
    data = (yaml.safe_load(raw) if yaml and args.inventory.endswith((".yaml", ".yml"))
            else json.loads(raw))
    devices = data["devices"] if isinstance(data, dict) else data

    praw = Path(args.profile).read_text(encoding="utf-8")
    pdata = yaml.safe_load(praw) if yaml else json.loads(praw)
    profiles = {p["name"]: p for p in (pdata["profiles"] if isinstance(pdata, dict) else pdata)}
    if args.profile_name not in profiles:
        LOG.error("Profile '%s' not found. Available: %s",
                  args.profile_name, ", ".join(sorted(profiles)))
        return 2
    profile = profiles[args.profile_name]

    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    LOG.info("%s | profile '%s' | %d device(s) | cap %d intf/device",
             "APPLY" if args.apply else "DRY-RUN", args.profile_name,
             len(devices), args.max_change)
    if args.apply:
        LOG.warning("Live change. Undo fragments will be written to %s", args.undo_dir)

    results: list[DeviceResult] = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futs = {pool.submit(process, d, profile, args, stamp): d for d in devices}
        for fut in as_completed(futs):
            d = futs[fut]
            try:
                results.append(fut.result())
            except Exception as exc:  # noqa: BLE001
                LOG.exception("%s: worker crashed", d["host"])
                results.append(DeviceResult(host=d["host"], os=d.get("os", "?"),
                                            status="failed", message=f"crashed: {exc}"))

    if args.json_out:
        Path(args.json_out).write_text(
            json.dumps([asdict(r) for r in results], indent=2), encoding="utf-8")

    return render(results, args)


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted. Devices mid-write may be partially configured - verify.",
              file=sys.stderr)
        sys.exit(130)