ssl aki check.py
Jump to navigation
Jump to search
See Python/3.14/Authority Key Identifier
#!/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())