Pre-Flight Config Validator

#!/usr/bin/env python3
"""
Pre-Flight Config Validator

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

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 ipaddress
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 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("validator")

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

SEVERITY_ORDER = {"error": 0, "warning": 1, "info": 2}


# --------------------------------------------------------------------------
# Findings
# --------------------------------------------------------------------------

@dataclass
class Finding:
    severity: str          # error | warning | info
    rule: str
    line_no: int
    line: str
    message: str
    fix: str = ""


@dataclass
class Report:
    host: str
    os: str
    layer: str = "offline"
    findings: list[Finding] = field(default_factory=list)
    lines_checked: int = 0
    elapsed_s: float = 0.0

    @property
    def errors(self) -> int:
        return sum(1 for f in self.findings if f.severity == "error")

    @property
    def warnings(self) -> int:
        return sum(1 for f in self.findings if f.severity == "warning")


# --------------------------------------------------------------------------
# Layer 1: offline static rules
# --------------------------------------------------------------------------

# Commands that are almost never intended inside an automated change.
DANGEROUS = {
    r"^\s*no\s+router\s+\w+": ("error", "Removes an entire routing process"),
    r"^\s*no\s+interface\s": ("error", "Deletes an interface and all sub-config"),
    r"^\s*no\s+vlan\s+\d": ("error", "Deletes a VLAN; hosts in it lose connectivity"),
    r"^\s*(reload|reboot)\b": ("error", "Reboots the device"),
    r"^\s*write\s+erase|^\s*erase\s+startup": ("error", "Erases the startup config"),
    r"^\s*delete\s+/force": ("error", "Force-deletes a file without confirmation"),
    r"^\s*no\s+ip\s+routing": ("error", "Disables routing entirely"),
    r"^\s*shutdown\s*$": ("warning", "Shuts an interface; confirm this is intended"),
    r"^\s*no\s+ip\s+access-list": ("warning", "Removes an ACL; traffic may be denied or permitted unexpectedly"),
    r"^\s*request\s+system\s+(reboot|halt)": ("error", "Reboots or halts the device"),
    r"^\s*execute\s+factoryreset": ("error", "Factory resets the device"),
    r"^\s*delete\s+(vdom|system)": ("error", "Deletes a VDOM or system object"),
}

# Per-OS line shape. A line failing all patterns for its OS is suspicious.
OS_HINTS = {
    "junos": {
        "requires_set": True,
        "note": "JunOS fragments must be in 'set' / 'delete' form for scripted push",
    },
    "fortios": {
        "requires_config_block": True,
        "note": "FortiOS fragments must open with 'config ...' and close with 'end'",
    },
    "panos": {
        "requires_set": True,
        "note": "PAN-OS fragments must be in 'set' form",
    },
}


def check_dangerous(lines: list[str]) -> list[Finding]:
    out = []
    for i, ln in enumerate(lines, 1):
        for pat, (sev, msg) in DANGEROUS.items():
            if re.search(pat, ln, re.IGNORECASE):
                out.append(Finding(sev, "dangerous-command", i, ln.strip(), msg,
                                   "Confirm intent, or scope the change more narrowly"))
    return out


def check_ip_sanity(lines: list[str]) -> list[Finding]:
    """Validate every address/mask pair and flag classic mistakes."""
    out = []
    pat_ios = re.compile(r"ip address (\d+\.\d+\.\d+\.\d+) (\d+\.\d+\.\d+\.\d+)")
    pat_cidr = re.compile(r"(\d+\.\d+\.\d+\.\d+)/(\d{1,2})")

    for i, ln in enumerate(lines, 1):
        m = pat_ios.search(ln)
        if m:
            addr, mask = m.groups()
            try:
                net = ipaddress.ip_network(f"{addr}/{mask}", strict=False)
            except ValueError as exc:
                out.append(Finding("error", "invalid-ip", i, ln.strip(),
                                   f"Invalid address/mask: {exc}", "Correct the address or mask"))
                continue
            ip = ipaddress.ip_address(addr)
            if net.prefixlen < 31:
                if ip == net.network_address:
                    out.append(Finding("error", "network-address", i, ln.strip(),
                                       f"{addr} is the network address of {net}",
                                       f"Use a host address, e.g. {net.network_address + 1}"))
                elif ip == net.broadcast_address:
                    out.append(Finding("error", "broadcast-address", i, ln.strip(),
                                       f"{addr} is the broadcast address of {net}",
                                       f"Use a host address, e.g. {net.network_address + 1}"))
            if net.prefixlen <= 23 and not ip.is_private:
                out.append(Finding("warning", "large-public-subnet", i, ln.strip(),
                                   f"{net} is a large public range on an interface",
                                   "Confirm this address space is yours"))

        for addr, plen in pat_cidr.findall(ln):
            try:
                ipaddress.ip_network(f"{addr}/{plen}", strict=False)
            except ValueError as exc:
                out.append(Finding("error", "invalid-cidr", i, ln.strip(),
                                   f"Invalid prefix {addr}/{plen}: {exc}", "Correct the prefix"))
    return out


def check_overlaps(lines: list[str]) -> list[Finding]:
    """Two interfaces in the same fragment must not share a subnet."""
    out = []
    seen: dict[Any, tuple[int, str]] = {}
    cur_intf = "<global>"
    pat = re.compile(r"ip address (\d+\.\d+\.\d+\.\d+) (\d+\.\d+\.\d+\.\d+)")

    for i, ln in enumerate(lines, 1):
        s = ln.strip()
        if re.match(r"^interface\s+\S+", s, re.IGNORECASE):
            cur_intf = s.split(None, 1)[1]
        m = pat.search(ln)
        if not m:
            continue
        try:
            net = ipaddress.ip_network(f"{m.group(1)}/{m.group(2)}", strict=False)
        except ValueError:
            continue
        if net in seen:
            prev_line, prev_intf = seen[net]
            out.append(Finding("error", "subnet-overlap", i, s,
                               f"{net} already assigned to {prev_intf} on line {prev_line}",
                               "Use a different subnet, or remove the duplicate"))
        else:
            seen[net] = (i, cur_intf)
    return out


def check_ordering(lines: list[str]) -> list[Finding]:
    """Objects must be defined before they are referenced."""
    out = []
    defined_acl, defined_pm, defined_rm = set(), set(), set()
    pending: list[tuple[int, str, str, set]] = []

    for i, ln in enumerate(lines, 1):
        s = ln.strip()
        m = re.match(r"ip access-list (?:standard|extended) (\S+)", s, re.IGNORECASE)
        if m:
            defined_acl.add(m.group(1))
        m = re.match(r"policy-map (\S+)", s, re.IGNORECASE)
        if m:
            defined_pm.add(m.group(1))
        m = re.match(r"route-map (\S+)", s, re.IGNORECASE)
        if m:
            defined_rm.add(m.group(1))

        m = re.search(r"ip access-group (\S+)", s, re.IGNORECASE)
        if m:
            pending.append((i, s, m.group(1), defined_acl))
        m = re.search(r"service-policy (?:input|output) (\S+)", s, re.IGNORECASE)
        if m:
            pending.append((i, s, m.group(1), defined_pm))
        m = re.search(r"route-map (\S+) (?:in|out)\b", s, re.IGNORECASE)
        if m:
            pending.append((i, s, m.group(1), defined_rm))

    for line_no, text, name, pool in pending:
        if name not in pool:
            out.append(Finding("warning", "forward-reference", line_no, text,
                               f"'{name}' is referenced but not defined in this fragment",
                               "Confirm it already exists on the device, or define it first"))
    return out


def check_os_shape(lines: list[str], os_key: str) -> list[Finding]:
    out = []
    hints = OS_HINTS.get(os_key)
    if not hints:
        return out

    if hints.get("requires_set"):
        bad = [(i, ln) for i, ln in enumerate(lines, 1)
               if ln.strip() and not re.match(r"^\s*(set|delete|deactivate|activate)\s", ln)]
        for i, ln in bad[:10]:
            out.append(Finding("error", "wrong-syntax-form", i, ln.strip(),
                               hints["note"], "Convert the fragment to set-form"))

    if hints.get("requires_config_block"):
        text = "\n".join(lines)
        opens = len(re.findall(r"^\s*config\s+\S", text, re.MULTILINE))
        closes = len(re.findall(r"^\s*end\s*$", text, re.MULTILINE))
        if opens and opens != closes:
            out.append(Finding("error", "unbalanced-block", 1, f"{opens} config / {closes} end",
                               f"{opens} 'config' blocks but {closes} 'end' statements",
                               "Every 'config' must be closed by a matching 'end'"))
        edits = len(re.findall(r"^\s*edit\s+\S", text, re.MULTILINE))
        nexts = len(re.findall(r"^\s*next\s*$", text, re.MULTILINE))
        if edits and edits != nexts:
            out.append(Finding("error", "unbalanced-edit", 1, f"{edits} edit / {nexts} next",
                               f"{edits} 'edit' statements but {nexts} 'next'",
                               "Every 'edit' must be closed by 'next'"))
    return out


def check_indentation(lines: list[str], os_key: str) -> list[Finding]:
    """A sub-command outside any parent block is a very common copy-paste bug."""
    if os_key in ("junos", "panos", "fortios"):
        return []
    out = []
    in_block = False
    sub_only = re.compile(
        r"^\s+(switchport|ip address|no shutdown|description|channel-group|"
        r"vrf forwarding|encapsulation|speed|duplex|mtu)\b", re.IGNORECASE)
    for i, ln in enumerate(lines, 1):
        if re.match(r"^(interface|router|line|vlan|policy-map|class-map|"
                    r"ip access-list|route-map|vrf)\s", ln, re.IGNORECASE):
            in_block = True
            continue
        if ln.strip() in ("!", "exit"):
            in_block = False
            continue
        if ln and not ln[0].isspace() and ln.strip():
            in_block = False
        if sub_only.match(ln) and not in_block:
            out.append(Finding("error", "orphaned-subcommand", i, ln.strip(),
                               "Sub-command appears outside any parent block",
                               "Add the parent (e.g. 'interface Gi0/1') above this line"))
    return out


def validate_offline(lines: list[str], os_key: str) -> list[Finding]:
    findings: list[Finding] = []
    findings += check_dangerous(lines)
    findings += check_ip_sanity(lines)
    findings += check_overlaps(lines)
    findings += check_ordering(lines)
    findings += check_os_shape(lines, os_key)
    findings += check_indentation(lines, os_key)
    return findings


# --------------------------------------------------------------------------
# Layer 2: semantic checks against live state
# --------------------------------------------------------------------------

STATE_CMDS = {
    "ios":    {"intf": "show ip interface brief", "vlan": "show vlan brief", "po": "show etherchannel summary"},
    "iosxe":  {"intf": "show ip interface brief", "vlan": "show vlan brief", "po": "show etherchannel summary"},
    "iosxr":  {"intf": "show ipv4 interface brief", "vlan": "", "po": "show bundle brief"},
    "nxos":   {"intf": "show ip interface brief", "vlan": "show vlan brief", "po": "show port-channel summary"},
    "eos":    {"intf": "show ip interface brief", "vlan": "show vlan", "po": "show port-channel summary"},
    "junos":  {"intf": "show interfaces terse", "vlan": "show vlans", "po": "show interfaces ae0 terse"},
    "panos":  {"intf": "show interface all", "vlan": "", "po": ""},
    "fortios": {"intf": "get system interface physical", "vlan": "", "po": ""},
}


def validate_semantic(conn, os_key: str, lines: list[str]) -> list[Finding]:
    out: list[Finding] = []
    cmds = STATE_CMDS.get(os_key, {})

    intf_state = conn.send_command(cmds["intf"], read_timeout=60) if cmds.get("intf") else ""
    po_state = conn.send_command(cmds["po"], read_timeout=60) if cmds.get("po") else ""
    vlan_state = conn.send_command(cmds["vlan"], read_timeout=60) if cmds.get("vlan") else ""

    for i, ln in enumerate(lines, 1):
        s = ln.strip()

        m = re.match(r"interface\s+(\S+)", s, re.IGNORECASE)
        if m:
            name = m.group(1)
            short = name.replace("GigabitEthernet", "Gi").replace("TenGigabitEthernet", "Te")
            if name not in intf_state and short not in intf_state:
                out.append(Finding("warning", "unknown-interface", i, s,
                                   f"'{name}' does not appear in the live interface table",
                                   "Verify the name, including slot/port numbering"))
            if po_state and (name in po_state or short in po_state):
                out.append(Finding("error", "member-of-portchannel", i, s,
                                   f"'{name}' is a port-channel member; config belongs on the Po",
                                   "Apply the change to the port-channel interface instead"))

        m = re.match(r"vlan\s+(\d+)", s, re.IGNORECASE)
        if m and vlan_state:
            vid = m.group(1)
            if re.search(rf"^{vid}\s", vlan_state, re.MULTILINE):
                out.append(Finding("info", "vlan-exists", i, s,
                                   f"VLAN {vid} already exists; this will modify it",
                                   "Confirm the existing name and ports are safe to change"))
    return out


# --------------------------------------------------------------------------
# Layer 3: device-side parser, no commit
# --------------------------------------------------------------------------

def validate_on_device(conn, os_key: str, lines: list[str]) -> list[Finding]:
    """Let the platform's own parser judge the config. Never commits."""
    out: list[Finding] = []

    if os_key == "junos":
        conn.config_mode()
        try:
            for ln in lines:
                r = conn.send_command_timing(ln)
                if "error" in r.lower() or "unknown command" in r.lower():
                    out.append(Finding("error", "device-parser", lines.index(ln) + 1, ln.strip(),
                                       r.strip()[:200], "Fix the syntax"))
            chk = conn.send_command_timing("commit check")
            if "error" in chk.lower():
                out.append(Finding("error", "commit-check", 0, "commit check",
                                   chk.strip()[:300], "Resolve before pushing"))
            elif "succeeds" in chk.lower():
                out.append(Finding("info", "commit-check", 0, "commit check",
                                   "configuration check succeeds", ""))
        finally:
            conn.send_command_timing("rollback 0")
            conn.exit_config_mode()
        return out

    if os_key == "eos":
        session = f"val-{int(time.time())}"
        try:
            conn.send_command_timing(f"configure session {session}")
            for i, ln in enumerate(lines, 1):
                r = conn.send_command_timing(ln)
                if "% Invalid" in r or "Error" in r:
                    out.append(Finding("error", "device-parser", i, ln.strip(),
                                       r.strip()[:200], "Fix the syntax"))
            diff = conn.send_command_timing("show session-config diffs")
            if diff.strip():
                out.append(Finding("info", "session-diff", 0, "diff",
                                   f"{len(diff.strip().splitlines())} line(s) would change", ""))
        finally:
            conn.send_command_timing("abort")
        return out

    if os_key == "iosxr":
        try:
            conn.config_mode()
            for i, ln in enumerate(lines, 1):
                conn.send_command_timing(ln)
            failed = conn.send_command_timing("show configuration failed")
            if failed.strip() and "No failed" not in failed:
                out.append(Finding("error", "device-parser", 0, "staged config",
                                   failed.strip()[:400], "Fix the rejected lines"))
            else:
                diff = conn.send_command_timing("show configuration")
                out.append(Finding("info", "staged-diff", 0, "diff",
                                   f"{len(diff.strip().splitlines())} line(s) staged cleanly", ""))
        finally:
            conn.send_command_timing("abort")
        return out

    if os_key in ("ios", "iosxe", "nxos"):
        # No non-committing parser exists. Validate command existence via the
        # parser help, which is read-only and safe.
        for i, ln in enumerate(lines, 1):
            s = ln.strip()
            if not s or s.startswith("!") or s.startswith("interface") or " " not in s:
                continue
            head = s.split()[0]
            if head in ("no", "default"):
                continue
            probe = conn.send_command_timing(f"show parser dump exec | include ^{head}$")
            if "Invalid" in probe:
                out.append(Finding("warning", "unverified-command", i, s,
                                   f"Could not confirm '{head}' is a valid keyword",
                                   "Validate manually in a lab"))
        out.append(Finding("info", "no-native-check", 0, os_key,
                           f"{os_key} has no non-committing validator; rely on layers 1-2 "
                           f"plus config-rollback's confirm window", ""))
        return out

    out.append(Finding("info", "no-native-check", 0, os_key,
                       f"Device-side validation not implemented for {os_key}", ""))
    return out


# --------------------------------------------------------------------------
# Orchestration
# --------------------------------------------------------------------------

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 load_lines(cfg_dir: str, host: str, os_key: str) -> list[str]:
    base = Path(cfg_dir)
    for cand in (base / f"{host}.cfg", base / f"{os_key}.cfg", base / "common.cfg"):
        if cand.exists():
            return [ln.rstrip() for ln in cand.read_text(encoding="utf-8").splitlines()
                    if ln.strip() and not ln.lstrip().startswith("#")]
    return []


def process(dev: dict, args) -> Report:
    rep = Report(host=dev["host"], os=dev["os"])
    t0 = time.time()
    lines = load_lines(args.config_dir, dev["host"], dev["os"])
    rep.lines_checked = len(lines)
    if not lines:
        rep.findings.append(Finding("warning", "no-fragment", 0, "",
                                    "No config fragment found for this host", ""))
        rep.elapsed_s = round(time.time() - t0, 1)
        return rep

    rep.findings += validate_offline(lines, dev["os"])

    if args.offline_only:
        rep.elapsed_s = round(time.time() - t0, 1)
        return rep

    conn = None
    try:
        conn = connect(dev)
        rep.layer = "semantic"
        rep.findings += validate_semantic(conn, dev["os"], lines)
        if args.device_check:
            rep.layer = "device"
            rep.findings += validate_on_device(conn, dev["os"], lines)
    except NetmikoAuthenticationException as exc:
        rep.findings.append(Finding("error", "auth", 0, "", f"authentication failed: {exc}", ""))
    except NetmikoTimeoutException as exc:
        rep.findings.append(Finding("error", "unreachable", 0, "", f"timeout: {exc}", ""))
    except Exception as exc:  # noqa: BLE001
        rep.findings.append(Finding("error", "validator", 0, "",
                                    f"{type(exc).__name__}: {exc}", ""))
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass

    rep.elapsed_s = round(time.time() - t0, 1)
    return rep


def render(reports: list[Report], show_info: bool) -> int:
    sev_tag = {"error": "ERROR  ", "warning": "WARN   ", "info": "INFO   "}
    total_e = total_w = 0

    for rep in sorted(reports, key=lambda r: r.host):
        shown = [f for f in rep.findings if show_info or f.severity != "info"]
        total_e += rep.errors
        total_w += rep.warnings
        state = "FAIL" if rep.errors else ("WARN" if rep.warnings else "PASS")
        print(f"\n{'=' * 78}\n[{state}] {rep.host} ({rep.os})  "
              f"{rep.lines_checked} line(s)  layer={rep.layer}  {rep.elapsed_s}s")
        if not shown:
            print("       no findings")
            continue
        for f in sorted(shown, key=lambda x: (SEVERITY_ORDER[x.severity], x.line_no)):
            loc = f"L{f.line_no}" if f.line_no else "--"
            print(f"  {sev_tag[f.severity]} {loc:<5} [{f.rule}] {f.message}")
            if f.line:
                print(f"          > {f.line[:88]}")
            if f.fix:
                print(f"          fix: {f.fix}")

    print(f"\n{'=' * 78}")
    print(f"  {len(reports)} device(s)   errors={total_e}   warnings={total_w}")
    if total_e:
        print("  BLOCKED - resolve errors before pushing.")
        return 2
    if total_w:
        print("  Warnings only - review, then push with config-rollback.")
        return 1
    print("  Clean.")
    return 0


def main() -> int:
    ap = argparse.ArgumentParser(description="Validate config before pushing it.")
    ap.add_argument("--config-dir", required=True)
    ap.add_argument("--inventory", help="Required unless --offline-only")
    ap.add_argument("--offline-only", action="store_true",
                    help="Static analysis only; never contacts a device")
    ap.add_argument("--device-check", action="store_true",
                    help="Also use the platform's own parser (config mode, never commits)")
    ap.add_argument("--show-info", action="store_true", help="Include informational findings")
    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 args.offline_only:
        cfgs = sorted(Path(args.config_dir).glob("*.cfg"))
        if not cfgs:
            LOG.error("No .cfg files in %s", args.config_dir)
            return 3
        reports = []
        for p in cfgs:
            os_key = p.stem if p.stem in NETMIKO_DRIVER else "ios"
            lines = [ln.rstrip() for ln in p.read_text(encoding="utf-8").splitlines()
                     if ln.strip() and not ln.lstrip().startswith("#")]
            rep = Report(host=p.stem, os=os_key, lines_checked=len(lines))
            rep.findings = validate_offline(lines, os_key)
            reports.append(rep)
        return render(reports, args.show_info)

    if not args.inventory:
        LOG.error("--inventory is required unless --offline-only is used.")
        return 3
    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 3

    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

    LOG.info("Validating %d device(s) | device-check=%s", len(devices), args.device_check)

    reports: list[Report] = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futs = {pool.submit(process, d, args): d for d in devices}
        for fut in as_completed(futs):
            d = futs[fut]
            try:
                reports.append(fut.result())
            except Exception as exc:  # noqa: BLE001
                LOG.exception("%s: validator crashed", d["host"])
                r = Report(host=d["host"], os=d.get("os", "?"))
                r.findings.append(Finding("error", "crash", 0, "", str(exc), ""))
                reports.append(r)

    if args.json_out:
        Path(args.json_out).write_text(
            json.dumps([asdict(r) for r in reports], indent=2), encoding="utf-8")
        LOG.info("Results written to %s", args.json_out)

    return render(reports, args.show_info)


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