VLAN Table Auditor

#!/usr/bin/env python3
"""
VLAN Table Auditor

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

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 csv
import re
from collections import defaultdict
from dataclasses import dataclass, asdict

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

COMMAND_MAP = {
    "ios": "show vlan brief",
    "iosxe": "show vlan brief",
    "nxos": "show vlan brief",
    "eos": "show vlan",
    "iosxr": "show interfaces",          # XR has no VLAN database; dot1q subifs
    "junos": "show vlans",
    "fortios": "show system interface",
    "panos": "show interface all",
}

# An SVI with an IP makes the VLAN routed (L3) rather than switched only.
L3_CMD = {
    "ios": "show ip interface brief | include Vlan",
    "iosxe": "show ip interface brief | include Vlan",
    "nxos": "show ip interface brief | include Vlan",
    "eos": "show ip interface brief | include Vlan",
}

# Which ports carry each VLAN, used to find VLANs defined but never used.
PORTS_CMD = {
    "ios": "show vlan brief",
    "iosxe": "show vlan brief",
    "nxos": "show vlan brief",
    "eos": "show vlan",
}

RESERVED = {1002, 1003, 1004, 1005}


@dataclass
class Vlan:
    switch: str
    vlan_id: int
    name: str
    status: str
    layer: str = "L2"
    ports: int = 0


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

def parse_l3_vlans(output: str) -> set[int]:
    out: set[int] = set()
    for line in output.splitlines():
        m = re.match(r"Vlan(\d+)\s+(\S+)", line, re.IGNORECASE)
        if m and m.group(2) not in ("unassigned", "", "--"):
            out.add(int(m.group(1)))
    return out


def parse_vlans(output: str, os_key: str, switch: str, l3: set[int]) -> list[Vlan]:
    rows: list[Vlan] = []

    if os_key in ("ios", "iosxe", "nxos", "eos"):
        # VLAN_ID  Name  Status  Ports...   Ports can wrap onto following lines.
        current: Vlan | None = None
        for line in output.splitlines():
            m = re.match(r"^(\d{1,4})\s+(\S+)\s+(active|act|suspended|unsupported|act/lshut)",
                         line, re.IGNORECASE)
            if m:
                vid = int(m.group(1))
                if vid >= 4094 or vid in RESERVED:
                    current = None
                    continue
                tail = line[m.end():].strip()
                current = Vlan(switch, vid, m.group(2), m.group(3).lower(),
                               "L3" if vid in l3 else "L2",
                               len([p for p in tail.split(",") if p.strip()]))
                rows.append(current)
            elif current and line.startswith((" ", "\t")) and "/" in line:
                # continuation line of the port list
                current.ports += len([p for p in line.split(",") if p.strip()])
        return rows

    if os_key == "iosxr":
        cur_if, cur_state = None, "unknown"
        for line in output.splitlines():
            m = re.match(r"^(\S+\.\d+) is (\S+)", line)
            if m:
                cur_if, cur_state = m.group(1), m.group(2).lower().rstrip(",")
            m = re.search(r"Encapsulation 802\.1Q[^\d]*(\d+)", line)
            if m and cur_if:
                rows.append(Vlan(switch, int(m.group(1)), cur_if, cur_state, "L3", 1))
                cur_if = None
        return rows

    if os_key == "junos":
        for line in output.splitlines():
            m = re.match(r"^(\S+)\s+(\d{1,4})\s", line)
            if m:
                rows.append(Vlan(switch, int(m.group(2)), m.group(1), "active", "L2"))
        return rows

    if os_key == "fortios":
        cur_if, cur_vlan = None, None
        for line in output.splitlines():
            m = re.match(r'\s+edit "([^"]+)"', line)
            if m:
                cur_if, cur_vlan = m.group(1), None
            m = re.search(r"set vlanid (\d+)", line)
            if m:
                cur_vlan = int(m.group(1))
            if cur_if and cur_vlan and re.match(r"\s*next\s*$", line):
                rows.append(Vlan(switch, cur_vlan, cur_if, "active", "L3", 1))
                cur_if, cur_vlan = None, None
        return rows

    if os_key == "panos":
        for line in output.splitlines():
            m = re.match(r"(\S+)\.(\d{1,4})\s+\S+\s+(\S+)", line)
            if m:
                rows.append(Vlan(switch, int(m.group(2)),
                                 f"{m.group(1)}.{m.group(2)}", m.group(3).lower(), "L3", 1))
        return rows

    return rows


# ---------------------------------------------------------------------------
# Per-device collection
# ---------------------------------------------------------------------------

def collect(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    conn = None
    try:
        conn = connect(dev)
        l3: set[int] = set()
        if L3_CMD.get(os_key):
            try:
                l3 = parse_l3_vlans(conn.send_command(L3_CMD[os_key], read_timeout=60))
            except Exception:  # noqa: BLE001 - L3 marking is informational
                l3 = set()

        raw = conn.send_command(COMMAND_MAP[os_key], read_timeout=args.read_timeout)
        vlans = parse_vlans(raw, os_key, host, l3)
        return {"host": host, "os": os_key, "status": "ok", "vlans": vlans}

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


# ---------------------------------------------------------------------------
# Fleet-wide consistency analysis
# ---------------------------------------------------------------------------

def analyse(results: list[dict], args) -> list[str]:
    """
    The real value of a VLAN audit is not the per-switch list - it is what is
    inconsistent ACROSS switches. A VLAN with two different names, or present
    on four of six switches, causes intermittent failures that are painful to
    trace back to their cause.
    """
    issues: list[str] = []
    reachable = [r for r in results if r["status"] == "ok"]
    if len(reachable) < 2:
        return issues

    by_id: dict[int, list[Vlan]] = defaultdict(list)
    for r in reachable:
        for v in r["vlans"]:
            by_id[v.vlan_id].append(v)

    switch_count = len(reachable)

    for vid in sorted(by_id):
        entries = by_id[vid]

        names = defaultdict(list)
        for v in entries:
            names[v.name].append(v.switch)
        if len(names) > 1:
            detail = "; ".join(f"'{n}' on {', '.join(sorted(h)[:3])}"
                               + (f" +{len(h)-3}" if len(h) > 3 else "")
                               for n, h in sorted(names.items()))
            issues.append(f"VLAN {vid}: inconsistent names - {detail}")

        present = {v.switch for v in entries}
        if args.expect_everywhere and len(present) < switch_count:
            missing = sorted({r["host"] for r in reachable} - present)
            issues.append(f"VLAN {vid}: on {len(present)}/{switch_count} switches, "
                          f"missing from {', '.join(missing[:4])}"
                          + (f" +{len(missing)-4}" if len(missing) > 4 else ""))

        layers = {v.layer for v in entries}
        if layers == {"L2", "L3"}:
            l3_on = sorted(v.switch for v in entries if v.layer == "L3")
            issues.append(f"VLAN {vid}: routed (SVI with an IP) on "
                          f"{', '.join(l3_on[:3])} but switched-only elsewhere - "
                          f"confirm this is the intended gateway placement")

        suspended = [v.switch for v in entries if v.status not in ("active", "act")]
        if suspended:
            issues.append(f"VLAN {vid}: suspended on {', '.join(sorted(suspended)[:4])}")

    if args.find_unused:
        for r in reachable:
            if r["os"] not in PORTS_CMD:
                continue
            empty = [v for v in r["vlans"]
                     if v.ports == 0 and v.vlan_id != 1 and v.layer == "L2"]
            if empty:
                ids = ", ".join(str(v.vlan_id) for v in sorted(empty, key=lambda x: x.vlan_id)[:10])
                issues.append(f"{r['host']}: {len(empty)} VLAN(s) defined with no ports "
                              f"assigned - {ids}")
    return issues


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Audit VLAN tables across a switching domain and report what is "
                    "inconsistent between switches.",
        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 switches to audit")
    ap.add_argument("--limit", help="Only audit hosts whose name contains this text")
    ap.add_argument("--expect-everywhere", action="store_true",
                    help="Report any VLAN that is not present on every switch. Correct "
                         "for a single L2 domain, noisy if the inventory spans sites")
    ap.add_argument("--find-unused", action="store_true",
                    help="Report VLANs that exist in the database with no ports assigned")
    ap.add_argument("--show-vlans", action="store_true",
                    help="List every VLAN per switch, not just the summary")
    ap.add_argument("--csv", help="Write the full VLAN table to this CSV file")
    ap.add_argument("--read-timeout", type=int, default=90,
                    help="Seconds to wait per command. Default: 90")
    ap.add_argument("--workers", type=int, default=10,
                    help="How many switches 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"Auditing VLANs on {len(devices)} switch(es)...")
    print("=" * 78)

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

    for r in sorted(results, key=lambda x: x["host"]):
        if r["status"] == "error":
            print(f"[ERROR] {r['host']} ({r['os']})  {r['error']}")
            continue
        v = r["vlans"]
        active = sum(1 for x in v if x.status in ("active", "act"))
        l3 = sum(1 for x in v if x.layer == "L3")
        print(f"[OK]    {r['host']:<24} {len(v):>3} VLANs  "
              f"{active} active, {len(v)-active} suspended  {len(v)-l3} L2, {l3} L3")
        if args.show_vlans:
            for x in sorted(v, key=lambda y: y.vlan_id):
                mark = "ok " if x.status in ("active", "act") else "SUSP"
                ports = f"{x.ports} port(s)" if x.ports else "no ports"
                print(f"           {mark}  {x.vlan_id:>4}  {x.name:<22} {x.layer}  {ports}")

    issues = analyse(results, args)
    print("=" * 78)
    if issues:
        print(f"DOMAIN CONSISTENCY: {len(issues)} issue(s)\n")
        for i in issues:
            print(f"  ! {i}")
    else:
        reachable = sum(1 for r in results if r["status"] == "ok")
        print(f"DOMAIN CONSISTENCY: clean across {reachable} switch(es)")

    if args.csv:
        with open(args.csv, "w", newline="", encoding="utf-8") as f:
            w = csv.DictWriter(f, fieldnames=["switch", "vlan_id", "name", "status",
                                              "layer", "ports"])
            w.writeheader()
            for r in results:
                for v in sorted(r["vlans"], key=lambda x: x.vlan_id):
                    w.writerow(asdict(v))
        total = sum(len(r["vlans"]) for r in results)
        print(f"\nCSV written: {args.csv} ({total} VLAN entries)")

    if args.json_out:
        payload = [{"host": r["host"], "os": r["os"], "status": r["status"],
                    "error": r.get("error"), "vlans": [asdict(v) for v in r["vlans"]]}
                   for r in results]
        Path(args.json_out).write_text(
            json.dumps({"devices": payload, "issues": issues}, indent=2), encoding="utf-8")

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


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