Crypto Policy Checker

#!/usr/bin/env python3
"""
Crypto Policy Checker

READ THE DOCUMENTATION BEFORE YOU RUN THIS:
    https://permitany.com/scripts/crypto-policy-checker

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" "paramiko>=3.0" "cryptography>=42.0"
"""

from __future__ import annotations

import argparse
import fnmatch
import logging
import json
import os
import re
import socket
import ssl
import warnings
import sys
from concurrent.futures import ThreadPoolExecutor, as_completed
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

sys.stdout.reconfigure(encoding="utf-8", errors="replace")

# A refused algorithm is an expected, informative result here - paramiko logs
# it as an exception with a full traceback, which would bury the report.
logging.getLogger("paramiko").setLevel(logging.CRITICAL)
logging.getLogger("paramiko.transport").setLevel(logging.CRITICAL)

# Asking for SSLv3 on purpose, in order to prove the server refuses it, is not
# a deprecation the user needs to see on stderr.
warnings.filterwarnings("ignore", message=r".*ssl\.TLSVersion\..*deprecated.*")

try:
    import yaml
except ImportError:
    yaml = None

try:
    from cryptography import x509
    from cryptography.hazmat.primitives.asymmetric import ec, rsa, dsa
except ImportError:
    x509 = None

try:
    import paramiko
except ImportError:
    paramiko = None

try:
    from netmiko import ConnectHandler
    from netmiko.exceptions import (
        NetmikoAuthenticationException,
        NetmikoTimeoutException,
    )
except ImportError:
    ConnectHandler = None

    class NetmikoAuthenticationException(Exception):
        pass

    class NetmikoTimeoutException(Exception):
        pass


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


# ---------------------------------------------------------------------------
# Result model
#
# Three states, and the difference between the last two is the whole point of
# this script: PASS and FAIL are both verified facts. UNVERIFIED is the
# absence of a fact, and it is never reported as a finding.
# ---------------------------------------------------------------------------

PASS, FAIL, UNVERIFIED = "PASS", "FAIL", "UNVERIFIED"

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

DEFAULT_SEVERITY = {
    "tls.protocol": "CRITICAL",
    "tls.cipher": "HIGH",
    "tls.curve": "MEDIUM",
    "tls.dh_bits": "HIGH",
    "ssh.kex": "HIGH",
    "ssh.cipher": "HIGH",
    "ssh.mac": "MEDIUM",
    "ssh.hostkey": "MEDIUM",
    "ssh.hostkey_bits": "HIGH",
    "certificate.expiry": "CRITICAL",
    "certificate.expiry_warning": "MEDIUM",
    "certificate.signature": "HIGH",
    "certificate.key_bits": "HIGH",
    "certificate.curve": "MEDIUM",
    "certificate.san": "LOW",
    "ipsec.encryption": "CRITICAL",
    "ipsec.integrity": "HIGH",
    "ipsec.dh_group": "HIGH",
    "ipsec.ike_version": "MEDIUM",
    "ipsec.pfs": "MEDIUM",
}


@dataclass
class Check:
    target: str
    surface: str          # tls | ssh | certificate | ipsec
    rule: str             # e.g. tls.protocol
    status: str           # PASS | FAIL | UNVERIFIED
    detail: str
    severity: str = "LOW"
    context: str = ""     # the object the finding is about, where there is one
    reason: str = ""      # for UNVERIFIED: WHY it could not be determined


@dataclass
class TargetResult:
    target: str
    kind: str
    status: str = "ok"            # ok | error
    error: str = ""
    checks: list[Check] = field(default_factory=list)


# ---------------------------------------------------------------------------
# The policy
# ---------------------------------------------------------------------------

EXAMPLE_POLICY = """\
# crypto-policy.yaml - the policy is YOURS. Nothing below is built in.
version: 1
name: Corporate crypto baseline

tls:
  allowed_protocols:   [TLSv1.2, TLSv1.3]
  forbidden_protocols: [SSLv3, TLSv1, TLSv1.1]
  forbidden_ciphers:   ["*RC4*", "*3DES*", "*NULL*", "*_CBC_SHA"]
  allowed_curves:      [X25519, secp256r1, secp384r1]
  min_dh_bits:         2048

ssh:
  forbidden_kex:          ["diffie-hellman-group1-sha1",
                           "diffie-hellman-group14-sha1",
                           "diffie-hellman-group-exchange-sha1"]
  forbidden_ciphers:      ["3des-cbc", "aes128-cbc", "aes256-cbc", "arcfour*"]
  forbidden_macs:         ["hmac-md5*", "hmac-sha1", "hmac-sha1-96"]
  allowed_hostkey_types:  [ssh-ed25519, rsa-sha2-512, rsa-sha2-256,
                           ecdsa-sha2-nistp256, ecdsa-sha2-nistp384]
  min_rsa_hostkey_bits:   3072

certificate:
  min_days_remaining:  14
  warn_days_remaining: 45
  forbidden_signature_algorithms: ["md5*", "sha1*"]
  min_rsa_key_bits: 2048
  min_ec_key_bits:  256
  allowed_curves:   [secp256r1, secp384r1]
  require_san:      true

ipsec:
  # A mode suffix is stripped before matching, so "3des" also catches
  # "3des-cbc". Platform spelling is not: IOS writes `hash sha` for SHA-1,
  # so "sha" has to be listed as well as "sha1".
  forbidden_encryption: [des, 3des, "*null*"]
  forbidden_integrity:  [md5, sha, sha1]
  forbidden_dh_groups:  [1, 2, 5, 22, 23, 24]
  min_dh_group:         14
  forbidden_ike_versions: [1]
  require_pfs: true

# Optional. Any rule name from the output can be given your own severity.
severity:
  tls.protocol: CRITICAL
  certificate.expiry: CRITICAL
  ipsec.ike_version: LOW
"""


def load_policy(path: str) -> dict:
    p = Path(path)
    if not p.exists():
        raise SystemExit(
            f"Policy file not found: {path}\n\n"
            f"This script has no built-in rules on purpose - the policy is yours.\n"
            f"Write a starter one with:\n\n"
            f"    python crypto_policy_checker.py --write-example-policy {path}\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 not isinstance(data, dict):
        raise SystemExit(f"{path} must be a mapping with tls/ssh/certificate/ipsec keys.")

    known = {"version", "name", "tls", "ssh", "certificate", "ipsec", "severity"}
    for key in data:
        if key not in known:
            raise SystemExit(
                f"Unknown top-level key '{key}' in {path}.\n"
                f"Valid keys: {', '.join(sorted(known))}\n"
                f"A typo here would silently disable a whole section, so it is an error."
            )
    for sec in ("tls", "ssh", "certificate", "ipsec"):
        if sec in data and not isinstance(data[sec], dict):
            raise SystemExit(f"Section '{sec}' in {path} must be a mapping.")
    for name, sev in (data.get("severity") or {}).items():
        if str(sev).upper() not in SEV_RANK:
            raise SystemExit(f"severity.{name} is '{sev}'. Use CRITICAL, HIGH, "
                             f"MEDIUM or LOW.")
    if not any(data.get(s) for s in ("tls", "ssh", "certificate", "ipsec")):
        raise SystemExit(f"{path} has no tls, ssh, certificate or ipsec section - "
                         f"there is nothing to check.")
    return data


def severity_for(policy: dict, rule: str) -> str:
    override = (policy.get("severity") or {}).get(rule)
    if override:
        return str(override).upper()
    return DEFAULT_SEVERITY.get(rule, "MEDIUM")


# A mode suffix is not a different algorithm. Junos writes "3des-cbc" where
# IOS writes "3des", and a policy that forbids "3des" must catch both - the
# first version of this script reported 3des-cbc as PASS, which is worse than
# missing it, because PASS means "verified good".
MODE_SUFFIX = re.compile(r"-(cbc|ctr|gcm|ccm|hmac|96)$")


def algo_variants(value: str) -> list[str]:
    """The value itself, plus the same token without its mode suffix."""
    v = str(value).lower()
    out = [v]
    while True:
        stripped = MODE_SUFFIX.sub("", v)
        if stripped == v:
            break
        v = stripped
        out.append(v)
    return out


def matches_any(value: str, patterns) -> str | None:
    """Return the pattern that matched, case-insensitively, or None."""
    variants = algo_variants(value)
    for pat in patterns or []:
        p = str(pat).lower()
        for v in variants:
            if fnmatch.fnmatch(v, p):
                return str(pat)
    return None


# ---------------------------------------------------------------------------
# TLS
# ---------------------------------------------------------------------------

TLS_VERSIONS = {
    "SSLv3": getattr(ssl.TLSVersion, "SSLv3", None),
    "TLSv1": ssl.TLSVersion.TLSv1,
    "TLSv1.1": ssl.TLSVersion.TLSv1_1,
    "TLSv1.2": ssl.TLSVersion.TLSv1_2,
    "TLSv1.3": ssl.TLSVersion.TLSv1_3,
}


def _tls_context(min_v=None, max_v=None) -> ssl.SSLContext:
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    try:
        ctx.set_ciphers("ALL:@SECLEVEL=0")
    except ssl.SSLError:
        try:
            ctx.set_ciphers("ALL")
        except ssl.SSLError:
            pass
    if min_v is not None:
        ctx.minimum_version = min_v
    if max_v is not None:
        ctx.maximum_version = max_v
    return ctx


def tls_probe_version(host: str, port: int, name: str, timeout: float) -> tuple[str, str]:
    """
    Try a handshake pinned to exactly one protocol version.

    Returns (status, detail) where status is "accepted", "refused" or
    "cannot-offer". "cannot-offer" is the one that matters: if the local
    OpenSSL build will not speak TLSv1.0 at all, a failed handshake proves
    NOTHING about the server. Reporting that as "server rejects TLSv1.0"
    would be inventing a pass.
    """
    version = TLS_VERSIONS.get(name)
    if version is None:
        return "cannot-offer", f"this Python/OpenSSL build has no {name} support"
    try:
        ctx = _tls_context(min_v=version, max_v=version)
    except (ValueError, ssl.SSLError) as exc:
        return "cannot-offer", f"local OpenSSL refuses to offer {name} ({exc})"

    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with ctx.wrap_socket(sock, server_hostname=host) as tls:
                return "accepted", tls.version() or name
    except ssl.SSLError as exc:
        text = str(exc)
        if "no protocols available" in text or "unsupported protocol" in text.lower():
            return "cannot-offer", f"local OpenSSL refuses to offer {name}"
        return "refused", text.splitlines()[0][:120]
    except (socket.timeout, TimeoutError):
        return "cannot-offer", "the handshake timed out - inconclusive, not a pass"
    except OSError as exc:
        return "cannot-offer", f"connection failed ({exc})"


def check_tls(target: dict, policy: dict, timeout: float) -> TargetResult:
    host = target["host"]
    port = int(target.get("port", 443))
    label = target.get("name") or f"{host}:{port}"
    res = TargetResult(target=label, kind="tls")
    tls_pol = policy.get("tls") or {}
    cert_pol = policy.get("certificate") or {}

    def add(rule, status, detail, context="", reason=""):
        res.checks.append(Check(label, "tls" if rule.startswith("tls") else "certificate",
                                rule, status, detail,
                                severity_for(policy, rule), context, reason))

    # 1. Baseline handshake - gives the negotiated suite and the certificate.
    sni = target.get("server_name") or host
    try:
        with socket.create_connection((host, port), timeout=timeout) as sock:
            with _tls_context().wrap_socket(sock, server_hostname=sni) as tls:
                negotiated = tls.version()
                cipher = tls.cipher()
                der = tls.getpeercert(binary_form=True)
                shared = tls.shared_ciphers()
    except (OSError, ssl.SSLError) as exc:
        res.status = "error"
        res.error = f"{type(exc).__name__}: {exc}"
        return res

    if cipher:
        name, _proto, bits = cipher
        pat = matches_any(name, tls_pol.get("forbidden_ciphers"))
        if pat:
            add("tls.cipher", FAIL,
                f"negotiated cipher {name} matches forbidden pattern '{pat}'")
        else:
            add("tls.cipher", PASS, f"negotiated cipher {name}")
    else:
        add("tls.cipher", UNVERIFIED, "no cipher reported",
            reason="the TLS library did not expose the negotiated suite")

    # 2. Protocol versions, one pinned handshake each.
    wanted = list(tls_pol.get("forbidden_protocols") or []) + \
        [v for v in TLS_VERSIONS if v not in (tls_pol.get("allowed_protocols") or [])
         and v not in (tls_pol.get("forbidden_protocols") or [])]
    seen = set()
    for name in wanted:
        if name in seen:
            continue
        seen.add(name)
        forbidden = name in (tls_pol.get("forbidden_protocols") or [])
        status, detail = tls_probe_version(host, port, name, timeout)
        if status == "cannot-offer":
            if forbidden:
                add("tls.protocol", UNVERIFIED, f"{name} could not be tested",
                    context=name, reason=detail)
            continue
        if status == "accepted":
            if forbidden:
                add("tls.protocol", FAIL, f"{name} is accepted and the policy "
                    f"forbids it", context=name)
            elif tls_pol.get("allowed_protocols") and \
                    name not in tls_pol["allowed_protocols"]:
                add("tls.protocol", FAIL, f"{name} is accepted and is not in "
                    f"allowed_protocols", context=name)
            else:
                add("tls.protocol", PASS, f"{name} accepted", context=name)
        elif forbidden:
            add("tls.protocol", PASS, f"{name} is refused", context=name)

    if negotiated and tls_pol.get("allowed_protocols") and \
            negotiated not in tls_pol["allowed_protocols"]:
        add("tls.protocol", FAIL,
            f"the default handshake negotiated {negotiated}, which is not in "
            f"allowed_protocols", context=negotiated)

    # 3. Key exchange group. Python's ssl module does not expose it at all, so
    #    this is reported as unknown rather than quietly skipped.
    if tls_pol.get("allowed_curves") or tls_pol.get("min_dh_bits"):
        add("tls.curve", UNVERIFIED, "key exchange group not read",
            reason="Python's ssl module does not expose the negotiated group; "
                   "use `openssl s_client -connect host:port` and read "
                   "'Server Temp Key'")

    # 4. The certificate.
    res.checks.extend(check_certificate(der, label, cert_pol, policy))
    return res


# ---------------------------------------------------------------------------
# Certificates
# ---------------------------------------------------------------------------

def _pubkey_facts(cert) -> tuple[str, int | None, str | None]:
    key = cert.public_key()
    if isinstance(key, rsa.RSAPublicKey):
        return "RSA", key.key_size, None
    if isinstance(key, dsa.DSAPublicKey):
        return "DSA", key.key_size, None
    if isinstance(key, ec.EllipticCurvePublicKey):
        return "EC", key.curve.key_size, key.curve.name
    return type(key).__name__, getattr(key, "key_size", None), None


CURVE_ALIAS = {
    "secp256r1": {"secp256r1", "prime256v1", "p-256", "nistp256"},
    "secp384r1": {"secp384r1", "p-384", "nistp384"},
    "secp521r1": {"secp521r1", "p-521", "nistp521"},
}


def _curve_allowed(curve: str, allowed) -> bool:
    c = curve.lower()
    for a in allowed:
        a = str(a).lower()
        if c == a or c in CURVE_ALIAS.get(a, set()) or a in CURVE_ALIAS.get(c, set()):
            return True
    return False


def check_certificate(der: bytes | None, label: str, cert_pol: dict,
                      policy: dict) -> list[Check]:
    out: list[Check] = []

    def add(rule, status, detail, context="", reason=""):
        out.append(Check(label, "certificate", rule, status, detail,
                         severity_for(policy, rule), context, reason))

    if not cert_pol:
        return out
    if not der:
        add("certificate.expiry", UNVERIFIED, "no certificate was presented",
            reason="the peer sent no certificate on this handshake")
        return out
    if x509 is None:
        add("certificate.expiry", UNVERIFIED, "certificate not parsed",
            reason="the `cryptography` library is not installed - "
                   "pip install cryptography")
        return out

    try:
        cert = x509.load_der_x509_certificate(der)
    except Exception as exc:  # noqa: BLE001 - a malformed cert is not a finding
        add("certificate.expiry", UNVERIFIED, "certificate could not be parsed",
            reason=f"{type(exc).__name__}: {exc}")
        return out

    subject = cert.subject.rfc4514_string()

    # Expiry
    try:
        not_after = cert.not_valid_after_utc
    except AttributeError:
        not_after = cert.not_valid_after.replace(tzinfo=timezone.utc)
    days = (not_after - datetime.now(timezone.utc)).days
    min_days = cert_pol.get("min_days_remaining")
    warn_days = cert_pol.get("warn_days_remaining")
    if min_days is not None and days < int(min_days):
        word = "expired" if days < 0 else "expires"
        add("certificate.expiry", FAIL,
            f"{word} in {days} day(s) ({not_after:%Y-%m-%d}), policy minimum is "
            f"{min_days}", context=subject)
    elif warn_days is not None and days < int(warn_days):
        add("certificate.expiry_warning", FAIL,
            f"expires in {days} day(s) ({not_after:%Y-%m-%d}), inside the "
            f"{warn_days}-day warning window", context=subject)
    elif min_days is not None or warn_days is not None:
        add("certificate.expiry", PASS,
            f"valid for {days} more day(s) (until {not_after:%Y-%m-%d})",
            context=subject)

    # Signature algorithm
    if cert_pol.get("forbidden_signature_algorithms"):
        try:
            sig = cert.signature_algorithm_oid._name
        except AttributeError:
            sig = None
        if not sig or sig == "Unknown OID":
            add("certificate.signature", UNVERIFIED, "signature algorithm unknown",
                context=subject,
                reason="the OID is not one this version of `cryptography` names")
        else:
            pat = matches_any(sig, cert_pol["forbidden_signature_algorithms"])
            if pat:
                add("certificate.signature", FAIL,
                    f"signed with {sig}, which matches forbidden pattern '{pat}'",
                    context=subject)
            else:
                add("certificate.signature", PASS, f"signed with {sig}",
                    context=subject)

    # Key size and curve
    try:
        kind, bits, curve = _pubkey_facts(cert)
    except Exception as exc:  # noqa: BLE001
        kind, bits, curve = None, None, None
        add("certificate.key_bits", UNVERIFIED, "public key not readable",
            context=subject, reason=f"{type(exc).__name__}: {exc}")

    if kind in ("RSA", "DSA") and cert_pol.get("min_rsa_key_bits"):
        floor = int(cert_pol["min_rsa_key_bits"])
        if bits is None:
            add("certificate.key_bits", UNVERIFIED, f"{kind} key size not readable",
                context=subject, reason="the library did not report a key size")
        elif bits < floor:
            add("certificate.key_bits", FAIL,
                f"{kind} key is {bits} bits, policy minimum is {floor}",
                context=subject)
        else:
            add("certificate.key_bits", PASS, f"{kind} {bits}-bit key",
                context=subject)
    if kind == "EC":
        if cert_pol.get("min_ec_key_bits"):
            floor = int(cert_pol["min_ec_key_bits"])
            if bits is None:
                add("certificate.key_bits", UNVERIFIED, "EC key size not readable",
                    context=subject, reason="the library did not report a key size")
            elif bits < floor:
                add("certificate.key_bits", FAIL,
                    f"EC key is {bits} bits, policy minimum is {floor}",
                    context=subject)
            else:
                add("certificate.key_bits", PASS, f"EC {bits}-bit key",
                    context=subject)
        if cert_pol.get("allowed_curves"):
            if not curve:
                add("certificate.curve", UNVERIFIED, "curve not readable",
                    context=subject, reason="the library did not name the curve")
            elif not _curve_allowed(curve, cert_pol["allowed_curves"]):
                add("certificate.curve", FAIL,
                    f"curve {curve} is not in allowed_curves", context=subject)
            else:
                add("certificate.curve", PASS, f"curve {curve}", context=subject)

    # SAN
    if cert_pol.get("require_san"):
        try:
            ext = cert.extensions.get_extension_for_class(x509.SubjectAlternativeName)
            names = ext.value.get_values_for_type(x509.DNSName) + \
                [str(v) for v in ext.value.get_values_for_type(x509.IPAddress)]
            if names:
                add("certificate.san", PASS,
                    f"subjectAltName present ({', '.join(names[:3])})", context=subject)
            else:
                add("certificate.san", FAIL,
                    "subjectAltName is present but empty", context=subject)
        except x509.ExtensionNotFound:
            add("certificate.san", FAIL,
                "no subjectAltName - modern clients reject CN-only certificates",
                context=subject)
    return out


# ---------------------------------------------------------------------------
# SSH
# ---------------------------------------------------------------------------

def _ssh_offers(host: str, port: int, kind: str, candidate: str,
                timeout: float) -> tuple[str, str]:
    """
    Ask the server whether it will negotiate exactly ONE algorithm.

    We restrict our own proposal to the single candidate. If the handshake
    reaches the point where keys are exchanged, the server supports it. No
    authentication is attempted and no credentials are used.

    Returns (status, detail): "offered", "not-offered" or "cannot-test".
    """
    if paramiko is None:
        return "cannot-test", "paramiko is not installed"
    sock = None
    transport = None
    try:
        sock = socket.create_connection((host, port), timeout=timeout)
        transport = paramiko.Transport(sock)
        opts = transport.get_security_options()
        attr = {"kex": "kex", "cipher": "ciphers", "mac": "digests",
                "hostkey": "key_types"}[kind]
        supported = set(getattr(opts, attr))
        if candidate not in supported:
            return "cannot-test", (f"this paramiko build cannot speak "
                                   f"{candidate}, so the server was never asked")
        setattr(opts, attr, (candidate,))
        transport.start_client(timeout=timeout)
        return "offered", candidate
    except paramiko.SSHException as exc:
        text = str(exc).lower()
        if "incompatible" in text or "no acceptable" in text or "negotiat" in text:
            return "not-offered", candidate
        return "cannot-test", f"{type(exc).__name__}: {exc}"
    except (OSError, EOFError) as exc:
        return "cannot-test", f"{type(exc).__name__}: {exc}"
    finally:
        if transport is not None:
            try:
                transport.close()
            except OSError:
                pass
        if sock is not None:
            try:
                sock.close()
            except OSError:
                pass


def check_ssh(target: dict, policy: dict, timeout: float) -> TargetResult:
    host = target["host"]
    port = int(target.get("port", 22))
    label = target.get("name") or f"{host}:{port}"
    res = TargetResult(target=label, kind="ssh")
    pol = policy.get("ssh") or {}

    def add(rule, status, detail, context="", reason=""):
        res.checks.append(Check(label, "ssh", rule, status, detail,
                                severity_for(policy, rule), context, reason))

    if paramiko is None:
        res.status = "error"
        res.error = ("paramiko is not installed - it ships with netmiko, so this "
                     "means the install is incomplete")
        return res

    # Baseline handshake: banner and host key.
    sock = None
    transport = None
    try:
        sock = socket.create_connection((host, port), timeout=timeout)
        transport = paramiko.Transport(sock)
        transport.start_client(timeout=timeout)
        banner = transport.remote_version
        hostkey = transport.get_remote_server_key()
    except (OSError, EOFError, paramiko.SSHException) as exc:
        res.status = "error"
        res.error = f"{type(exc).__name__}: {exc}"
        return res
    finally:
        if transport is not None:
            try:
                transport.close()
            except OSError:
                pass
        if sock is not None:
            try:
                sock.close()
            except OSError:
                pass

    if pol.get("allowed_hostkey_types"):
        name = hostkey.get_name()
        if name in [str(x) for x in pol["allowed_hostkey_types"]]:
            add("ssh.hostkey", PASS, f"host key type {name} ({banner})")
        else:
            add("ssh.hostkey", FAIL,
                f"host key type {name} is not in allowed_hostkey_types")

    if pol.get("min_rsa_hostkey_bits") and hostkey.get_name().startswith("ssh-rsa"):
        bits = hostkey.get_bits()
        floor = int(pol["min_rsa_hostkey_bits"])
        if not bits:
            add("ssh.hostkey_bits", UNVERIFIED, "RSA host key size not reported",
                reason="paramiko did not expose the modulus size")
        elif bits < floor:
            add("ssh.hostkey_bits", FAIL,
                f"RSA host key is {bits} bits, policy minimum is {floor}")
        else:
            add("ssh.hostkey_bits", PASS, f"RSA host key is {bits} bits")

    for kind, key, rule in (("kex", "forbidden_kex", "ssh.kex"),
                            ("cipher", "forbidden_ciphers", "ssh.cipher"),
                            ("mac", "forbidden_macs", "ssh.mac")):
        for candidate in pol.get(key) or []:
            if any(ch in str(candidate) for ch in "*?["):
                add(rule, UNVERIFIED, f"'{candidate}' not tested", context=str(candidate),
                    reason="SSH algorithms are probed one exact name at a time, so "
                           "a wildcard cannot be offered - list the names in full")
                continue
            status, detail = _ssh_offers(host, port, kind, str(candidate), timeout)
            if status == "offered":
                add(rule, FAIL, f"server accepts {candidate}, which the policy forbids",
                    context=str(candidate))
            elif status == "not-offered":
                add(rule, PASS, f"server refuses {candidate}", context=str(candidate))
            else:
                add(rule, UNVERIFIED, f"'{candidate}' could not be tested",
                    context=str(candidate), reason=detail)
    return res


# ---------------------------------------------------------------------------
# IPsec, from the device configuration
#
# There are no built-in IPsec rules here. The tokens below are extracted and
# then compared against YOUR policy - nothing is weak or strong until your
# policy says so. For fixed NIST SP 800-131A / RFC 8247 rules, use
# `vpn-crypto-audit` instead; this script answers a different question.
# ---------------------------------------------------------------------------

IPSEC_CMD = {
    "ios": "show running-config | section crypto",
    "iosxe": "show running-config | section crypto",
    "iosxr": "show running-config formal | include ipsec|ikev2",
    "nxos": "show running-config | section crypto",
    "eos": "show running-config | section ^crypto",
    "junos": "show configuration security | display set",
    "fortios": "show vpn ipsec phase1-interface\nshow vpn ipsec phase2-interface",
    "panos": 'show config running xpath "/config/devices/entry/network/ike"',
}


@dataclass
class CryptoFact:
    context: str
    attribute: str          # encryption | integrity | dh_group | ike_version | pfs
    value: str


def extract_ipsec_facts(os_key: str, text: str) -> list[CryptoFact]:
    """Turn raw crypto configuration into (object, attribute, value) tuples."""
    facts: list[CryptoFact] = []
    ctx = "?"

    if os_key in ("ios", "iosxe", "iosxr", "nxos", "eos"):
        for line in text.splitlines():
            s = line.strip()
            m = re.match(r"crypto (isakmp policy \S+|ikev2 proposal \S+|"
                         r"ipsec transform-set \S+|map \S+ \d+)", s)
            if m:
                ctx = m.group(1)
                continue
            m = re.match(r"(encryption|encr)\s+(\S+)", s)
            if m:
                facts.append(CryptoFact(ctx, "encryption", m.group(2)))
            m = re.match(r"(?:hash|integrity)\s+(\S+)", s)
            if m:
                facts.append(CryptoFact(ctx, "integrity", m.group(1)))
            m = re.match(r"group\s+(\d+)", s)
            if m:
                facts.append(CryptoFact(ctx, "dh_group", m.group(1)))
            m = re.match(r"crypto ipsec transform-set (\S+)\s+(.+)", s)
            if m:
                ctx = f"transform-set {m.group(1)}"
                for tok in m.group(2).split():
                    low = tok.lower()
                    if low.startswith("esp-") and ("hmac" in low or "md5" in low
                                                   or "sha" in low):
                        facts.append(CryptoFact(ctx, "integrity",
                                                low.replace("esp-", "").replace("-hmac", "")))
                    elif low.startswith("esp-"):
                        facts.append(CryptoFact(ctx, "encryption",
                                                low.replace("esp-", "")))
            if re.search(r"set pfs group(\d+)", s):
                facts.append(CryptoFact(ctx, "pfs",
                                        re.search(r"set pfs group(\d+)", s).group(1)))

    elif os_key == "junos":
        for line in text.splitlines():
            m = re.search(r"security (?:ike|ipsec) proposal (\S+) "
                          r"(encryption-algorithm|authentication-algorithm|dh-group) (\S+)",
                          line)
            if m:
                attr = {"encryption-algorithm": "encryption",
                        "authentication-algorithm": "integrity",
                        "dh-group": "dh_group"}[m.group(2)]
                val = m.group(3)
                if attr == "dh_group":
                    val = re.sub(r"^group", "", val)
                facts.append(CryptoFact(f"proposal {m.group(1)}", attr, val))
            m = re.search(r"security ike gateway (\S+) version (\S+)", line)
            if m:
                v = "1" if "v1" in m.group(2).lower() else "2"
                facts.append(CryptoFact(f"gateway {m.group(1)}", "ike_version", v))
            m = re.search(r"security ipsec (?:vpn|policy) (\S+) .*perfect-forward-secrecy "
                          r"keys group(\S+)", line)
            if m:
                facts.append(CryptoFact(f"ipsec {m.group(1)}", "pfs", m.group(2)))

    elif os_key == "fortios":
        for line in text.splitlines():
            s = line.strip()
            m = re.match(r'edit "?([^"]+)"?$', s)
            if m:
                ctx = m.group(1)
                continue
            m = re.match(r"set proposal (.+)", s)
            if m:
                for tok in m.group(1).split():
                    parts = tok.split("-")
                    if len(parts) >= 2:
                        facts.append(CryptoFact(ctx, "encryption", "-".join(parts[:-1])))
                        facts.append(CryptoFact(ctx, "integrity", parts[-1]))
            m = re.match(r"set dhgrp (.+)", s)
            if m:
                for g in m.group(1).split():
                    facts.append(CryptoFact(ctx, "dh_group", g))
            m = re.match(r"set ike-version (\d)", s)
            if m:
                facts.append(CryptoFact(ctx, "ike_version", m.group(1)))
            m = re.match(r"set pfs (\S+)", s)
            if m:
                facts.append(CryptoFact(ctx, "pfs",
                                        "none" if m.group(1) == "disable" else "enable"))

    elif os_key == "panos":
        for m in re.finditer(r'<entry name="([^"]+)">(.*?)</entry>', text, re.S):
            name, body = m.group(1), m.group(2)
            for tag, attr in (("encryption", "encryption"), ("hash", "integrity"),
                              ("dh-group", "dh_group")):
                for mm in re.finditer(rf"<{tag}>(.*?)</{tag}>", body, re.S):
                    for v in re.findall(r"<member>([^<]+)</member>", mm.group(1)) \
                            or [mm.group(1).strip()]:
                        val = re.sub(r"^group", "", v.strip())
                        facts.append(CryptoFact(name, attr, val))
    return facts


def _norm_dh(value: str) -> int | None:
    m = re.search(r"(\d+)", str(value))
    return int(m.group(1)) if m else None


def evaluate_ipsec(label: str, facts: list[CryptoFact], policy: dict) -> list[Check]:
    out: list[Check] = []
    pol = policy.get("ipsec") or {}
    if not pol:
        return out

    def add(rule, status, detail, context="", reason=""):
        out.append(Check(label, "ipsec", rule, status, detail,
                         severity_for(policy, rule), context, reason))

    if not facts:
        add("ipsec.encryption", UNVERIFIED, "no crypto configuration was parsed",
            reason="the command returned nothing this script could parse - the "
                   "device may have no IPsec configured, or the syntax may differ. "
                   "Re-run with --save-raw and look at the file before assuming "
                   "this device is clean")
        return out

    for f in facts:
        if f.attribute == "encryption" and pol.get("forbidden_encryption"):
            pat = matches_any(f.value, pol["forbidden_encryption"])
            if pat:
                add("ipsec.encryption", FAIL,
                    f"encryption {f.value} matches forbidden '{pat}'", context=f.context)
            else:
                add("ipsec.encryption", PASS, f"encryption {f.value}", context=f.context)
        elif f.attribute == "integrity" and pol.get("forbidden_integrity"):
            pat = matches_any(f.value, pol["forbidden_integrity"])
            if pat:
                add("ipsec.integrity", FAIL,
                    f"integrity {f.value} matches forbidden '{pat}'", context=f.context)
            else:
                add("ipsec.integrity", PASS, f"integrity {f.value}", context=f.context)
        elif f.attribute == "dh_group":
            g = _norm_dh(f.value)
            if g is None:
                add("ipsec.dh_group", UNVERIFIED, f"DH group '{f.value}' not a number",
                    context=f.context, reason="the value could not be read as a group")
                continue
            bad = pol.get("forbidden_dh_groups") and g in [int(x) for x in
                                                           pol["forbidden_dh_groups"]]
            low = pol.get("min_dh_group") and g < int(pol["min_dh_group"])
            if bad:
                add("ipsec.dh_group", FAIL, f"DH group {g} is in forbidden_dh_groups",
                    context=f.context)
            elif low:
                add("ipsec.dh_group", FAIL,
                    f"DH group {g} is below min_dh_group {pol['min_dh_group']}",
                    context=f.context)
            elif pol.get("forbidden_dh_groups") or pol.get("min_dh_group"):
                add("ipsec.dh_group", PASS, f"DH group {g}", context=f.context)
        elif f.attribute == "ike_version" and pol.get("forbidden_ike_versions"):
            if f.value in [str(v) for v in pol["forbidden_ike_versions"]]:
                add("ipsec.ike_version", FAIL, f"IKEv{f.value} is forbidden",
                    context=f.context)
            else:
                add("ipsec.ike_version", PASS, f"IKEv{f.value}", context=f.context)
        elif f.attribute == "pfs" and pol.get("require_pfs"):
            if str(f.value).lower() in ("none", "disable", "0"):
                add("ipsec.pfs", FAIL, "Perfect Forward Secrecy is disabled",
                    context=f.context)
            else:
                add("ipsec.pfs", PASS, f"PFS group {f.value}", context=f.context)

    if pol.get("require_pfs") and not any(f.attribute == "pfs" for f in facts):
        add("ipsec.pfs", UNVERIFIED, "PFS setting not found in the configuration",
            reason="no pfs keyword appeared - absence in a config dump is not proof "
                   "that PFS is off, because some platforms do not print defaults")
    return out


def check_device(dev: dict, policy: dict, read_timeout: int,
                 save_raw: str | None) -> TargetResult:
    label = dev.get("device_name") or dev["host"]
    res = TargetResult(target=f"{label} ({dev['os']})", kind="ipsec")
    if ConnectHandler is None:
        res.status = "error"
        res.error = "netmiko is not installed - pip install netmiko"
        return res
    conn = None
    try:
        conn = connect(dev)
        chunks = []
        for cmd in IPSEC_CMD[dev["os"]].split("\n"):
            chunks.append(conn.send_command(cmd, read_timeout=read_timeout))
        raw = "\n".join(chunks)
        if save_raw:
            Path(save_raw).mkdir(parents=True, exist_ok=True)
            safe = re.sub(r"[^A-Za-z0-9._-]", "_", dev["host"])
            (Path(save_raw) / f"{safe}.txt").write_text(raw, encoding="utf-8")
        res.checks = evaluate_ipsec(res.target, extract_ipsec_facts(dev["os"], raw),
                                    policy)
    except NetmikoAuthenticationException as exc:
        res.status = "error"
        res.error = f"authentication failed - check NET_USER/NET_PASS ({exc})"
    except NetmikoTimeoutException as exc:
        res.status = "error"
        res.error = f"unreachable or SSH timed out ({exc})"
    except Exception as exc:  # noqa: BLE001 - one device must not abort the run
        res.status = "error"
        res.error = f"{type(exc).__name__}: {exc}"
    finally:
        if conn:
            try:
                conn.disconnect()
            except Exception:  # noqa: BLE001
                pass
    return res


# ---------------------------------------------------------------------------
# Targets file
# ---------------------------------------------------------------------------

def load_targets(path: str) -> list[dict]:
    p = Path(path)
    if not p.exists():
        raise SystemExit(
            f"Targets file not found: {path}\n"
            f"Create it with:\n\n"
            f"  targets:\n"
            f"    - host: www.example.com\n"
            f"      kind: tls\n"
            f"      port: 443\n"
        )
    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(p.read_text(encoding="utf-8"))
        else:
            data = json.loads(p.read_text(encoding="utf-8"))
    except SystemExit:
        raise
    except Exception as exc:
        raise SystemExit(f"Could not parse {path}: {exc}")

    if isinstance(data, dict):
        defaults = data.get("defaults", {}) or {}
        items = data.get("targets", []) or []
    else:
        defaults, items = {}, data or []
    if not items:
        raise SystemExit(f"No targets found in {path}. The file needs a `targets:` list.")

    out = []
    for i, t in enumerate(items, 1):
        if not isinstance(t, dict):
            raise SystemExit(f"Target #{i} in {path} is not a mapping.")
        merged = {**defaults, **t}
        if "host" not in merged:
            raise SystemExit(f"Target #{i} in {path} has no `host:` field.")
        kind = str(merged.get("kind", "tls")).lower()
        if kind not in ("tls", "ssh"):
            raise SystemExit(f"Target '{merged['host']}' has kind '{kind}'. "
                             f"Valid values: tls, ssh.")
        merged["kind"] = kind
        merged.setdefault("port", 443 if kind == "tls" else 22)
        out.append(merged)
    return out


# ---------------------------------------------------------------------------
# Reporting
# ---------------------------------------------------------------------------

def report(results: list[TargetResult], policy: dict, args) -> tuple[dict, int, int]:
    floor = SEV_RANK[args.min_severity]
    tally = {k: 0 for k in SEV_RANK}
    unverified = 0
    unreachable = 0

    for r in sorted(results, key=lambda x: (x.kind, x.target)):
        if r.status == "error":
            unreachable += 1
            print()
            print(f"[ERROR     ] {r.target}")
            print(f"               {r.error}")
            continue

        fails = [c for c in r.checks
                 if c.status == FAIL and SEV_RANK[c.severity] <= floor]
        unv = [c for c in r.checks if c.status == UNVERIFIED]
        unverified += len(unv)
        passes = sum(1 for c in r.checks if c.status == PASS)

        if not fails and not unv:
            print(f"[COMPLIANT ] {r.target:<34} {passes} check(s) passed")
            continue

        head = "FINDINGS  " if fails else "UNVERIFIED"
        print(f"\n[{head}] {r.target}  -  {len(fails)} finding(s), "
              f"{len(unv)} unverified, {passes} passed")
        for c in sorted(fails, key=lambda x: SEV_RANK[x.severity]):
            tally[c.severity] += 1
            where = f"  [{c.context}]" if c.context else ""
            print(f"  [{c.severity:<10}] {c.rule}{where}")
            print(f"               {c.detail}")
        if unv and not args.hide_unverified:
            for c in unv:
                where = f"  [{c.context}]" if c.context else ""
                print(f"  [UNVERIFIED] {c.rule}{where}")
                print(f"               {c.detail} - NOT a finding")
                print(f"               why: {c.reason}")

    print("\n" + "=" * 84)
    print(f"{tally['CRITICAL']} CRITICAL, {tally['HIGH']} HIGH, "
          f"{tally['MEDIUM']} MEDIUM, {tally['LOW']} LOW "
          f"across {len(results)} target(s).")
    if unverified:
        print(f"{unverified} check(s) could not be determined. They are NOT "
              f"findings and NOT passes - they are unknown.")
    if unreachable:
        print(f"{unreachable} target(s) could not be reached at all.")
    return tally, unverified, unreachable


def main() -> int:
    ap = argparse.ArgumentParser(
        description="Check TLS, SSH, certificates and IPsec against a crypto "
                    "policy you write yourself.",
        epilog="Credentials come from NET_USER and NET_PASS. The policy file has "
               "no built-in rules - write one with --write-example-policy.",
    )
    ap.add_argument("--policy", help="Path to your crypto-policy.yaml. Required "
                                     "unless --write-example-policy is used.")
    ap.add_argument("--targets", help="Path to targets.yaml: the TLS and SSH "
                                      "endpoints to probe from outside")
    ap.add_argument("--inventory", help="Path to devices.yaml: managed devices "
                                        "whose IPsec configuration is read over SSH")
    ap.add_argument("--write-example-policy", metavar="PATH",
                    help="Write a starter policy file to PATH and exit")
    ap.add_argument("--limit", help="Only check targets/devices whose host contains "
                                    "this text")
    ap.add_argument("--min-severity", choices=["CRITICAL", "HIGH", "MEDIUM", "LOW"],
                    default="LOW", help="Only report findings at or above this. "
                                        "Default: LOW")
    ap.add_argument("--hide-unverified", action="store_true",
                    help="Do not print the UNVERIFIED block. They are still counted "
                         "in the summary - they never become passes")
    ap.add_argument("--fail-on-unverified", action="store_true",
                    help="Exit 1 when any check could not be determined. For a CI "
                         "gate that must not pass on unknowns")
    ap.add_argument("--timeout", type=float, default=10.0,
                    help="Seconds per TLS/SSH probe. Default: 10")
    ap.add_argument("--read-timeout", type=int, default=120,
                    help="Seconds per device command. Default: 120")
    ap.add_argument("--workers", type=int, default=10,
                    help="How many targets to contact at once. Default: 10")
    ap.add_argument("--save-raw", metavar="DIR",
                    help="Save each device's raw crypto configuration here, so an "
                         "UNVERIFIED result can be investigated")
    ap.add_argument("--json-out", help="Write machine-readable results here")
    args = ap.parse_args()

    if args.write_example_policy:
        Path(args.write_example_policy).write_text(EXAMPLE_POLICY, encoding="utf-8")
        print(f"Wrote a starter policy to {args.write_example_policy}")
        print("Read it, change it to match YOUR standard, then run with --policy.")
        return 0

    if not args.policy:
        print("--policy is required. This script ships no rules of its own - the "
              "policy is yours.\n\nStart with:\n"
              "    python crypto_policy_checker.py --write-example-policy "
              "crypto-policy.yaml", file=sys.stderr)
        return 3
    if not args.targets and not args.inventory:
        print("Give it something to check: --targets (TLS/SSH endpoints), "
              "--inventory (devices), or both.", file=sys.stderr)
        return 3

    policy = load_policy(args.policy)

    jobs = []
    if args.targets:
        for t in load_targets(args.targets):
            if args.limit and args.limit not in t["host"]:
                continue
            jobs.append(("target", t))
    if args.inventory:
        credentials()  # fail fast, before any connection is opened
        for d in load_inventory(args.inventory):
            if args.limit and args.limit not in d["host"]:
                continue
            jobs.append(("device", d))

    if not jobs:
        print(f"Nothing matched --limit '{args.limit}'.", file=sys.stderr)
        return 3

    name = policy.get("name") or Path(args.policy).name
    print(f"Checking {len(jobs)} target(s) against policy: {name}")
    print("=" * 84)

    def run(job) -> TargetResult:
        kind, item = job
        if kind == "device":
            return check_device(item, policy, args.read_timeout, args.save_raw)
        if item["kind"] == "ssh":
            return check_ssh(item, policy, args.timeout)
        return check_tls(item, policy, args.timeout)

    results: list[TargetResult] = []
    with ThreadPoolExecutor(max_workers=args.workers) as pool:
        futures = {pool.submit(run, j): j for j in jobs}
        for fut in as_completed(futures):
            kind, item = futures[fut]
            try:
                results.append(fut.result())
            except Exception as exc:  # noqa: BLE001
                results.append(TargetResult(target=item.get("host", "?"), kind=kind,
                                            status="error",
                                            error=f"worker crashed: {exc}"))

    tally, unverified, unreachable = report(results, policy, args)

    if args.json_out:
        payload = {
            "generated_at": datetime.now(timezone.utc).isoformat(),
            "policy": name,
            "targets": [
                {"target": r.target, "kind": r.kind, "status": r.status,
                 "error": r.error or None,
                 "checks": [asdict(c) for c in r.checks]}
                for r in results
            ],
        }
        Path(args.json_out).write_text(json.dumps(payload, indent=2), encoding="utf-8")

    if tally["CRITICAL"]:
        return 2
    if tally["HIGH"] or unreachable:
        return 1
    if unverified and args.fail_on_unverified:
        return 1
    return 0


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