Python/3.14/Authority Key Identifier: Difference between revisions
| Line 153: | Line 153: | ||
== ssl_aki_check.py == | == ssl_aki_check.py == | ||
[[ssl_aki_check.py?action=raw]] | |||
usage: | usage: | ||
Revision as of 18:05, 7 October 2026
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 = 'api.github.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://api.github.com' response = requests.get(url) print(response.status_code) print(response.text[:200]) # Print the first 200 characters of the response
Example Failures
SSL Library Failure
$ python test_ssl_library.py
Traceback (most recent call last):
File "/home/someuser/test_ssl_library.py", line 8, in <module>
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
~~~~~~~~~~~~~~~~~~~^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
File "/usr/lib/python3.14/ssl.py", line 469, in wrap_socket
return self.sslsocket_class._create(
~~~~~~~~~~~~~~~~~~~~~~~~~~~~^
sock=sock,
^^^^^^^^^^
...<5 lines>...
session=session
^^^^^^^^^^^^^^^
)
^
File "/usr/lib/python3.14/ssl.py", line 1087, in _create
self.do_handshake()
~~~~~~~~~~~~~~~~~^^
File "/usr/lib/python3.14/ssl.py", line 1383, in do_handshake
self._sslobj.do_handshake()
~~~~~~~~~~~~~~~~~~~~~~~~~^^
ssl.SSLCertVerificationError: [SSL: CERTIFICATE_VERIFY_FAILED] certificate verify failed: Missing Authority Key Identifier (_ssl.c:1082)
Requests Library Failure
Did not fail - Requests seemed to be okay with the AKI issue...
$ python test_requests_library.py
200
{"current_user_url":"https://api.github.com/user","current_user_authorizations_html_url":"https://github.com/settings/connections/applications{/client_id}","authorizations_url":"https://api.github.com
check aki sample output
sample check aki success
$ python3.14 check_aki.py api.github.com Python 3.14.8 OpenSSL library (loaded by Python): OpenSSL 3.3.1 4 Jun 2024 - enforces the AKI check (OpenSSL 3.0+) default verify flags: VERIFY_X509_STRICT|VERIFY_X509_TRUSTED_FIRST|VERIFY_X509_PARTIAL_CHAIN trust store: /usr/lib/ssl/cert.pem
api.github.com
strict verification (Python 3.13+ default): OK verified chain:
[0] leaf AKI present CN=*.github.com [1] intermediate AKI present C=GB, O=Sectigo Limited, CN=Sectigo Public Server Authentication CA DV E36 [2] trust anchor AKI absent (exempt) C=GB, O=Sectigo Limited, CN=Sectigo Public Server Authentication Root E46
RESULT: OK - passes VERIFY_X509_STRICT
sample check aki failure
$ python3.14 check_aki.py api.github.com
Python 3.14.8
OpenSSL library (loaded by Python): OpenSSL 3.0.2 15 Mar 2022 - enforces the AKI check (OpenSSL 3.0+)
default verify flags: VERIFY_X509_STRICT|VERIFY_X509_TRUSTED_FIRST|VERIFY_X509_PARTIAL_CHAIN
trust store: /usr/lib/ssl/cert.pem
== api.github.com ==
strict verification (Python 3.13+ default): FAILED - Missing Authority Key Identifier (code 85)
relaxed verification (VERIFY_X509_STRICT off): OK
verified chain:
[0] leaf AKI MISSING CN=*.github.com <- issued by C=US, ST=XXX, L=XXX, O=XXX, OU=XXX, CN=decrypt.example.com
[1] trust anchor AKI present C=US, ST=XXX, L=XXX, O=XXX, OU=XXX, CN=decrypt.example.com
RESULT: AKI ISSUE - rejected under VERIFY_X509_STRICT (Python 3.13+ default)
FIX: have the issuing CA reissue the flagged certificate(s) with an authorityKeyIdentifier
extension (OpenSSL config: authorityKeyIdentifier = keyid,issuer)
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())