Ssl aki check.py

ssl_aki_check.py

#!/usr/bin/env python3.14
"""Check TLS servers and certificate files for the strict-mode Authority Key Identifier failure.

Python 3.13+ (including 3.14) adds VERIFY_X509_STRICT to ssl.create_default_context(). On
OpenSSL 3.x that enforces RFC 5280, and the failure most people hit is

    certificate verify failed: Missing Authority Key Identifier

Every certificate in the verified chain except the trust anchor (the leaf and any
intermediates) must carry an authorityKeyIdentifier extension.

TARGET is one of:
  host[:port], [v6addr]:port or https://URL
      handshake with and without VERIFY_X509_STRICT and show which certificate in the
      verified chain breaks the rule
  PEM or DER file (single cert, chain or CA bundle)
      static per-certificate check, no network access

Exit status: 0 no problem, 1 AKI (or other strict-mode) problem, 2 could not check.
"""

import argparse
import base64
import platform
import re
import socket
import ssl
import sys
from dataclasses import dataclass
from pathlib import Path
from urllib.parse import urlsplit

if sys.version_info < (3, 13):
    sys.exit(f"Python 3.13+ required (strict verification default), this is {platform.python_version()}")

AKI_OID = "2.5.29.35"
BASIC_CONSTRAINTS_OID = "2.5.29.19"
MISSING_AKI_CODE = 85  # X509_V_ERR_MISSING_AUTHORITY_KEY_IDENTIFIER
NAME_ATTRS = {
    "2.5.4.3": "CN", "2.5.4.5": "serialNumber", "2.5.4.6": "C", "2.5.4.7": "L", "2.5.4.8": "ST",
    "2.5.4.10": "O", "2.5.4.11": "OU", "0.9.2342.19200300.100.1.25": "DC",
    "1.2.840.113549.1.9.1": "emailAddress",
}
STRING_CODECS = {0x0C: "utf-8", 0x1C: "utf-32-be", 0x1E: "utf-16-be"}
PEM_CERT = re.compile(rb"-----BEGIN CERTIFICATE-----(.+?)-----END CERTIFICATE-----", re.DOTALL)
OK, ISSUE, ERROR = "ok", "issue", "error"


@dataclass(frozen=True)
class Cert:
    version: int
    subject: str
    issuer: str
    self_issued: bool
    is_ca: bool
    aki: bool


def _tlv(der: bytes, pos: int) -> tuple[int, int, int]:
    """Return (tag, value_start, value_end) of the DER element at pos."""
    tag, length = der[pos], der[pos + 1]
    pos += 2
    if length & 0x80:
        size = length & 0x7F
        length = int.from_bytes(der[pos:pos + size])
        pos += size
    if pos + length > len(der):
        raise ValueError("truncated DER")
    return tag, pos, pos + length


def _children(der: bytes, start: int, end: int) -> list[tuple[int, int, int]]:
    items = []
    while start < end:
        items.append(_tlv(der, start))
        start = items[-1][2]
    return items


def _oid(raw: bytes) -> str:
    arcs, value = [], 0
    for byte in raw:
        value = (value << 7) | (byte & 0x7F)
        if not byte & 0x80:
            arcs.append(value)
            value = 0
    first = min(arcs[0] // 40, 2)
    return ".".join(map(str, [first, arcs[0] - 40 * first, *arcs[1:]]))


def _name(der: bytes, start: int, end: int) -> str:
    parts = []
    for _, rdn_start, rdn_end in _children(der, start, end):
        for _, atv_start, atv_end in _children(der, rdn_start, rdn_end):
            (_, oid_start, oid_end), (tag, val_start, val_end) = _children(der, atv_start, atv_end)
            oid = _oid(der[oid_start:oid_end])
            value = der[val_start:val_end].decode(STRING_CODECS.get(tag, "latin-1"), "replace")
            parts.append(f"{NAME_ATTRS.get(oid, oid)}={value}")
    return ", ".join(parts) or "<empty>"


def parse_cert(der: bytes) -> Cert:
    try:
        _, start, end = _tlv(der, 0)
        _, start, end = _tlv(der, start)  # tbsCertificate
        fields = _children(der, start, end)
        version = 1
        if fields[0][0] == 0xA0:  # [0] EXPLICIT version, absent means v1
            _, v_start, v_end = _tlv(der, fields[0][1])
            version = int.from_bytes(der[v_start:v_end]) + 1
            fields = fields[1:]
        issuer, subject = fields[2], fields[4]
        extensions = {}
        for tag, ext_start, _ in fields[6:]:
            if tag != 0xA3:  # [3] EXPLICIT extensions
                continue
            _, seq_start, seq_end = _tlv(der, ext_start)
            for _, e_start, e_end in _children(der, seq_start, seq_end):
                oid, *_, value = _children(der, e_start, e_end)
                extensions[_oid(der[oid[1]:oid[2]])] = der[value[1]:value[2]]
        is_ca = False
        if bc := extensions.get(BASIC_CONSTRAINTS_OID):
            _, bc_start, bc_end = _tlv(bc, 0)
            items = _children(bc, bc_start, bc_end)
            is_ca = bool(items) and items[0][0] == 0x01 and bc[items[0][1]] != 0
        return Cert(
            version=version,
            subject=_name(der, subject[1], subject[2]),
            issuer=_name(der, issuer[1], issuer[2]),
            self_issued=der[issuer[1]:issuer[2]] == der[subject[1]:subject[2]],
            is_ca=is_ca,
            aki=AKI_OID in extensions,
        )
    except (IndexError, ValueError) as exc:
        raise ValueError(f"not a parsable X.509 certificate ({exc})") from exc


def _load_ders(path: Path) -> list[bytes]:
    data = path.read_bytes()
    blocks = PEM_CERT.findall(data)
    return [base64.b64decode(block) for block in blocks] if blocks else [data]


def _host_port(target: str) -> tuple[str, int]:
    url = urlsplit(target if "://" in target else f"//{target}")
    return url.hostname or target, url.port or 443


def _unverified_context() -> ssl.SSLContext:
    ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
    ctx.check_hostname = False
    ctx.verify_mode = ssl.CERT_NONE
    return ctx


def _fetch_chain(host: str, port: int, ctx: ssl.SSLContext, timeout: float) -> list[Cert]:
    with socket.create_connection((host, port), timeout=timeout) as sock:
        with ctx.wrap_socket(sock, server_hostname=host) as tls:
            ders = tls.get_unverified_chain() if ctx.verify_mode == ssl.CERT_NONE else tls.get_verified_chain()
    return [parse_cert(der) for der in ders]


def _attempt(
    host: str, port: int, ctx: ssl.SSLContext, timeout: float
) -> tuple[list[Cert], ssl.SSLCertVerificationError | None]:
    try:
        return _fetch_chain(host, port, ctx, timeout), None
    except ssl.SSLCertVerificationError as exc:
        return [], exc


def _outcome(error: ssl.SSLCertVerificationError | None) -> str:
    return "OK" if error is None else f"FAILED - {error.verify_message} (code {error.verify_code})"


def _report(chain: list[Cert], verified: bool) -> list[Cert]:
    """Print one line per certificate and return the ones that break the strict AKI rule."""
    bad = []
    last = len(chain) - 1
    for i, cert in enumerate(chain):
        if verified:  # same rule as OpenSSL check_extensions(): all but the trust anchor need an AKI
            role = "leaf" if i == 0 else "trust anchor" if i == last else "intermediate"
            required = i < last
        else:
            role = "self-issued" if cert.self_issued else "CA" if cert.is_ca else "end-entity"
            required = not cert.self_issued
        if cert.aki:
            state = "present"
        elif required and cert.version >= 3:
            state = "MISSING"
        else:
            state = "absent (exempt)"
        line = f"  [{i}] {role:<13} AKI {state:<16} {cert.subject}"
        if state == "MISSING":
            bad.append(cert)
            line += f"  <- issued by {cert.issuer}"
        print(line)
    if not verified and any(cert.is_ca for cert in bad):
        print("NOTE: a flagged CA is exempt only where clients trust it directly (it is then the trust anchor)")
    return bad


def _aki_issue() -> str:
    print("RESULT: AKI ISSUE - rejected under VERIFY_X509_STRICT (Python 3.13+ default)")
    print("FIX: have the issuing CA reissue the flagged certificate(s) with an authorityKeyIdentifier")
    print("     extension (OpenSSL config: authorityKeyIdentifier = keyid,issuer)")
    return ISSUE


def check_host(target: str, strict: ssl.SSLContext, relaxed: ssl.SSLContext, timeout: float) -> str:
    print(f"== {target} ==")
    relaxed_error = None
    try:
        host, port = _host_port(target)
        chain, strict_error = _attempt(host, port, strict, timeout)
        if strict_error is not None:
            chain, relaxed_error = _attempt(host, port, relaxed, timeout)
            if relaxed_error is not None:
                chain, _ = _attempt(host, port, _unverified_context(), timeout)
    except (OSError, ValueError) as exc:
        print(f"RESULT: ERROR - {exc}")
        return ERROR
    print(f"strict verification (Python 3.13+ default): {_outcome(strict_error)}")
    if strict_error is not None:
        print(f"relaxed verification (VERIFY_X509_STRICT off): {_outcome(relaxed_error)}")
    print("verified chain:" if relaxed_error is None else "served chain (unverified):")
    bad = _report(chain, verified=relaxed_error is None)
    if bad or (strict_error is not None and strict_error.verify_code == MISSING_AKI_CODE):
        if relaxed_error is not None:
            print(f"NOTE: verification also fails without VERIFY_X509_STRICT ({relaxed_error.verify_message})")
        return _aki_issue()
    if strict_error is None:
        print("RESULT: OK - passes VERIFY_X509_STRICT")
        return OK
    if relaxed_error is None:
        print(f"RESULT: STRICT-MODE ISSUE (not AKI) - {strict_error.verify_message}")
        return ISSUE
    print("RESULT: ERROR - verification fails even without VERIFY_X509_STRICT")
    return ERROR


def check_file(path: Path) -> str:
    print(f"== {path} ==")
    try:
        certs = [parse_cert(der) for der in _load_ders(path)]
    except (OSError, ValueError) as exc:
        print(f"RESULT: ERROR - {exc}")
        return ERROR
    print(f"{len(certs)} certificate(s), each checked on its own (chain position unknown):")
    if _report(certs, verified=False):
        return _aki_issue()
    print("RESULT: OK - no AKI problems")
    return OK


def _openssl_summary() -> str:
    version = ssl.OPENSSL_VERSION
    if not version.startswith("OpenSSL "):
        return f"{version} - not OpenSSL, AKI enforcement unknown; compare the strict and relaxed results"
    if ssl.OPENSSL_VERSION_INFO < (3,):
        return f"{version} - does not enforce the AKI check (needs 3.0+); live checks pass, trust the AKI column"
    return f"{version} - enforces the AKI check (OpenSSL 3.0+)"


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    parser.add_argument("targets", nargs="+", metavar="TARGET", help="host[:port], https:// URL, or certificate file")
    parser.add_argument("--cafile", help="CA bundle to trust instead of the default store")
    parser.add_argument("--timeout", type=float, default=10.0, help="connect timeout in seconds (default: %(default)s)")
    args = parser.parse_args(argv)
    try:
        strict = ssl.create_default_context(cafile=args.cafile)
        relaxed = ssl.create_default_context(cafile=args.cafile)
    except OSError as exc:
        parser.error(f"--cafile: {exc}")
    strict.verify_flags |= ssl.VERIFY_X509_STRICT
    relaxed.verify_flags &= ~ssl.VERIFY_X509_STRICT

    paths = ssl.get_default_verify_paths()
    print(f"Python {platform.python_version()}")
    print(f"OpenSSL library (loaded by Python): {_openssl_summary()}")
    print(f"default verify flags: {ssl.create_default_context().verify_flags.name}")
    print(f"trust store: {args.cafile or paths.cafile or paths.capath}")

    statuses = []
    for target in args.targets:
        print()
        path = Path(target)
        statuses.append(check_file(path) if path.is_file() else check_host(target, strict, relaxed, args.timeout))
    return 1 if ISSUE in statuses else 2 if ERROR in statuses else 0


if __name__ == "__main__":
    sys.exit(main())