Bulk VLAN Deployment with Consistency Verification

#!/usr/bin/env python3
"""
Bulk VLAN Deployment with Consistency Verification

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

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 collections import defaultdict
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("bulkvlan")

NETMIKO_DRIVER = {
    "ios": "cisco_ios", "iosxe": "cisco_ios", "nxos": "cisco_nxos",
    "eos": "arista_eos", "junos": "juniper_junos", "iosxr": "cisco_xr",
    "fortios": "fortinet", "panos": "paloalto_panos",
}

SHOW_VLAN = {
    "ios": "show vlan brief", "iosxe": "show vlan brief",
    "nxos": "show vlan brief", "eos": "show vlan",
    "junos": "show vlans", "fortios": "show system interface",
}

SHOW_TRUNK = {
    "ios": "show interfaces trunk", "iosxe": "show interfaces trunk",
    "nxos": "show interface trunk", "eos": "show interfaces trunk",
    "junos": "show ethernet-switching interface",
}

# Ranges that must never be touched without an explicit override.
RESERVED = [(0, 0), (1, 1), (1002, 1005), (3968, 4047), (4094, 4095)]


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

@dataclass
class VlanSpec:
    id: int
    name: str
    svi: str = ""              # e.g. "10.20.30.2/24"
    gateway: str = ""          # HSRP/VRRP virtual IP, or anycast address
    gateway_group: int = 1
    vrf: str = ""
    mtu: int = 0
    description: str = ""
    trunks: list[str] = field(default_factory=list)   # [] means all trunks


@dataclass
class DeviceAudit:
    host: str
    os: str
    present: dict[int, str] = field(default_factory=dict)      # vlan id -> name
    trunk_allowed: dict[str, set[int]] = field(default_factory=dict)
    error: str = ""


@dataclass
class DeviceResult:
    host: str
    os: str
    status: str = "pending"      # ok | dry_run | failed | skipped | audit
    created: list[int] = field(default_factory=list)
    renamed: list[int] = field(default_factory=list)
    trunks_updated: list[str] = field(default_factory=list)
    already_present: list[int] = field(default_factory=list)
    message: str = ""
    undo_file: str = ""
    elapsed_s: float = 0.0


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

def parse_vlans(raw: str, os_key: str) -> dict[int, str]:
    out: dict[int, str] = {}
    if os_key in ("ios", "iosxe", "nxos", "eos"):
        for ln in raw.splitlines():
            m = re.match(r"^(\d{1,4})\s+(\S+)\s+(active|suspended|act/lshut)", ln)
            if m:
                out[int(m.group(1))] = m.group(2)
        return out
    if os_key == "junos":
        cur_name = ""
        for ln in raw.splitlines():
            m = re.match(r"^(\S+)\s+(\d{1,4})\s", ln)
            if m:
                out[int(m.group(2))] = m.group(1)
                continue
            m = re.match(r"^\s*(\S+)\s*$", ln)
            if m and not m.group(1).isdigit():
                cur_name = m.group(1)
            m = re.search(r"VLAN ID:\s*(\d+)", ln)
            if m and cur_name:
                out[int(m.group(1))] = cur_name
        return out
    return out


def parse_trunks(raw: str, os_key: str) -> dict[str, set[int]]:
    """Map trunk interface -> set of allowed VLAN ids."""
    out: dict[str, set[int]] = {}
    if os_key not in ("ios", "iosxe", "nxos", "eos") or not raw:
        return out

    # 'show interfaces trunk' has several sections; the allowed-vlan one is
    # identified by its header. Entries can wrap across lines.
    in_allowed = False
    cur_if = None
    for ln in raw.splitlines():
        if re.search(r"Vlans allowed on trunk", ln, re.IGNORECASE):
            in_allowed = True
            continue
        if re.search(r"Vlans allowed and active|Vlans in spanning tree", ln, re.IGNORECASE):
            in_allowed = False
            continue
        if not in_allowed:
            continue

        m = re.match(r"^(\S+)\s+([\d,\-\s]+)$", ln.strip())
        if m:
            cur_if = m.group(1)
            out.setdefault(cur_if, set()).update(expand_ranges(m.group(2)))
        elif cur_if and re.match(r"^[\d,\-\s]+$", ln.strip()) and ln.strip():
            out[cur_if].update(expand_ranges(ln.strip()))
    return out


def expand_ranges(text: str) -> set[int]:
    """'10,20-22,30' -> {10,20,21,22,30}"""
    out: set[int] = set()
    for part in text.replace(" ", "").split(","):
        if not part:
            continue
        if "-" in part:
            try:
                a, b = part.split("-", 1)
                out.update(range(int(a), int(b) + 1))
            except ValueError:
                continue
        elif part.isdigit():
            out.add(int(part))
    return out


# --------------------------------------------------------------------------
# Command construction
# --------------------------------------------------------------------------

def is_reserved(vid: int) -> bool:
    return any(lo <= vid <= hi for lo, hi in RESERVED)


def build_vlan_cmds(os_key: str, spec: VlanSpec, create: bool, rename: bool) -> list[str]:
    cmds: list[str] = []

    if os_key in ("ios", "iosxe", "nxos", "eos"):
        if create or rename:
            cmds += [f"vlan {spec.id}", f" name {spec.name}"]
        if spec.svi:
            cmds.append(f"interface Vlan{spec.id}")
            if spec.description:
                cmds.append(f" description {spec.description}")
            if spec.vrf:
                cmds.append(f" vrf {'member' if os_key == 'nxos' else 'forwarding'} {spec.vrf}")
            cmds.append(f" ip address {cidr_to_ios(spec.svi)}")
            if spec.mtu:
                cmds.append(f" mtu {spec.mtu}")
            if spec.gateway:
                if os_key == "nxos":
                    cmds += [f" hsrp {spec.gateway_group}", f"  ip {spec.gateway}",
                             "  priority 110", "  preempt"]
                elif os_key == "eos":
                    cmds.append(f" ip virtual-router address {spec.gateway}")
                else:
                    cmds += [f" standby {spec.gateway_group} ip {spec.gateway}",
                             f" standby {spec.gateway_group} priority 110",
                             f" standby {spec.gateway_group} preempt"]
            cmds.append(" no shutdown")
        return cmds

    if os_key == "junos":
        if create or rename:
            cmds.append(f"set vlans {spec.name} vlan-id {spec.id}")
        if spec.svi:
            cmds += [
                f"set vlans {spec.name} l3-interface irb.{spec.id}",
                f"set interfaces irb unit {spec.id} family inet address {spec.svi}",
            ]
            if spec.description:
                cmds.append(f"set interfaces irb unit {spec.id} description \"{spec.description}\"")
            if spec.vrf:
                cmds.append(f"set routing-instances {spec.vrf} interface irb.{spec.id}")
            if spec.gateway:
                cmds.append(
                    f"set interfaces irb unit {spec.id} family inet address {spec.svi} "
                    f"vrrp-group {spec.gateway_group} virtual-address {spec.gateway}")
        return cmds

    raise ValueError(f"VLAN creation not supported on os '{os_key}'")


def cidr_to_ios(cidr: str) -> str:
    """'10.20.30.2/24' -> '10.20.30.2 255.255.255.0'"""
    import ipaddress
    iface = ipaddress.ip_interface(cidr)
    return f"{iface.ip} {iface.network.netmask}"


def build_trunk_cmds(os_key: str, trunk: str, vids: list[int]) -> list[str]:
    """Always ADD to the allowed list; never replace it."""
    vid_str = ",".join(str(v) for v in sorted(vids))
    if os_key in ("ios", "iosxe", "eos"):
        return [f"interface {trunk}", f" switchport trunk allowed vlan add {vid_str}"]
    if os_key == "nxos":
        return [f"interface {trunk}", f" switchport trunk allowed vlan add {vid_str}"]
    if os_key == "junos":
        return [f"set interfaces {trunk} unit 0 family ethernet-switching "
                f"vlan members {v}" for v in sorted(vids)]
    raise ValueError(f"trunk update not supported on os '{os_key}'")


def build_undo(os_key: str, created: list[int], trunks: dict[str, list[int]],
               specs: dict[int, VlanSpec]) -> list[str]:
    out: list[str] = []
    for trunk, vids in trunks.items():
        vid_str = ",".join(str(v) for v in sorted(vids))
        if os_key in ("ios", "iosxe", "nxos", "eos"):
            out += [f"interface {trunk}", f" switchport trunk allowed vlan remove {vid_str}"]
        elif os_key == "junos":
            out += [f"delete interfaces {trunk} unit 0 family ethernet-switching "
                    f"vlan members {v}" for v in sorted(vids)]
    for vid in created:
        if os_key in ("ios", "iosxe", "nxos", "eos"):
            out += [f"no interface Vlan{vid}", f"no vlan {vid}"]
        elif os_key == "junos":
            out.append(f"delete vlans {specs[vid].name}")
    return out


# --------------------------------------------------------------------------
# Audit
# --------------------------------------------------------------------------

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 audit(dev: dict) -> DeviceAudit:
    a = DeviceAudit(host=dev["host"], os=dev["os"])
    conn = None
    try:
        conn = connect(dev)
        if SHOW_VLAN.get(dev["os"]):
            a.present = parse_vlans(conn.send_command(SHOW_VLAN[dev["os"]], read_timeout=90),
                                    dev["os"])
        if SHOW_TRUNK.get(dev["os"]):
            a.trunk_allowed = parse_trunks(conn.send_command(SHOW_TRUNK[dev["os"]],
                                                             read_timeout=90), dev["os"])
    except NetmikoAuthenticationException as exc:
        a.error = f"authentication failed: {exc}"
    except NetmikoTimeoutException as exc:
        a.error = f"timeout: {exc}"
    except Exception as exc:  # noqa: BLE001
        a.error = f"{type(exc).__name__}: {exc}"
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass
    return a


def domain_report(audits: list[DeviceAudit], specs: list[VlanSpec]) -> list[str]:
    """Describe inconsistency across the domain, independent of the intended change."""
    issues: list[str] = []
    reachable = [a for a in audits if not a.error]
    if not reachable:
        return ["no device could be audited"]

    # Name mismatches for the same VLAN id
    names: dict[int, dict[str, list[str]]] = defaultdict(lambda: defaultdict(list))
    for a in reachable:
        for vid, name in a.present.items():
            names[vid][name].append(a.host)
    for vid, by_name in sorted(names.items()):
        if len(by_name) > 1:
            detail = "; ".join(f"'{n}' on {len(h)} ({', '.join(h[:3])})"
                               for n, h in by_name.items())
            issues.append(f"VLAN {vid} has inconsistent names: {detail}")

    # Partial presence of the VLANs we care about
    wanted = {s.id for s in specs}
    for vid in sorted(wanted):
        have = [a.host for a in reachable if vid in a.present]
        if have and len(have) != len(reachable):
            missing = [a.host for a in reachable if vid not in a.present]
            issues.append(f"VLAN {vid} present on {len(have)}/{len(reachable)} switches, "
                          f"missing on: {', '.join(missing[:5])}")

    # Trunks that do not carry a VLAN the switch has
    for a in reachable:
        for vid in sorted(set(a.present) & wanted):
            gaps = [t for t, allowed in a.trunk_allowed.items() if vid not in allowed]
            if gaps:
                issues.append(f"{a.host}: VLAN {vid} exists but is not allowed on "
                              f"{len(gaps)} trunk(s): {', '.join(gaps[:3])}")
    return issues


# --------------------------------------------------------------------------
# Push
# --------------------------------------------------------------------------

def process(dev: dict, specs: list[VlanSpec], pre: DeviceAudit,
            args, stamp: str) -> DeviceResult:
    res = DeviceResult(host=dev["host"], os=dev["os"])
    t0 = time.time()
    conn = None
    try:
        if pre.error:
            res.status, res.message = "failed", pre.error
            return res

        cmds: list[str] = []
        trunk_plan: dict[str, list[int]] = defaultdict(list)
        spec_by_id = {s.id: s for s in specs}

        for s in specs:
            if is_reserved(s.id) and not args.allow_reserved:
                res.message = f"VLAN {s.id} is reserved; use --allow-reserved to override"
                res.status = "failed"
                return res

            existing = pre.present.get(s.id)
            if existing is None:
                cmds += build_vlan_cmds(dev["os"], s, create=True, rename=False)
                res.created.append(s.id)
            elif existing != s.name:
                if not args.allow_rename:
                    res.already_present.append(s.id)
                    LOG.warning("%s: VLAN %d exists as '%s', wanted '%s' "
                                "(use --allow-rename to change)",
                                dev["host"], s.id, existing, s.name)
                    continue
                cmds += build_vlan_cmds(dev["os"], s, create=False, rename=True)
                res.renamed.append(s.id)
            else:
                res.already_present.append(s.id)

            # Trunk allowed-list extension
            targets = s.trunks or list(pre.trunk_allowed.keys())
            for t in targets:
                if s.id not in pre.trunk_allowed.get(t, set()):
                    trunk_plan[t].append(s.id)

        for trunk, vids in trunk_plan.items():
            cmds += build_trunk_cmds(dev["os"], trunk, vids)
            res.trunks_updated.append(trunk)

        if not cmds:
            res.status = "skipped"
            res.message = "already in the desired state"
            return res

        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"! Removes {len(res.created)} VLAN(s) and reverts "
                f"{len(res.trunks_updated)} trunk(s)",
                "!", *build_undo(dev["os"], res.created, trunk_plan, spec_by_id), ""]),
                encoding="utf-8")
            res.undo_file = str(p)

        if not args.apply:
            res.status = "dry_run"
            res.message = (f"would create {len(res.created)}, rename {len(res.renamed)}, "
                           f"update {len(res.trunks_updated)} trunk(s)")
            return res

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

        # Verify on-box before claiming success
        time.sleep(args.settle_seconds)
        after = parse_vlans(conn.send_command(SHOW_VLAN[dev["os"]], read_timeout=90), dev["os"])
        missing = [s.id for s in specs if s.id not in after]
        if missing:
            res.status = "failed"
            res.message = f"VLANs still missing after push: {missing}; undo at {res.undo_file}"
            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"created {len(res.created)}, renamed {len(res.renamed)}, "
                       f"{len(res.trunks_updated)} trunk(s) updated")

    except ValueError as exc:
        res.status, res.message = "failed", str(exc)
    except NetmikoTimeoutException as exc:
        res.status, res.message = "failed", f"timeout: {exc} - VERIFY DEVICE STATE"
    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


# --------------------------------------------------------------------------
# CLI
# --------------------------------------------------------------------------

def load_specs(path: str) -> list[VlanSpec]:
    raw = Path(path).read_text(encoding="utf-8")
    data = yaml.safe_load(raw) if (yaml and path.endswith((".yaml", ".yml"))) else json.loads(raw)
    items = data["vlans"] if isinstance(data, dict) else data
    return [VlanSpec(**v) for v in items]


def render(results: list[DeviceResult], issues_before: list[str],
           issues_after: list[str], args) -> int:
    icon = {"ok": "[OK]", "dry_run": "[DRY]", "failed": "[FAIL]",
            "skipped": "[SKIP]", "audit": "[AUDIT]"}

    if issues_before:
        print(f"\n{'=' * 78}\nDOMAIN INCONSISTENCY BEFORE CHANGE ({len(issues_before)})")
        for i in issues_before[:20]:
            print(f"  ! {i}")
        if len(issues_before) > 20:
            print(f"  ... {len(issues_before) - 20} more")

    if args.audit_only:
        print(f"\n{'=' * 78}\n  Audit only. No change attempted.")
        return 1 if issues_before else 0

    print(f"\n{'=' * 78}")
    for r in sorted(results, key=lambda x: x.host):
        print(f"{icon.get(r.status, '[?]'):<8} {r.host:<20} {r.os:<7} "
              f"{r.elapsed_s:>5.1f}s  {r.message}")
        if r.created:
            print(f"           created: {', '.join(str(v) for v in r.created)}")
        if r.trunks_updated:
            print(f"           trunks:  {', '.join(r.trunks_updated[:6])}")
        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("  " + "   ".join(f"{k}={v}" for k, v in sorted(tally.items())))

    if issues_after:
        print(f"\n  DOMAIN STILL INCONSISTENT ({len(issues_after)} issue(s)):")
        for i in issues_after[:10]:
            print(f"    ! {i}")
        return 1
    if args.apply:
        print("\n  Domain consistent across all reachable switches.")
    else:
        print("\n  Dry-run. Re-run with --apply to write.")
    return 1 if tally.get("failed") else 0


def main() -> int:
    ap = argparse.ArgumentParser(description="Deploy VLANs across a switching domain.")
    ap.add_argument("--inventory", required=True)
    ap.add_argument("--vlans", required=True, help="YAML/JSON list of VLAN specs")
    ap.add_argument("--apply", action="store_true", help="Write changes (default dry-run)")
    ap.add_argument("--audit-only", action="store_true",
                    help="Only report current domain consistency; never build a change")
    ap.add_argument("--allow-rename", action="store_true",
                    help="Rename a VLAN that exists with a different name")
    ap.add_argument("--allow-reserved", action="store_true",
                    help="Permit reserved VLAN ids (1, 1002-1005, 3968-4047, 4094)")
    ap.add_argument("--undo-dir", default="./undo")
    ap.add_argument("--settle-seconds", type=int, default=5)
    ap.add_argument("--workers", type=int, default=8)
    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 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
    specs = load_specs(args.vlans)

    stamp = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
    LOG.info("%s | %d VLAN(s) | %d switch(es)",
             "AUDIT" if args.audit_only else ("APPLY" if args.apply else "DRY-RUN"),
             len(specs), len(devices))

    LOG.info("Auditing current VLAN state ...")
    audits: dict[str, DeviceAudit] = {}
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futs = {pool.submit(audit, d): d for d in devices}
        for fut in as_completed(futs):
            d = futs[fut]
            try:
                a = fut.result()
            except Exception as exc:  # noqa: BLE001
                a = DeviceAudit(host=d["host"], os=d.get("os", "?"), error=str(exc))
            audits[d["host"]] = a

    issues_before = domain_report(list(audits.values()), specs)

    results: list[DeviceResult] = []
    if not args.audit_only:
        with ThreadPoolExecutor(max_workers=args.workers) as pool:
            futs = {pool.submit(process, d, specs, audits[d["host"]], 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}"))

    issues_after: list[str] = []
    if args.apply and results:
        LOG.info("Re-auditing to confirm domain consistency ...")
        post: list[DeviceAudit] = []
        with ThreadPoolExecutor(max_workers=args.workers) as pool:
            futs = {pool.submit(audit, d): d for d in devices}
            for fut in as_completed(futs):
                try:
                    post.append(fut.result())
                except Exception:  # noqa: BLE001
                    pass
        issues_after = domain_report(post, specs)

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

    return render(results, issues_before, issues_after, args)


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        print("\nInterrupted. Re-run with --audit-only to check domain state.", file=sys.stderr)
        sys.exit(130)