IKE SA Analyzer

#!/usr/bin/env python3
"""
IKE SA Analyzer

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/ike-sa-analyzer

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


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

COMMAND_MAP = {
    "ios": {"ikev2": "show crypto ikev2 sa detail",
            "ikev1": "show crypto isakmp sa detail",
            "ipsec": "show crypto ipsec sa"},
    "iosxe": {"ikev2": "show crypto ikev2 sa detail",
              "ikev1": "show crypto isakmp sa detail",
              "ipsec": "show crypto ipsec sa"},
    "iosxr": {"ikev2": "show crypto ikev2 sa detail",
              "ipsec": "show crypto ipsec sa summary"},
    "junos": {"ikev2": "show security ike security-associations detail",
              "ipsec": "show security ipsec security-associations detail"},
    "panos": {"ikev2": "show vpn ike-sa", "ipsec": "show vpn ipsec-sa"},
    "fortios": {"ikev2": "diagnose vpn ike status",
                "ipsec": "diagnose vpn tunnel list"},
}

NO_IKE = {
    "nxos": "NX-OS IPsec support is limited and does not expose IKE SA detail",
    "eos": "Arista EOS does not terminate IKE/IPsec in this form",
}

# RFC 8247 and RFC 9395 deprecations. These are not style preferences: DH 1, 2
# and 5 are breakable, and 3DES/DES are below current guidance.
WEAK_DH = {1, 2, 5, 22, 23, 24}
WEAK_ENCR = ("des", "3des", "null")
WEAK_HASH = ("md5", "sha1", "sha")
STRONG_DH = "14 (2048-bit MODP) minimum; 19/20 (ECP) or 21 preferred"

IPV4 = re.compile(r"\b(\d{1,3}(?:\.\d{1,3}){3})\b")


@dataclass
class SA:
    peer: str
    version: str = "IKEv2"
    state: str = "UNKNOWN"
    encr: str = "?"
    hash: str = "?"
    prf: str = "?"
    dh_group: int = 0
    lifetime_remaining: int | None = None
    local: str = ""
    encaps: int | None = None
    decaps: int | None = None


HEALTHY_STATES = {"READY", "UP-ACTIVE", "ESTAB", "ESTABLISHED", "ACTIVE", "UP", "MM_ACTIVE"}


# ---------------------------------------------------------------------------
# Parsers
# ---------------------------------------------------------------------------

def parse_ios_ikev2(output: str) -> list[SA]:
    """`show crypto ikev2 sa detail` - one block per SA."""
    sas: list[SA] = []
    for block in re.split(r"\n(?=\s*\d+\s+\d{1,3}(?:\.\d{1,3}){3}/)", output):
        peer = re.search(r"Remote\s+(\d{1,3}(?:\.\d{1,3}){3})", block) or \
               re.search(r"Peer:\s*(\d{1,3}(?:\.\d{1,3}){3})", block)
        if not peer:
            ips = IPV4.findall(block)
            if len(ips) < 2:
                continue
            peer_ip = ips[1]
        else:
            peer_ip = peer.group(1)

        sa = SA(peer=peer_ip, version="IKEv2")
        m = re.search(r"Status:\s*(\S+)", block, re.IGNORECASE)
        if m:
            sa.state = m.group(1).upper()
        m = re.search(r"Encr:\s*([\w-]+)", block, re.IGNORECASE)
        if m:
            sa.encr = m.group(1)
        m = re.search(r"Hash:\s*([\w-]+)", block, re.IGNORECASE)
        if m:
            sa.hash = m.group(1)
        m = re.search(r"PRF:\s*([\w-]+)", block, re.IGNORECASE)
        if m:
            sa.prf = m.group(1)
        m = re.search(r"DH\s*Grp:\s*(\d+)", block, re.IGNORECASE)
        if m:
            sa.dh_group = int(m.group(1))
        m = re.search(r"Life/Active Time:\s*\d+/(\d+)", block, re.IGNORECASE)
        if m:
            sa.lifetime_remaining = int(m.group(1))
        m = re.search(r"Remaining\s+\w*\s*Life\w*:?\s*(\d+)", block, re.IGNORECASE)
        if m:
            sa.lifetime_remaining = int(m.group(1))
        m = re.search(r"Local\s+(\d{1,3}(?:\.\d{1,3}){3})", block)
        if m:
            sa.local = m.group(1)
        if sa.state != "UNKNOWN" or sa.encr != "?":
            sas.append(sa)
    return sas


def parse_ios_ikev1(output: str) -> list[SA]:
    """`show crypto isakmp sa detail` - IKEv1, one line per SA."""
    sas: list[SA] = []
    for line in output.splitlines():
        m = re.match(r"\s*(\d{1,3}(?:\.\d{1,3}){3})\s+(\d{1,3}(?:\.\d{1,3}){3})\s+"
                     r"(\S+)\s+\S*\s*(\S+)", line)
        if m:
            sas.append(SA(peer=m.group(2), local=m.group(1), version="IKEv1",
                          state=m.group(3).upper()))
            continue
        m = re.search(r"IPv4 Crypto ISAKMP SA", line)
        if m:
            continue
    return sas


def parse_ios_ipsec(output: str) -> dict[str, tuple[int, int]]:
    out: dict[str, tuple[int, int]] = {}
    for block in re.split(r"(?=local\s+ident)", output)[1:]:
        m = re.search(r"current_peer\s+(\d{1,3}(?:\.\d{1,3}){3})", block)
        if not m:
            continue
        enc = re.search(r"#pkts encaps:\s*(\d+)", block)
        dec = re.search(r"#pkts decaps:\s*(\d+)", block)
        e, d = out.get(m.group(1), (0, 0))
        out[m.group(1)] = (e + (int(enc.group(1)) if enc else 0),
                           d + (int(dec.group(1)) if dec else 0))
    return out


def parse_junos_ike(output: str) -> list[SA]:
    sas: list[SA] = []
    cur: SA | None = None
    for line in output.splitlines():
        m = re.search(r"Remote Address:\s*(\d{1,3}(?:\.\d{1,3}){3})", line, re.IGNORECASE)
        if m:
            if cur:
                sas.append(cur)
            cur = SA(peer=m.group(1))
            continue
        if cur is None:
            continue
        m = re.search(r"State:\s*(\S+)", line, re.IGNORECASE)
        if m:
            cur.state = m.group(1).upper()
        m = re.search(r"Encryption\s*:\s*([\w-]+)", line, re.IGNORECASE)
        if m:
            cur.encr = m.group(1)
        m = re.search(r"Authentication\s*:\s*([\w-]+)", line, re.IGNORECASE)
        if m:
            cur.hash = m.group(1)
        m = re.search(r"Diffie-Hellman group\s*:\s*\D*(\d+)", line, re.IGNORECASE)
        if m:
            cur.dh_group = int(m.group(1))
        m = re.search(r"Lifetime:\s*Expires in (\d+) seconds", line, re.IGNORECASE)
        if m:
            cur.lifetime_remaining = int(m.group(1))
        m = re.search(r"IKE version\s*:?\s*IKEv(\d)", line, re.IGNORECASE)
        if m:
            cur.version = f"IKEv{m.group(1)}"
    if cur:
        sas.append(cur)
    return sas


def parse_panos_ike(output: str) -> list[SA]:
    sas: list[SA] = []
    for line in output.splitlines():
        ips = IPV4.findall(line)
        if not ips:
            continue
        state = re.search(r"\b(active|init|expired|down|established)\b", line, re.IGNORECASE)
        if not state:
            continue
        sa = SA(peer=ips[-1], state=state.group(1).upper())
        m = re.search(r"\b(aes-?\d*|3des|des|null)\b", line, re.IGNORECASE)
        if m:
            sa.encr = m.group(1)
        m = re.search(r"\b(sha-?\d*|md5)\b", line, re.IGNORECASE)
        if m:
            sa.hash = m.group(1)
        m = re.search(r"\bDH\s*(\d+)\b", line, re.IGNORECASE)
        if m:
            sa.dh_group = int(m.group(1))
        sas.append(sa)
    return sas


def parse_fortios_ike(output: str) -> list[SA]:
    """
    Parse `diagnose vpn ike status`.

    The original version of this parser had a dead branch: it assigned a match
    then immediately tested `if not m`, so the first pattern could never
    contribute anything. Both forms are handled here.
    """
    sas: list[SA] = []
    cur: SA | None = None
    for line in output.splitlines():
        m = re.search(r"remote-gateway:\s*(\d{1,3}(?:\.\d{1,3}){3})", line, re.IGNORECASE)
        if not m:
            m = re.search(r"\brem=(\d{1,3}(?:\.\d{1,3}){3})", line, re.IGNORECASE)
        if m:
            if cur:
                sas.append(cur)
            cur = SA(peer=m.group(1))
            v = re.search(r"ike-version[=:]\s*(\d)", line, re.IGNORECASE)
            if v:
                cur.version = f"IKEv{v.group(1)}"
            continue
        if cur is None:
            continue
        m = re.search(r"status[=:]\s*(\S+)", line, re.IGNORECASE)
        if m:
            cur.state = m.group(1).strip(",").upper()
        m = re.search(r"\b(aes\d*|3des|des|null)\b", line, re.IGNORECASE)
        if m and cur.encr == "?":
            cur.encr = m.group(1)
        m = re.search(r"\b(sha\d*|md5)\b", line, re.IGNORECASE)
        if m and cur.hash == "?":
            cur.hash = m.group(1)
        m = re.search(r"\bDH\s*group\s*(\d+)", line, re.IGNORECASE)
        if m:
            cur.dh_group = int(m.group(1))
    if cur:
        sas.append(cur)
    return sas


# ---------------------------------------------------------------------------
# Per-device analysis
# ---------------------------------------------------------------------------

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

    if os_key in NO_IKE:
        r["skipped"] = True
        r["error"] = NO_IKE[os_key]
        return r
    if os_key not in COMMAND_MAP:
        r["skipped"] = True
        r["error"] = f"IKE SA inspection 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 - a missing command must not lose the rest
                out[key] = ""

        if os_key in ("ios", "iosxe", "iosxr"):
            sas = parse_ios_ikev2(out.get("ikev2", ""))
            sas += parse_ios_ikev1(out.get("ikev1", ""))
            counters = parse_ios_ipsec(out.get("ipsec", ""))
            for sa in sas:
                if sa.peer in counters:
                    sa.encaps, sa.decaps = counters[sa.peer]
        elif os_key == "junos":
            sas = parse_junos_ike(out.get("ikev2", ""))
        elif os_key == "panos":
            sas = parse_panos_ike(out.get("ikev2", ""))
        else:
            sas = parse_fortios_ike(out.get("ikev2", ""))

        r["sas"] = [asdict(s) for s in sas]
        find: list[dict] = []

        def add(sev: str, peer: str, msg: str, fix: str = "") -> None:
            find.append({"severity": sev, "peer": peer, "message": msg, "fix": fix})

        for sa in sas:
            if sa.state not in HEALTHY_STATES:
                add("HIGH", sa.peer,
                    f"SA state is {sa.state}",
                    "IKE negotiation failed, or DPD tore the SA down. Check the "
                    "pre-shared key and that the proposals overlap")
                continue

            if sa.dh_group and sa.dh_group in WEAK_DH:
                add("HIGH", sa.peer,
                    f"DH group {sa.dh_group} is deprecated (RFC 8247)",
                    f"move to group {STRONG_DH}")

            if any(w in sa.encr.lower() for w in WEAK_ENCR):
                add("HIGH", sa.peer,
                    f"encryption '{sa.encr}' is weak",
                    "use AES-GCM-256, or AES-CBC-256 with a separate integrity "
                    "algorithm")

            if sa.hash != "?" and any(sa.hash.lower().startswith(w) for w in WEAK_HASH) \
                    and "256" not in sa.hash and "384" not in sa.hash \
                    and "512" not in sa.hash:
                add("MEDIUM", sa.peer,
                    f"integrity algorithm '{sa.hash}' is weak",
                    "use SHA-256 or better")

            if sa.version == "IKEv1":
                add("LOW", sa.peer,
                    "using IKEv1",
                    "IKEv2 (RFC 7296) negotiates faster, survives address changes, "
                    "and supports stronger algorithms. Migrate when the peer allows it")

            if (sa.lifetime_remaining is not None
                    and sa.lifetime_remaining < args.warn_lifetime):
                add("LOW", sa.peer,
                    f"SA rekeys in {sa.lifetime_remaining}s",
                    f"below the {args.warn_lifetime}s threshold - informational "
                    f"unless it repeats every poll, which means rekey is failing")

            if sa.encaps is not None and sa.decaps is not None:
                if sa.encaps > 0 and sa.decaps == 0:
                    add("HIGH", sa.peer,
                        f"SA is READY but one-way: {sa.encaps:,} encrypted, 0 decrypted",
                        "the peer is not replying - check its crypto ACL and return route")

        r["findings"] = find
        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


# ---------------------------------------------------------------------------
# Cross-device analysis
# ---------------------------------------------------------------------------

def find_asymmetry(results: list[dict]) -> list[str]:
    """
    Compare the two ends of each tunnel.

    When both peers of a tunnel are in the inventory, their SAs should agree.
    A different SA count, or a different proposal on each side, means one end
    renegotiated and the other did not - the classic DMVPN and FlexVPN failure
    where the tunnel works intermittently and each device reports itself as
    healthy.
    """
    issues: list[str] = []
    live = [r for r in results if not r["error"] and not r["skipped"]]
    by_host = {r["host"]: r for r in live}

    # Index every SA by (this device, its peer)
    seen: dict[tuple[str, str], dict] = {}
    for r in live:
        for sa in r["sas"]:
            seen[(r["host"], sa["peer"])] = sa

    checked: set[frozenset] = set()
    for (host, peer), sa in seen.items():
        other = by_host.get(peer)
        if not other:
            continue
        pair = frozenset((host, peer))
        if pair in checked:
            continue
        checked.add(pair)

        back = [s for s in other["sas"] if s["peer"] in (host, sa.get("local"))]
        if not back:
            issues.append(f"{host} has an SA to {peer}, but {peer} has none back - "
                          f"one side renegotiated and the other did not")
            continue

        b = back[0]
        if sa["dh_group"] and b["dh_group"] and sa["dh_group"] != b["dh_group"]:
            issues.append(f"{host} <-> {peer}: DH group differs "
                          f"({sa['dh_group']} vs {b['dh_group']})")
        if sa["encr"] != "?" and b["encr"] != "?" and sa["encr"].lower() != b["encr"].lower():
            issues.append(f"{host} <-> {peer}: encryption differs "
                          f"({sa['encr']} vs {b['encr']})")
        if sa["version"] != b["version"]:
            issues.append(f"{host} <-> {peer}: IKE version differs "
                          f"({sa['version']} vs {b['version']})")

    for r in live:
        counts = defaultdict(int)
        for sa in r["sas"]:
            counts[sa["peer"]] += 1
        for peer, n in counts.items():
            if n > 2:
                issues.append(f"{r['host']}: {n} SAs to the same peer {peer} - "
                              f"duplicate SAs usually mean rekey is failing and "
                              f"stale SAs are not being cleared")
    return issues


def summarise(r: dict) -> str:
    host, os_key = r["host"], r["os"]
    if r["skipped"]:
        return f"[SKIP]  {host:<24} {r['error']}"
    if r["error"]:
        return f"[ERROR] {host:<24} ({os_key}) {r['error']}"
    n = len(r["sas"])
    if not r["findings"]:
        return f"[OK]    {host:<24} ({os_key}) {n} SA(s), no issues"
    sev = {f["severity"] for f in r["findings"]}
    worst = "HIGH" if "HIGH" in sev else ("MEDIUM" if "MEDIUM" in sev else "LOW")
    return (f"[{worst:<6}] {host:<24} ({os_key}) {n} SA(s), "
            f"{len(r['findings'])} finding(s)")


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Deep-inspect IKE Phase 1 and Phase 2 security associations.",
        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 inspect")
    ap.add_argument("--warn-lifetime", type=int, default=1800,
                    help="Flag an SA rekeying in fewer seconds than this. Default: 1800")
    ap.add_argument("--min-severity", choices=["HIGH", "MEDIUM", "LOW"], default="LOW",
                    help="Only report findings at or above this severity. Default: LOW")
    ap.add_argument("--show-sas", action="store_true",
                    help="List every SA with its negotiated proposal")
    ap.add_argument("--limit", help="Only inspect 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=60,
                    help="Seconds to wait per command. Default: 60")
    ap.add_argument("--workers", type=int, default=10,
                    help="How many devices 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"Analysing IKE SAs on {len(devices)} device(s)...")
    print("=" * 86)

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

    rank = {"HIGH": 0, "MEDIUM": 1, "LOW": 2}
    floor = rank[args.min_severity]
    tally = defaultdict(int)

    for r in sorted(results, key=lambda x: x["host"]):
        print(summarise(r))
        if args.show_sas:
            for sa in sorted(r["sas"], key=lambda x: x["peer"]):
                life = (f"rekey {sa['lifetime_remaining']}s"
                        if sa["lifetime_remaining"] is not None else "")
                print(f"           {sa['peer']:<16} {sa['version']:<6} "
                      f"{sa['state']:<10} {sa['encr']}/{sa['hash']}/DH{sa['dh_group']} {life}")
        for f in sorted(r["findings"], key=lambda x: rank[x["severity"]]):
            if rank[f["severity"]] > floor:
                continue
            tally[f["severity"]] += 1
            print(f"           [{f['severity']:<6}] peer {f['peer']}: {f['message']}")
            if f["fix"]:
                print(f"                      -> {f['fix']}")

    asym = find_asymmetry(results)
    print("=" * 86)
    if asym:
        print(f"TUNNEL ASYMMETRY: {len(asym)} issue(s)")
        for a in asym:
            print(f"  ! {a}")
        print("\nBoth ends of a tunnel must agree. When they do not, the tunnel works\n"
              "intermittently and each device reports itself as healthy.")

    errs = sum(1 for r in results if r["error"] and not r["skipped"])
    parts = "  ".join(f"{k}={tally[k]}" for k in ("HIGH", "MEDIUM", "LOW") if tally[k])
    print(f"\n{parts or 'no findings'}   across {len(results)} device(s)"
          + (f", {errs} unreachable" if errs else ""))

    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": "*IKE SA alert*\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({"devices": results, "asymmetry": asym}, indent=2),
            encoding="utf-8")

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


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