Python/3.14/Authority Key Identifier
Jump to navigation
Jump to search
ssl.VERIFY_X509_STRICT
Python 3.13+ has hardened its security requirements by enabling strict X.509 verification flags (ssl.VERIFY_X509_STRICT) by default. This strict enforcement adheres fully to RFC 5280.
If your environment uses older Certificate Authorities (CAs), local Kubernetes clusters, or corporate SSL-decrypting proxies (like Palo Alto Networks, Zscaler, or Burp) that sign downstream leaf certificates without generating an Authority Key Identifier (AKI) extension, Python will instantly reject the connection with this error.
Fixes
Palo Alto Networks Firewall
debug dataplane set ssl-decrypt akid-disable no
Disable Strict Checking
Not a great solution...
If using ssl module:
import ssl ctx = ssl.create_default_context() # Remove the strict verification flag ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT
If using request library:
import ssl
import requests
from urllib3.poolmanager import PoolManager
from requests.adapters import HTTPAdapter
class RelaxedX509Adapter(HTTPAdapter):
def init_poolmanager(self, connections, maxsize, block=False, **pool_kwargs):
ctx = ssl.create_default_context()
ctx.verify_flags &= ~ssl.VERIFY_X509_STRICT
pool_kwargs['ssl_context'] = ctx
self.poolmanager = PoolManager(num_pools=connections, max_size=maxsize, block=block, **pool_kwargs)
session = requests.Session()
session.mount('https://', RelaxedX509Adapter())
# This will now succeed without throwing the AKI error
response = session.get('https://your-internal-domain.com')
Example SSL Test Scripts
SSL Library
import ssl
import socket
hostname = 'www.example.com'
context = ssl.create_default_context()
with socket.create_connection((hostname, 443)) as sock:
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
print(ssock.version())
Requests Library
import requests url = 'https://www.example.com' response = requests.get(url) print(response.status_code) print(response.text[:200]) # Print the first 200 characters of the response
ssl_aki_check.py
usage:
python3.14 ssl_aki_check.py api.github.com
#!/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())