VPN Crypto Audit

#!/usr/bin/env python3
"""
VPN Crypto Audit

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/vpn-crypto-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" "requests>=2.28"
"""

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

try:
    import requests
except ImportError:
    requests = None

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


# ---------------------------------------------------------------------------
# What counts as weak  (NIST SP 800-131A Rev.2, RFC 8247)
# ---------------------------------------------------------------------------

WEAK_ENCR = {
    "des": ("CRITICAL", "56-bit DES is trivially breakable"),
    "3des": ("HIGH", "64-bit block size; Sweet32 birthday attack, deprecated by RFC 8247"),
    "null": ("CRITICAL", "no encryption at all - traffic is in clear text"),
    "rc4": ("CRITICAL", "RC4 is broken and prohibited"),
}

WEAK_HASH = {
    "md5": ("HIGH", "MD5 collisions are practical"),
    "sha1": ("MEDIUM", "SHA-1 is deprecated for new deployments"),
    "sha": ("MEDIUM", "bare 'sha' means SHA-1, which is deprecated"),
}

WEAK_DH = {
    1: ("CRITICAL", "768-bit MODP - breakable with modest resources"),
    2: ("HIGH", "1024-bit MODP - within reach of a well-funded adversary (Logjam)"),
    5: ("HIGH", "1536-bit MODP - below current guidance"),
    22: ("HIGH", "1024-bit MODP with 160-bit subgroup - known weaknesses"),
    23: ("HIGH", "2048-bit MODP with 224-bit subgroup - known weaknesses"),
    24: ("MEDIUM", "2048-bit MODP with 256-bit subgroup - superseded"),
}

FIX_ENCR = "use AES-GCM-256, or AES-CBC-256 with a separate SHA-256 integrity algorithm"
FIX_HASH = "use SHA-256 or SHA-384. AES-GCM provides integrity on its own"
FIX_DH = "use group 14 (2048-bit MODP) at minimum; prefer 19/20/21 (ECP)"

RANK = {"CRITICAL": 0, "HIGH": 1, "MEDIUM": 2, "LOW": 3}


@dataclass
class Finding:
    severity: str
    location: str
    issue: str
    why: str
    fix: str
    phase: str = ""


# ---------------------------------------------------------------------------
# Commands
# ---------------------------------------------------------------------------

COMMAND_MAP = {
    "ios": {"ikev1": "show crypto isakmp policy",
            "ikev2": "show crypto ikev2 proposal",
            "ipsec": "show crypto ipsec transform-set",
            "profile": "show running-config | section crypto ipsec profile"},
    "iosxe": {"ikev1": "show crypto isakmp policy",
              "ikev2": "show crypto ikev2 proposal",
              "ipsec": "show crypto ipsec transform-set",
              "profile": "show running-config | section crypto ipsec profile"},
    "iosxr": {"ikev2": "show running-config crypto ikev2 proposal",
              "ipsec": "show running-config crypto ipsec transform-set"},
    "junos": {"ikev2": "show configuration security ike | display set",
              "ipsec": "show configuration security ipsec | display set"},
    "panos": {"ikev2": "show config running xpath "
                       "/config/devices/entry/network/ike/crypto-profiles",
              "ipsec": "show vpn ipsec-sa"},
    "fortios": {"ikev2": "show vpn ipsec phase1-interface",
                "ipsec": "show vpn ipsec phase2-interface"},
}

# Platforms whose crypto config this script does NOT know how to read. They are
# reported as UNAUDITED rather than clean - a security tool that says "no
# findings" about a device it never actually inspected is worse than one that
# refuses.
UNAUDITED = {
    "nxos": "NX-OS crypto configuration is not parsed by this script",
    "eos": "Arista EOS does not terminate IPsec in this form",
}


# ---------------------------------------------------------------------------
# Shared judgement
# ---------------------------------------------------------------------------

def judge_encr(value: str, location: str, phase: str) -> Finding | None:
    v = value.lower()
    # 3DES must be matched before DES. Platforms spell it "3des", "3DES",
    # "Three key triple DES" and "3-DES", and \bdes\b matches inside all of
    # them - which would rate 3DES as CRITICAL instead of HIGH.
    if re.search(r"3-?des|triple\s+des", v):
        sev, why = WEAK_ENCR["3des"]
        return Finding(sev, location, f"encryption {value}", why, FIX_ENCR, phase)
    for name in sorted(WEAK_ENCR, key=len, reverse=True):
        sev, why = WEAK_ENCR[name]
        if re.search(rf"\b{re.escape(name)}\b", v):
            return Finding(sev, location, f"encryption {value}", why, FIX_ENCR, phase)
    return None


def judge_hash(value: str, location: str, phase: str) -> Finding | None:
    v = value.lower().replace("-", "")
    if any(s in v for s in ("sha256", "sha384", "sha512", "sha2")):
        return None
    for name, (sev, why) in WEAK_HASH.items():
        if name.replace("-", "") in v:
            return Finding(sev, location, f"integrity {value}", why, FIX_HASH, phase)
    return None


def judge_dh(group: int, location: str, phase: str) -> Finding | None:
    if group in WEAK_DH:
        sev, why = WEAK_DH[group]
        return Finding(sev, location, f"DH group {group}", why, FIX_DH, phase)
    return None


# ---------------------------------------------------------------------------
# Parsers, one per platform
# ---------------------------------------------------------------------------

def audit_ios_isakmp(output: str) -> list[Finding]:
    """`show crypto isakmp policy` - IKEv1 Phase 1."""
    out: list[Finding] = []
    pri = None
    seen_v1: set[str] = set()
    for line in output.splitlines():
        m = re.search(r"priority\s+(\d+)", line, re.IGNORECASE)
        if m:
            pri = m.group(1)
            loc = f"ISAKMP policy {pri}"
            # Emit the IKEv1 note ONCE per policy, not once per attribute line.
            if pri not in seen_v1:
                seen_v1.add(pri)
                out.append(Finding("MEDIUM", loc, "IKEv1 in use",
                                   "IKEv1 lacks the DoS protection, MOBIKE support and "
                                   "algorithm agility of IKEv2",
                                   "migrate to IKEv2 (RFC 7296)", "Phase 1"))
            continue
        if not pri:
            continue
        loc = f"ISAKMP policy {pri}"
        m = re.search(r"encryption algorithm:\s*(.+)", line, re.IGNORECASE)
        if m:
            f = judge_encr(m.group(1).strip(), loc, "Phase 1")
            if f:
                out.append(f)
        m = re.search(r"hash algorithm:\s*(.+)", line, re.IGNORECASE)
        if m:
            f = judge_hash(m.group(1).strip(), loc, "Phase 1")
            if f:
                out.append(f)
        m = re.search(r"Diffie-Hellman group:\s*#?(\d+)", line, re.IGNORECASE)
        if m:
            f = judge_dh(int(m.group(1)), loc, "Phase 1")
            if f:
                out.append(f)
    return out


def audit_ios_ikev2(output: str) -> list[Finding]:
    """`show crypto ikev2 proposal`."""
    out: list[Finding] = []
    name = None
    for line in output.splitlines():
        m = re.search(r"IKEv2 proposal:\s*(\S+)", line, re.IGNORECASE)
        if m:
            name = m.group(1)
            continue
        if not name:
            continue
        loc = f"IKEv2 proposal {name}"
        m = re.search(r"Encryption\s*:\s*(.+)", line, re.IGNORECASE)
        if m:
            for tok in m.group(1).split():
                f = judge_encr(tok, loc, "Phase 1")
                if f:
                    out.append(f)
        m = re.search(r"Integrity\s*:\s*(.+)", line, re.IGNORECASE)
        if m:
            for tok in m.group(1).split():
                f = judge_hash(tok, loc, "Phase 1")
                if f:
                    out.append(f)
        m = re.search(r"DH Group\s*:\s*(.+)", line, re.IGNORECASE)
        if m:
            for g in re.findall(r"(?:DH_GROUP_|group\s*)(\d+)", m.group(1), re.IGNORECASE):
                f = judge_dh(int(g), loc, "Phase 1")
                if f:
                    out.append(f)
    return out


def audit_ios_transform(output: str) -> list[Finding]:
    """`show crypto ipsec transform-set` - IPsec Phase 2."""
    out: list[Finding] = []
    for line in output.splitlines():
        m = re.search(r"Transform set\s+(\S+?)\s*:", line, re.IGNORECASE)
        if not m:
            continue
        name = m.group(1)
        loc = f"transform-set {name}"
        body = line
        for tok in re.findall(r"esp-[\w-]+|ah-[\w-]+|comp-\w+", body, re.IGNORECASE):
            t = tok.lower()
            f = judge_encr(t.replace("esp-", ""), loc, "Phase 2")
            if f:
                out.append(f)
            f = judge_hash(t.replace("esp-", "").replace("-hmac", ""), loc, "Phase 2")
            if f:
                out.append(f)
    return out


def audit_ios_pfs(output: str) -> list[Finding]:
    """
    Check PFS on IPsec profiles and crypto maps.

    Without PFS, compromising the IKE SA key exposes every IPsec SA derived
    from it, including traffic already captured.
    """
    out: list[Finding] = []
    name = None
    has_pfs = False
    for line in output.splitlines() + ["!"]:
        m = re.match(r"crypto ipsec profile (\S+)", line, re.IGNORECASE)
        if m or line.strip() == "!":
            if name and not has_pfs:
                out.append(Finding("HIGH", f"IPsec profile {name}", "PFS is not configured",
                                   "without Perfect Forward Secrecy, compromise of the IKE "
                                   "key exposes every IPsec SA derived from it, including "
                                   "previously captured traffic",
                                   "add 'set pfs group14' (or 19/20) to the profile",
                                   "Phase 2"))
            name = m.group(1) if m else None
            has_pfs = False
            continue
        if name and re.search(r"set pfs", line, re.IGNORECASE):
            has_pfs = True
            m = re.search(r"group(\d+)", line, re.IGNORECASE)
            if m:
                f = judge_dh(int(m.group(1)), f"IPsec profile {name}", "Phase 2")
                if f:
                    out.append(f)
    return out


def audit_junos(output: str, phase: str) -> list[Finding]:
    """JunOS `| display set` config."""
    out: list[Finding] = []
    for line in output.splitlines():
        m = re.search(r"proposal (\S+)", line)
        loc = f"{phase} proposal {m.group(1)}" if m else phase
        m = re.search(r"encryption-algorithm\s+(\S+)", line, re.IGNORECASE)
        if m:
            f = judge_encr(m.group(1), loc, phase)
            if f:
                out.append(f)
        m = re.search(r"authentication-algorithm\s+(\S+)", line, re.IGNORECASE)
        if m:
            f = judge_hash(m.group(1), loc, phase)
            if f:
                out.append(f)
        m = re.search(r"dh-group\s+group(\d+)", line, re.IGNORECASE)
        if m:
            f = judge_dh(int(m.group(1)), loc, phase)
            if f:
                out.append(f)
        m = re.search(r"perfect-forward-secrecy keys group(\d+)", line, re.IGNORECASE)
        if m:
            f = judge_dh(int(m.group(1)), loc, "Phase 2")
            if f:
                out.append(f)
        if re.search(r"ike.*version\s+v1\b", line, re.IGNORECASE):
            out.append(Finding("MEDIUM", loc, "IKEv1 in use",
                               "IKEv1 lacks the DoS protection and algorithm agility of IKEv2",
                               "set security ike gateway <name> version v2-only", phase))
    return out


def audit_fortios(output: str, phase: str) -> list[Finding]:
    """FortiOS `show vpn ipsec phase1-interface` / `phase2-interface`."""
    out: list[Finding] = []
    name = "unnamed"
    for line in output.splitlines():
        m = re.match(r'\s*edit\s+"([^"]+)"', line)
        if m:
            name = m.group(1)
            continue
        loc = f"{phase} {name}"

        # FortiOS writes proposals as aes256-sha256 or des-md5, space separated
        m = re.search(r'set proposal\s+(.+)', line, re.IGNORECASE)
        if m:
            for prop in m.group(1).replace('"', "").split():
                parts = prop.split("-")
                if parts:
                    f = judge_encr(parts[0], loc, phase)
                    if f:
                        out.append(f)
                if len(parts) > 1:
                    f = judge_hash(parts[1], loc, phase)
                    if f:
                        out.append(f)
        m = re.search(r"set dhgrp\s+(.+)", line, re.IGNORECASE)
        if m:
            for g in m.group(1).replace('"', "").split():
                if g.isdigit():
                    f = judge_dh(int(g), loc, phase)
                    if f:
                        out.append(f)
        m = re.search(r"set ike-version\s+1", line, re.IGNORECASE)
        if m:
            out.append(Finding("MEDIUM", loc, "IKEv1 in use",
                               "IKEv1 lacks the DoS protection and algorithm agility of IKEv2",
                               "set ike-version 2", phase))
        if re.search(r"set pfs disable", line, re.IGNORECASE):
            out.append(Finding("HIGH", loc, "PFS is disabled",
                               "compromise of the IKE key exposes every IPsec SA derived "
                               "from it, including previously captured traffic",
                               "set pfs enable and choose dhgrp 14 or higher", "Phase 2"))
    return out


def audit_panos(output: str, phase: str) -> list[Finding]:
    """PAN-OS crypto profiles from the running config XML/CLI dump."""
    out: list[Finding] = []
    name = "unnamed"
    for line in output.splitlines():
        m = re.search(r'entry name="([^"]+)"', line)
        if m:
            name = m.group(1)
            continue
        loc = f"{phase} profile {name}"
        for tok in re.findall(r"<member>([^<]+)</member>", line):
            t = tok.strip().lower()
            f = judge_encr(t, loc, phase)
            if f:
                out.append(f)
            f = judge_hash(t, loc, phase)
            if f:
                out.append(f)
            m2 = re.fullmatch(r"group(\d+)", t)
            if m2:
                f = judge_dh(int(m2.group(1)), loc, phase)
                if f:
                    out.append(f)
    return out


# ---------------------------------------------------------------------------
# Per-device audit
# ---------------------------------------------------------------------------

def audit_device(dev: dict, args) -> dict:
    host = dev["host"]
    os_key = dev["os"]
    r = {"host": host, "os": os_key, "findings": [], "error": None,
         "unaudited": False, "commands_empty": []}

    if os_key in UNAUDITED:
        r["unaudited"] = True
        r["error"] = UNAUDITED[os_key]
        return r
    if os_key not in COMMAND_MAP:
        r["unaudited"] = True
        r["error"] = f"crypto configuration parsing not implemented for {os_key}"
        return r

    conn = None
    try:
        conn = connect(dev)
        out: dict[str, str] = {}
        for key, cmd in COMMAND_MAP[os_key].items():
            try:
                out[key] = conn.send_command(cmd, read_timeout=args.read_timeout)
            except Exception:  # noqa: BLE001
                out[key] = ""
            if not out[key].strip():
                r["commands_empty"].append(key)

        f: list[Finding] = []
        if os_key in ("ios", "iosxe"):
            f += audit_ios_isakmp(out.get("ikev1", ""))
            f += audit_ios_ikev2(out.get("ikev2", ""))
            f += audit_ios_transform(out.get("ipsec", ""))
            f += audit_ios_pfs(out.get("profile", ""))
        elif os_key == "iosxr":
            f += audit_ios_ikev2(out.get("ikev2", ""))
            f += audit_ios_transform(out.get("ipsec", ""))
        elif os_key == "junos":
            f += audit_junos(out.get("ikev2", ""), "Phase 1")
            f += audit_junos(out.get("ipsec", ""), "Phase 2")
        elif os_key == "fortios":
            f += audit_fortios(out.get("ikev2", ""), "Phase 1")
            f += audit_fortios(out.get("ipsec", ""), "Phase 2")
        elif os_key == "panos":
            f += audit_panos(out.get("ikev2", ""), "Phase 1")
            f += audit_panos(out.get("ipsec", ""), "Phase 2")

        # Deduplicate: the same weak algorithm named in several proposals is
        # one finding to fix, not five.
        seen: set[tuple] = set()
        uniq: list[Finding] = []
        for x in f:
            key = (x.location, x.issue)
            if key not in seen:
                seen.add(key)
                uniq.append(x)
        uniq.sort(key=lambda x: (RANK[x.severity], x.location))
        r["findings"] = [asdict(x) for x in uniq]

        # If every command came back empty we did not actually audit anything.
        if len(r["commands_empty"]) == len(COMMAND_MAP[os_key]):
            r["unaudited"] = True
            r["error"] = ("every crypto command returned nothing - the account "
                          "probably lacks privilege to read the configuration")
        return r

    except NetmikoAuthenticationException as exc:
        r["error"] = f"authentication failed - check NET_USER/NET_PASS ({exc})"
    except NetmikoTimeoutException as exc:
        r["error"] = f"unreachable or SSH timed out ({exc})"
    except Exception as exc:  # noqa: BLE001 - one device must not abort the run
        r["error"] = f"{type(exc).__name__}: {exc}"
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass
    return r


def summarise(r: dict) -> str:
    host, os_key = r["host"], r["os"]
    if r["unaudited"]:
        return f"[UNAUDITED] {host:<22} {r['error']}"
    if r["error"]:
        return f"[ERROR]     {host:<22} ({os_key}) {r['error']}"
    f = r["findings"]
    if not f:
        return f"[PASS]      {host:<22} ({os_key}) no weak crypto found"
    sev = sorted({x["severity"] for x in f}, key=lambda s: RANK[s])[0]
    return f"[{sev:<9}] {host:<22} ({os_key}) {len(f)} finding(s)"


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Audit IPsec/IKE crypto configuration against NIST SP 800-131A "
                    "and RFC 8247.",
        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 VPN endpoints to audit")
    ap.add_argument("--min-severity", choices=list(RANK), default="MEDIUM",
                    help="Only report findings at or above this severity. Default: MEDIUM")
    ap.add_argument("--limit", help="Only audit hosts whose name contains this text")
    ap.add_argument("--webhook", default="", help="Slack or Teams webhook URL")
    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 devices to audit at once. Default: 10")
    ap.add_argument("--json-out", help="Write machine-readable findings as audit evidence")
    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 VPN crypto on {len(devices)} device(s) "
          f"against NIST SP 800-131A / RFC 8247...")
    print("=" * 88)

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

    floor = RANK[args.min_severity]
    tally = defaultdict(int)

    for r in sorted(results, key=lambda x: x["host"]):
        print(summarise(r))
        for f in r["findings"]:
            if RANK[f["severity"]] > floor:
                continue
            tally[f["severity"]] += 1
            print(f"              [{f['severity']:<8}] {f['location']}: {f['issue']}")
            print(f"                          why: {f['why']}")
            print(f"                          fix: {f['fix']}")

    print("=" * 88)
    unaudited = [r for r in results if r["unaudited"]]
    errs = [r for r in results if r["error"] and not r["unaudited"]]
    clean = [r for r in results if not r["findings"] and not r["error"]]

    parts = "  ".join(f"{k}={tally[k]}" for k in ("CRITICAL", "HIGH", "MEDIUM", "LOW")
                      if tally[k])
    print(f"{parts or 'no findings'}   |   {len(clean)} device(s) clean, "
          f"{len(unaudited)} unaudited, {len(errs)} unreachable")

    if unaudited:
        print(f"\n{len(unaudited)} device(s) could NOT be audited. They are reported as "
              f"UNAUDITED, not clean -\nan audit that silently passes a device it never "
              f"inspected is worse than no audit:")
        for r in unaudited:
            print(f"  - {r['host']}: {r['error']}")

    if tally.get("CRITICAL"):
        print("\nCRITICAL findings present. These are algorithms that can be broken "
              "today, not\nmerely deprecated - traffic protected by them should be "
              "treated as exposed.")

    if args.webhook and (tally or errs) and requests is not None:
        lines = [summarise(r) for r in results if r["findings"] or r["error"]]
        try:
            requests.post(args.webhook,
                          json={"text": "*VPN crypto audit*\n" + "\n".join(lines)},
                          timeout=10)
        except Exception as exc:  # noqa: BLE001
            print(f"[WARN] webhook failed: {exc}", file=sys.stderr)

    if args.json_out:
        Path(args.json_out).write_text(json.dumps(results, indent=2), encoding="utf-8")
        print(f"\nAudit evidence written to {args.json_out}")

    return 1 if (tally.get("CRITICAL") or tally.get("HIGH") or errs) else 0


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