From 71ef4bb0ead2cd8c653ad7551536c46bce4258b5 Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 19:46:34 +0200 Subject: [PATCH 1/7] start new dev branch; add audit file --- .audit/oberstet_fix_2014.md | 8 ++++++++ 1 file changed, 8 insertions(+) create mode 100644 .audit/oberstet_fix_2014.md diff --git a/.audit/oberstet_fix_2014.md b/.audit/oberstet_fix_2014.md new file mode 100644 index 000000000..81c93aedf --- /dev/null +++ b/.audit/oberstet_fix_2014.md @@ -0,0 +1,8 @@ +- [ ] I did **not** use any AI-assistance tools to help create this pull request. +- [x] I **did** use AI-assistance tools to *help* create this pull request. +- [x] I have read, understood and followed the project's AI_POLICY.md when creating code, documentation etc. for this pull request. + +Submitted by: @oberstet +Date: 2026-10-20 +Related issue(s): #2014 +Branch: oberstet:fix_2014 From 96b399ec3934a35e1bedb92fba50edd4e1fe7bba Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 19:54:12 +0200 Subject: [PATCH 2/7] chore: bump version to 26.10.2.dev1 (#2014) Start the 26.10.2 development cycle; #2014 (extract_peer_certificate() with pyOpenSSL >= 26.2.0) ships in autobahn 26.10.2, which Crossbar.io 26.10.1 will depend on. Expected CI result: GREEN (version change only). Note: This work was completed with AI assistance (Claude Code). --- pyproject.toml | 2 +- src/autobahn/_version.py | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 7d4987c18..445da3048 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "autobahn" -version = "26.10.1" +version = "26.10.2.dev1" description = "WebSocket client & server library, WAMP real-time framework" readme = "README.md" requires-python = ">=3.11" diff --git a/src/autobahn/_version.py b/src/autobahn/_version.py index 9196f4716..867fddac7 100644 --- a/src/autobahn/_version.py +++ b/src/autobahn/_version.py @@ -24,6 +24,6 @@ # ############################################################################### -__version__ = "26.10.1" +__version__ = "26.10.2.dev1" __build__ = "00000000-0000000" From cf8edbc785a662d69f3c88d8f31359cd4efb2728 Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 20:15:29 +0200 Subject: [PATCH 3/7] #2014: red: tests for WAMP over TLS with client certificates and channel binding New tests over real TLS connections (loopback, certificates created per test: a CA, a server certificate for localhost, client certificates client_0 and client_1 as in crossbar-examples authentication/tls), on Twisted and asyncio, over WebSocket and RawSocket, on TLS 1.2 and TLS 1.3. The router side is a minimal FakeRouterSession (autobahn/test/tlsutil.py) that authenticates like Crossbar.io: authmethod "tls" by the SHA1 fingerprint of the client certificate (PendingAuthTLS), and WAMP-cryptosign with channel binding "tls-unique" (PendingAuthCryptosign, but denying instead of falling back when the router has no channel ID). Twisted (autobahn/twisted/test/test_tx_tls.py), 21 tests: - extract_peer_certificate() on a client certificate with SAN, key usage and EKU, field by field against values computed from the DER: FAILS (AttributeError: 'X509' object has no attribute 'get_extension', removed in pyOpenSSL 26.2.0) - extract_peer_certificate() on a certificate with a duplicate extension must not raise: FAILS (same) - authmethod "tls" accepted (default, TLS 1.2, TLS 1.3) and an unknown client certificate denied for that reason, WebSocket and RawSocket: FAIL, the router sees no client certificate (peer_cert is only read in connectionMade(), before the TLS handshake) - WAMP-cryptosign with channel binding over RawSocket (4 tests): FAIL, the signature does not match the router's channel ID. The Twisted RawSocket protocols never set is_server, so the client side takes the server's TLS Finished message for its tls-unique channel ID - passing: no client certificate denied (WebSocket, RawSocket), cryptosign with channel binding over WebSocket (4 tests), no peer certificate asyncio (autobahn/asyncio/test/test_aio_tls.py), 26 tests: - authmethod "tls" accepted / unknown certificate denied (8 tests): FAIL, asyncio never fills peer_cert - WAMP-cryptosign with channel binding without a client certificate (6 tests): FAIL, a TLS server connection without a client certificate is not is_secure and has no channel ID (is_secure is derived from the peer certificate) - passing: channel binding with a client certificate (6), no client certificate denied (2) Expected CI result: RED. Twisted trial fails in exactly these 14 tests (locally cpy311: 349 ok, 12 failures, 2 errors; same on pypy311), asyncio pytest in exactly these 14 tests (320 passed, 14 failed); ruff and ty 0.0.89 clean. Note: This work was completed with AI assistance (Claude Code). --- src/autobahn/asyncio/test/test_aio_tls.py | 272 ++++++++++++++++ src/autobahn/test/tlsutil.py | 365 ++++++++++++++++++++++ src/autobahn/twisted/test/test_tx_tls.py | 363 +++++++++++++++++++++ 3 files changed, 1000 insertions(+) create mode 100644 src/autobahn/asyncio/test/test_aio_tls.py create mode 100644 src/autobahn/test/tlsutil.py create mode 100644 src/autobahn/twisted/test/test_tx_tls.py diff --git a/src/autobahn/asyncio/test/test_aio_tls.py b/src/autobahn/asyncio/test/test_aio_tls.py new file mode 100644 index 000000000..06f72c040 --- /dev/null +++ b/src/autobahn/asyncio/test/test_aio_tls.py @@ -0,0 +1,272 @@ +############################################################################### +# +# Copyright (C) typedef int GmbH +# SPDX-License-Identifier: MIT +# +############################################################################### +""" +WAMP over TLS on asyncio (#2014): the TLS client certificate in the transport details, +TLS client certificate authentication (authmethod "tls"), and WAMP-cryptosign with TLS +channel binding ("tls-unique"), over WebSocket and RawSocket, on TLS 1.2 and TLS 1.3. + +The same scenarios as autobahn/twisted/test/test_tx_tls.py, over real TLS connections on +the loopback interface, with certificates created for each test (see autobahn.test.tlsutil). +""" + +import asyncio +import binascii +import os +import ssl + +import pytest + +pytestmark = pytest.mark.skipif( + not os.environ.get("USE_ASYNCIO", False), reason="test runs on asyncio only" +) + +if os.environ.get("USE_ASYNCIO", False): + from autobahn.asyncio.rawsocket import ( + WampRawSocketClientFactory, + WampRawSocketServerFactory, + ) + from autobahn.asyncio.wamp import ApplicationSession + from autobahn.asyncio.websocket import ( + WampWebSocketClientFactory, + WampWebSocketServerFactory, + ) + from autobahn.test.tlsutil import ( + CLIENT_SUBJECT, + FakeRouterSession, + ExamplePKI, + expected_peer_cert, + ) + from autobahn.wamp import cryptosign + from autobahn.wamp.auth import AuthCryptoSign + from autobahn.wamp.types import ComponentConfig + + class _Client(ApplicationSession): + """ + A WAMP client that joins with one authmethod and reports the outcome. + """ + + def __init__(self, config, authmethod, authenticator, done): + ApplicationSession.__init__(self, config) + self._authmethod = authmethod + self._authenticator = authenticator + self._done = done + + def onConnect(self): + authextra = self._authenticator.authextra if self._authenticator else None + self.join( + self.config.realm, authmethods=[self._authmethod], authextra=authextra + ) + + def onChallenge(self, challenge): + return self._authenticator.on_challenge(self, challenge) + + def onJoin(self, details): + if not self._done.done(): + self._done.set_result( + ("joined", details, self._transport.transport_details) + ) + self.leave() + + def onLeave(self, details): + if not self._done.done(): + self._done.set_result(("left", details, None)) + self.disconnect() + + def onDisconnect(self): + if not self._done.done(): + self._done.set_result(("disconnected", None, None)) + + +_TRANSPORTS = ["websocket", "rawsocket"] +_TLS_VERSIONS = [None, "1.2", "1.3"] +_PRIVKEY = b"\x01" * 32 + + +class _Setup: + def __init__(self, certdir): + self.pki = ExamplePKI(certdir) + self.observed = [] + self.principals = { + expected_peer_cert(self.pki.client_certs["client_0"], {}, {})["sha1"]: ( + "client_0", + "backend", + ), + } + self.pubkeys = { + cryptosign.CryptosignKey.from_bytes(_PRIVKEY).public_key(): ( + "alice", + "frontend", + ) + } + + def aborts(self): + return [what for event, what in self.observed if event == "abort"] + + def hello_details(self): + return [td for event, td in self.observed if event == "hello"][0] + + def server_context(self, tls_version): + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER) + ctx.load_cert_chain( + self.pki.path("server", "crt"), self.pki.path("server", "key") + ) + ctx.load_verify_locations(self.pki.path("ca", "crt")) + ctx.verify_mode = ssl.CERT_OPTIONAL + if tls_version: + version = {"1.2": ssl.TLSVersion.TLSv1_2, "1.3": ssl.TLSVersion.TLSv1_3}[ + tls_version + ] + ctx.minimum_version = ctx.maximum_version = version + return ctx + + def client_context(self, client): + ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(self.pki.path("ca", "crt")) + if client: + ctx.load_cert_chain( + self.pki.path(client, "crt"), self.pki.path(client, "key") + ) + return ctx + + async def join( + self, transport, client, authmethod, authenticator=None, tls_version=None + ): + """ + Connect a client with the TLS client certificate ``client`` (or none), join with + ``authmethod``, and return ``(outcome, details, client transport details)``. + """ + loop = asyncio.get_running_loop() + server_closed = loop.create_future() + done = loop.create_future() + + def router_session(): + return FakeRouterSession( + principals=self.principals, + pubkeys=self.pubkeys, + observed=self.observed, + on_close=lambda: server_closed.done() or server_closed.set_result(None), + ) + + def client_session(): + return _Client(ComponentConfig("realm1"), authmethod, authenticator, done) + + if transport == "websocket": + server_factory = WampWebSocketServerFactory(router_session) + else: + server_factory = WampRawSocketServerFactory(router_session) + + server = await loop.create_server( + server_factory, "127.0.0.1", 0, ssl=self.server_context(tls_version) + ) + port = server.sockets[0].getsockname()[1] + try: + if transport == "websocket": + client_factory = WampWebSocketClientFactory( + client_session, url=f"wss://localhost:{port}" + ) + else: + client_factory = WampRawSocketClientFactory(client_session) + + client_transport, _ = await loop.create_connection( + client_factory, + "127.0.0.1", + port, + ssl=self.client_context(client), + server_hostname="localhost", + ) + result = await asyncio.wait_for(done, 10) + # wait until both sides closed the connection + await asyncio.wait_for(server_closed, 10) + client_transport.close() + return result + finally: + server.close() + + +@pytest.fixture +def setup(tmp_path): + return _Setup(str(tmp_path / "certs")) + + +# TLS client certificate authentication (authmethod "tls") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("tls_version", _TLS_VERSIONS) +@pytest.mark.parametrize("transport", _TRANSPORTS) +async def test_tls_auth_accepted(setup, transport, tls_version): + outcome, details, _ = await setup.join( + transport, "client_0", "tls", tls_version=tls_version + ) + assert setup.aborts() == [] + assert outcome == "joined" + assert (details.authid, details.authrole, details.authmethod) == ( + "client_0", + "backend", + "tls", + ) + + # what the router sees: the client certificate, after the TLS handshake + peer_cert = setup.hello_details().peer_cert + cert = setup.pki.client_certs["client_0"] + expected = expected_peer_cert( + cert, + dict(CLIENT_SUBJECT, cn="client_0"), + {"o": "Example GmbH", "cn": "Example Test CA"}, + ) + for key, value in expected.items(): + assert peer_cert[key] == value, key + + +@pytest.mark.asyncio +@pytest.mark.parametrize("transport", _TRANSPORTS) +async def test_tls_auth_unknown_certificate(setup, transport): + """ + A client certificate the CA signed, but that is not a principal: denied, for that reason. + """ + outcome, _, _ = await setup.join(transport, "client_1", "tls") + assert outcome == "left" + sha1 = expected_peer_cert(setup.pki.client_certs["client_1"], {}, {})["sha1"] + assert setup.aborts() == [f"client certificate {sha1} not accepted"] + + +@pytest.mark.asyncio +@pytest.mark.parametrize("transport", _TRANSPORTS) +async def test_tls_auth_no_certificate(setup, transport): + outcome, _, _ = await setup.join(transport, None, "tls") + assert outcome == "left" + assert setup.aborts() == ["no client certificate presented"] + assert setup.hello_details().peer_cert is None + + +# WAMP-cryptosign with TLS channel binding + + +@pytest.mark.asyncio +@pytest.mark.parametrize("client", [None, "client_0"]) +@pytest.mark.parametrize("tls_version", _TLS_VERSIONS) +@pytest.mark.parametrize("transport", _TRANSPORTS) +async def test_cryptosign_channel_binding(setup, transport, tls_version, client): + """ + With and without a TLS client certificate: the router side of a TLS connection is secure + and has the same, non-zero channel ID as the client side. + """ + authenticator = AuthCryptoSign( + privkey=binascii.b2a_hex(_PRIVKEY).decode(), + authextra={"channel_binding": "tls-unique"}, + ) + outcome, details, client_td = await setup.join( + transport, client, "cryptosign", authenticator, tls_version + ) + assert setup.aborts() == [] + assert outcome == "joined" + assert (details.authid, details.authmethod) == ("alice", "cryptosign") + + server_td = setup.hello_details() + assert server_td.is_secure + assert server_td.channel_id["tls-unique"] == client_td.channel_id["tls-unique"] + assert server_td.channel_id["tls-unique"] != FakeRouterSession.ZERO_CHANNEL_ID diff --git a/src/autobahn/test/tlsutil.py b/src/autobahn/test/tlsutil.py new file mode 100644 index 000000000..3c634ae3f --- /dev/null +++ b/src/autobahn/test/tlsutil.py @@ -0,0 +1,365 @@ +############################################################################### +# +# Copyright (C) typedef int GmbH +# SPDX-License-Identifier: MIT +# +############################################################################### +""" +Test helpers for WAMP over TLS (#2014), shared by the Twisted and the asyncio tests. + +This module imports neither Twisted nor asyncio networking code: it only creates +certificates (with ``cryptography``) and provides the router side of a WAMP session +for authentication tests. +""" + +import binascii +import datetime +import hashlib +import os + +from cryptography import x509 +from cryptography.hazmat.primitives import hashes, serialization +from cryptography.hazmat.primitives.asymmetric import ec +from cryptography.x509.oid import ExtendedKeyUsageOID, NameOID + +from autobahn import util +from autobahn.wamp import message, role + +# a subject with all the attribute types a client certificate typically carries +CLIENT_SUBJECT = { + "c": "DE", + "st": "Bavaria", + "l": "Erlangen", + "o": "Example GmbH", + "ou": "Testing", + "emailaddress": "client@example.com", +} + +_NAME_OIDS = { + "c": NameOID.COUNTRY_NAME, + "st": NameOID.STATE_OR_PROVINCE_NAME, + "l": NameOID.LOCALITY_NAME, + "o": NameOID.ORGANIZATION_NAME, + "ou": NameOID.ORGANIZATIONAL_UNIT_NAME, + "cn": NameOID.COMMON_NAME, + "emailaddress": NameOID.EMAIL_ADDRESS, +} + + +def _name(attrs): + return x509.Name([x509.NameAttribute(_NAME_OIDS[k], v) for k, v in attrs.items()]) + + +def fingerprint(cert, algorithm): + """ + Fingerprint of a certificate as ``AA:BB:..`` (upper case hex), computed from its DER encoding. + """ + digest = hashlib.new( + algorithm, cert.public_bytes(serialization.Encoding.DER) + ).digest() + return ":".join(f"{b:02X}" for b in digest) + + +def _make_cert(subject, issuer_name, issuer_key, key, extensions, serial=None): + now = datetime.datetime.now(datetime.UTC) + builder = x509.CertificateBuilder( + issuer_name=issuer_name, + subject_name=subject, + public_key=key.public_key(), + serial_number=serial or x509.random_serial_number(), + not_valid_before=(now - datetime.timedelta(days=1)).replace(microsecond=0), + not_valid_after=(now + datetime.timedelta(days=1)).replace(microsecond=0), + extensions=extensions, + ) + return builder.sign(issuer_key, hashes.SHA256()) + + +class ExamplePKI: + """ + A certificate authority, a server certificate for ``localhost``, and client certificates + ``client_0`` and ``client_1`` issued by it (as in the crossbar-examples TLS authentication + example), written as PEM files to a directory. + """ + + def __init__(self, certdir): + os.makedirs(certdir, exist_ok=True) + self.certdir = certdir + + self.ca_key = ec.generate_private_key(ec.SECP256R1()) + self.ca_name = _name({"o": "Example GmbH", "cn": "Example Test CA"}) + self.ca_cert = _make_cert( + self.ca_name, + self.ca_name, + self.ca_key, + self.ca_key, + [ + x509.Extension( + x509.BasicConstraints.oid, + True, + x509.BasicConstraints(ca=True, path_length=None), + ), + x509.Extension( + x509.KeyUsage.oid, + True, + x509.KeyUsage( + False, False, False, False, False, True, True, False, False + ), + ), + ], + ) + self._write("ca", self.ca_key, self.ca_cert) + + self.server_key = ec.generate_private_key(ec.SECP256R1()) + self.server_cert = _make_cert( + _name({"cn": "localhost"}), + self.ca_name, + self.ca_key, + self.server_key, + [ + x509.Extension( + x509.SubjectAlternativeName.oid, + False, + x509.SubjectAlternativeName([x509.DNSName("localhost")]), + ), + ], + ) + self._write("server", self.server_key, self.server_cert) + + self.client_keys = {} + self.client_certs = {} + for name in ["client_0", "client_1"]: + key = ec.generate_private_key(ec.SECP256R1()) + cert = _make_cert( + _name(dict(CLIENT_SUBJECT, cn=name)), + self.ca_name, + self.ca_key, + key, + self.client_extensions(name), + ) + self.client_keys[name] = key + self.client_certs[name] = cert + self._write(name, key, cert) + + @staticmethod + def client_extensions(name): + """ + The extensions of a client certificate: SAN, key usage (critical), extended key usage. + """ + return [ + x509.Extension( + x509.SubjectAlternativeName.oid, + False, + x509.SubjectAlternativeName( + [ + x509.DNSName(name), + x509.RFC822Name(CLIENT_SUBJECT["emailaddress"]), + ] + ), + ), + x509.Extension( + x509.KeyUsage.oid, + True, + x509.KeyUsage( + True, False, False, False, False, False, False, False, False + ), + ), + x509.Extension( + x509.ExtendedKeyUsage.oid, + False, + x509.ExtendedKeyUsage([ExtendedKeyUsageOID.CLIENT_AUTH]), + ), + ] + + def duplicate_extension_cert(self): + """ + A client certificate with its SAN extension twice: OpenSSL accepts it, ``cryptography`` + refuses to parse its extensions (``DuplicateExtension``). + """ + key = ec.generate_private_key(ec.SECP256R1()) + san = self.client_extensions("client_dup")[0] + return _make_cert( + _name(dict(CLIENT_SUBJECT, cn="client_dup")), + self.ca_name, + self.ca_key, + key, + [san, san], + ) + + def path(self, name, kind): + return os.path.join(self.certdir, f"{name}.{kind}") + + def pem(self, name, kind): + with open(self.path(name, kind), "rb") as f: + return f.read() + + def _write(self, name, key, cert): + with open(self.path(name, "key"), "wb") as f: + f.write( + key.private_bytes( + serialization.Encoding.PEM, + serialization.PrivateFormat.TraditionalOpenSSL, + serialization.NoEncryption(), + ) + ) + with open(self.path(name, "crt"), "wb") as f: + f.write(cert.public_bytes(serialization.Encoding.PEM)) + + +def expected_peer_cert(cert, subject, issuer): + """ + The ``TransportDetails.peer_cert`` dict expected for a certificate, for the fields whose + format does not depend on the implementation. ``subject`` and ``issuer`` are the attributes + as given when creating the certificate (OpenSSL short names, lower case). + """ + return { + "md5": fingerprint(cert, "md5"), + "sha1": fingerprint(cert, "sha1"), + "sha256": fingerprint(cert, "sha256"), + "expired": False, + "serial": cert.serial_number, + "signature_algorithm": "ecdsa-with-SHA256", + "version": 2, + "not_before": cert.not_valid_before_utc.strftime("%Y%m%d%H%M%SZ"), + "not_after": cert.not_valid_after_utc.strftime("%Y%m%d%H%M%SZ"), + "subject": subject, + "issuer": issuer, + } + + +class FakeRouterSession: + """ + The router side of a WAMP session (``ITransportHandler``), doing only what a router does to + authenticate a client: + + * ``tls``: accept the client if the SHA1 fingerprint of its TLS client certificate + (``transport_details.peer_cert["sha1"]``) is a known principal, as Crossbar.io's + ``PendingAuthTLS`` does. + * ``cryptosign``: send a challenge and verify the signature; with + ``channel_binding="tls-unique"`` the signed message is the challenge XOR the TLS channel ID + of the router side of the connection, as in Crossbar.io's ``PendingAuthCryptosign``. + Unlike Crossbar.io, it denies the client when the channel binding is requested but the + router has no channel ID, instead of falling back to no channel binding. + + ``observed`` collects ``(event, transport_details)`` pairs, and the reason of any ABORT; + ``on_close`` is called when the transport is closed. + """ + + ZERO_CHANNEL_ID = b"\x00" * 32 + + def __init__(self, principals=None, pubkeys=None, observed=None, on_close=None): + self._principals = principals or {} + self._pubkeys = pubkeys or {} + self.observed = observed if observed is not None else [] + self._transport = None + self._pending = None + self._on_close = on_close + # read by the WAMP transports (for logging) + self._authid = None + self._session_id = None + + def onOpen(self, transport): + self._transport = transport + self.observed.append(("open", transport.transport_details)) + + def onMessage(self, msg): + if isinstance(msg, message.Hello): + td = self._transport.transport_details + self.observed.append(("hello", td)) + if "tls" in msg.authmethods: + self._hello_tls(td) + elif "cryptosign" in msg.authmethods: + self._hello_cryptosign(msg, td) + else: + self._abort("wamp.error.no_auth_method", "no supported authmethod") + elif isinstance(msg, message.Authenticate) and self._pending: + self._authenticate(msg) + elif isinstance(msg, message.Goodbye): + self._transport.send(message.Goodbye()) + self._transport.close() + + def onClose(self, wasClean): + self.observed.append(("close", None)) + if self._on_close: + self._on_close() + + def _hello_tls(self, td): + peer_cert = td.peer_cert + if not peer_cert: + self._abort("wamp.error.not_authorized", "no client certificate presented") + elif peer_cert.get("sha1") not in self._principals: + self._abort( + "wamp.error.not_authorized", + "client certificate {} not accepted".format(peer_cert.get("sha1")), + ) + else: + authid, authrole = self._principals[peer_cert["sha1"]] + self._welcome(authid, authrole, "tls") + + def _hello_cryptosign(self, msg, td): + pubkey = (msg.authextra or {}).get("pubkey") + if pubkey not in self._pubkeys: + self._abort("wamp.error.not_authorized", "unknown public key") + return + channel_binding = msg.authextra.get("channel_binding") + challenge = os.urandom(32) + if channel_binding == "tls-unique": + channel_id = td.channel_id.get("tls-unique") if td.channel_id else None + if not channel_id or channel_id == self.ZERO_CHANNEL_ID: + self._abort( + "wamp.error.not_authorized", + "channel binding requested, but no TLS channel ID", + ) + return + expected = util.xor(challenge, channel_id) + else: + expected = challenge + self._pending = (pubkey, expected) + extra = { + "challenge": binascii.b2a_hex(challenge).decode(), + "channel_binding": channel_binding, + } + self._transport.send(message.Challenge("cryptosign", extra)) + + def _authenticate(self, msg): + from nacl.exceptions import BadSignatureError + from nacl.signing import VerifyKey + + pubkey, expected = self._pending + self._pending = None + try: + signed = VerifyKey(binascii.a2b_hex(pubkey)).verify( + binascii.a2b_hex(msg.signature) + ) + except BadSignatureError: + self._abort("wamp.error.not_authorized", "invalid signature") + return + if signed != expected: + self._abort( + "wamp.error.not_authorized", + "signed message does not match the TLS channel", + ) + return + authid, authrole = self._pubkeys[pubkey] + self._welcome(authid, authrole, "cryptosign") + + def _welcome(self, authid, authrole, authmethod): + roles = { + "broker": role.RoleBrokerFeatures(), + "dealer": role.RoleDealerFeatures(), + } + self._authid = authid + self._session_id = util.id() + self._transport.send( + message.Welcome( + self._session_id, + roles, + authid=authid, + authrole=authrole, + authmethod=authmethod, + authprovider="static", + ) + ) + + def _abort(self, reason, msg): + self.observed.append(("abort", msg)) + self._transport.send(message.Abort(reason, msg)) diff --git a/src/autobahn/twisted/test/test_tx_tls.py b/src/autobahn/twisted/test/test_tx_tls.py new file mode 100644 index 000000000..c72af8755 --- /dev/null +++ b/src/autobahn/twisted/test/test_tx_tls.py @@ -0,0 +1,363 @@ +############################################################################### +# +# Copyright (C) typedef int GmbH +# SPDX-License-Identifier: MIT +# +############################################################################### +""" +WAMP over TLS on Twisted (#2014): the TLS client certificate in the transport details, +TLS client certificate authentication (authmethod "tls"), and WAMP-cryptosign with TLS +channel binding ("tls-unique"), over WebSocket and RawSocket, on TLS 1.2 and TLS 1.3. + +The connections are real TLS connections on the loopback interface, with certificates +created for each test (see :mod:`autobahn.test.tlsutil`). +""" + +import binascii + +import txaio + +txaio.use_twisted() # noqa + +from OpenSSL import crypto +from twisted.internet import reactor, ssl +from twisted.internet.defer import Deferred, inlineCallbacks +from twisted.internet.endpoints import SSL4ClientEndpoint, SSL4ServerEndpoint +from twisted.internet.interfaces import ISSLTransport +from twisted.trial import unittest +from zope.interface import implementer + +from autobahn.test.tlsutil import ( + CLIENT_SUBJECT, + FakeRouterSession, + ExamplePKI, + expected_peer_cert, +) +from autobahn.twisted.rawsocket import ( + WampRawSocketClientFactory, + WampRawSocketServerFactory, +) +from autobahn.twisted.util import extract_peer_certificate +from autobahn.twisted.wamp import ApplicationSession +from autobahn.twisted.websocket import ( + WampWebSocketClientFactory, + WampWebSocketServerFactory, +) +from autobahn.wamp import cryptosign +from autobahn.wamp.auth import AuthCryptoSign +from autobahn.wamp.types import ComponentConfig + +_TLS_VERSIONS = {"1.2": ssl.TLSVersion.TLSv1_2, "1.3": ssl.TLSVersion.TLSv1_3} + + +@implementer(ISSLTransport) +class _FakeTLSTransport: + """ + Just enough of a TLS transport for extract_peer_certificate(). + """ + + def __init__(self, cert): + self._cert = cert + + def getPeerCertificate(self): + return self._cert + + +class ExtractPeerCertificateTests(unittest.TestCase): + """ + extract_peer_certificate() on certificates as clients present them. + """ + + def setUp(self): + self.pki = ExamplePKI(self.mktemp()) + + def _extract(self, cert): + x509 = crypto.X509.from_cryptography(cert) + return extract_peer_certificate(_FakeTLSTransport(x509)) + + def test_certificate_with_extensions(self): + """ + All fields of a client certificate with SAN, key usage and extended key usage. + """ + cert = self.pki.client_certs["client_0"] + result = self._extract(cert) + + expected = expected_peer_cert( + cert, + dict(CLIENT_SUBJECT, cn="client_0"), + {"o": "Example GmbH", "cn": "Example Test CA"}, + ) + for key, value in expected.items(): + self.assertEqual(result[key], value, key) + + extensions = {ext["name"]: ext for ext in result["extensions"]} + self.assertEqual( + sorted(extensions), ["extendedKeyUsage", "keyUsage", "subjectAltName"] + ) + self.assertEqual(extensions["keyUsage"]["critical"], True) + self.assertEqual(extensions["subjectAltName"]["critical"], False) + self.assertIn("client_0", extensions["subjectAltName"]["value"]) + + def test_malformed_certificate_does_not_raise(self): + """ + A certificate whose extensions cannot be parsed (a duplicate extension) still yields + the certificate's other fields: this runs on peer data before authentication. + """ + cert = self.pki.duplicate_extension_cert() + result = self._extract(cert) + self.assertEqual( + result["sha1"], + expected_peer_cert(cert, {}, {})["sha1"], + ) + self.assertEqual(result["subject"]["cn"], "client_dup") + + def test_no_peer_certificate(self): + self.assertIsNone(extract_peer_certificate(_FakeTLSTransport(None))) + + +class _Client(ApplicationSession): + """ + A WAMP client that joins with one authmethod and reports the outcome. + """ + + def __init__(self, config, authmethod, authenticator, done): + ApplicationSession.__init__(self, config) + self._authmethod = authmethod + self._authenticator = authenticator + self._done = done + + def onConnect(self): + authextra = None + if self._authenticator: + authextra = self._authenticator.authextra + self.join( + self.config.realm, authmethods=[self._authmethod], authextra=authextra + ) + + def onChallenge(self, challenge): + return self._authenticator.on_challenge(self, challenge) + + def onJoin(self, details): + if not self._done.called: + self._done.callback(("joined", details, self._transport.transport_details)) + self.leave() + + def onLeave(self, details): + if not self._done.called: + self._done.callback(("left", details, None)) + self.disconnect() + + def onDisconnect(self): + if not self._done.called: + self._done.callback(("disconnected", None, None)) + + +class _TlsWampTests: + """ + WAMP authentication over TLS; mixed into one test case per transport (WebSocket, RawSocket). + """ + + transport = None + timeout = 20 + + def setUp(self): + self.pki = ExamplePKI(self.mktemp()) + self.observed = [] + self.cryptosign_key = cryptosign.CryptosignKey.from_bytes(b"\x01" * 32) + self.principals = { + expected_peer_cert(self.pki.client_certs["client_0"], {}, {})["sha1"]: ( + "client_0", + "backend", + ), + } + self.pubkeys = {self.cryptosign_key.public_key(): ("alice", "frontend")} + + def _server_options(self, tls_version): + server = ssl.PrivateCertificate.loadPEM( + self.pki.pem("server", "crt") + self.pki.pem("server", "key") + ) + ca = ssl.Certificate.loadPEM(self.pki.pem("ca", "crt")) + kwargs = {} + if tls_version: + kwargs = dict( + raiseMinimumTo=_TLS_VERSIONS[tls_version], + lowerMaximumSecurityTo=_TLS_VERSIONS[tls_version], + ) + return ssl.CertificateOptions( + privateKey=server.privateKey.original, + certificate=server.original, + verify=True, + caCerts=[ca.original], + requireCertificate=False, + **kwargs, + ) + + def _client_options(self, client): + ca = ssl.Certificate.loadPEM(self.pki.pem("ca", "crt")) + client_cert = None + if client: + client_cert = ssl.PrivateCertificate.loadPEM( + self.pki.pem(client, "crt") + self.pki.pem(client, "key") + ) + return ssl.optionsForClientTLS( + "localhost", trustRoot=ca, clientCertificate=client_cert + ) + + @inlineCallbacks + def _join(self, client, authmethod, authenticator=None, tls_version=None): + """ + Connect a client with the TLS client certificate ``client`` (or none), join with + ``authmethod``, and return ``(outcome, details, client transport details)``. + """ + server_closed = Deferred() + + def router_session(): + return FakeRouterSession( + principals=self.principals, + pubkeys=self.pubkeys, + observed=self.observed, + on_close=lambda: server_closed.called or server_closed.callback(None), + ) + + done = Deferred() + + def client_session(): + return _Client(ComponentConfig("realm1"), authmethod, authenticator, done) + + if self.transport == "websocket": + server_factory = WampWebSocketServerFactory(router_session) + else: + server_factory = WampRawSocketServerFactory(router_session) + + endpoint = SSL4ServerEndpoint( + reactor, 0, self._server_options(tls_version), interface="127.0.0.1" + ) + port = yield endpoint.listen(server_factory) + self.addCleanup(port.stopListening) + portno = port.getHost().port + + if self.transport == "websocket": + client_factory = WampWebSocketClientFactory( + client_session, url=f"wss://localhost:{portno}" + ) + else: + client_factory = WampRawSocketClientFactory(client_session) + + client_endpoint = SSL4ClientEndpoint( + reactor, "127.0.0.1", portno, self._client_options(client) + ) + proto = yield client_endpoint.connect(client_factory) + + result = yield done + # wait until both sides closed the connection + yield server_closed + if proto.transport.connected: + proto.transport.loseConnection() + return result + + def _aborts(self): + return [what for event, what in self.observed if event == "abort"] + + def _hello_details(self): + return [td for event, td in self.observed if event == "hello"][0] + + # TLS client certificate authentication (authmethod "tls") + + @inlineCallbacks + def _check_tls_auth_accepted(self, tls_version): + outcome, details, _ = yield self._join( + "client_0", "tls", tls_version=tls_version + ) + self.assertEqual(self._aborts(), []) + self.assertEqual(outcome, "joined") + self.assertEqual( + (details.authid, details.authrole, details.authmethod), + ("client_0", "backend", "tls"), + ) + + # what the router sees: the client certificate, after the TLS handshake + peer_cert = self._hello_details().peer_cert + cert = self.pki.client_certs["client_0"] + self.assertEqual(peer_cert["sha1"], expected_peer_cert(cert, {}, {})["sha1"]) + self.assertEqual(peer_cert["subject"]["cn"], "client_0") + + def test_tls_auth_accepted(self): + return self._check_tls_auth_accepted(None) + + def test_tls_auth_accepted_tls12(self): + return self._check_tls_auth_accepted("1.2") + + def test_tls_auth_accepted_tls13(self): + return self._check_tls_auth_accepted("1.3") + + @inlineCallbacks + def test_tls_auth_unknown_certificate(self): + """ + A client certificate the CA signed, but that is not a principal: denied, for that reason. + """ + outcome, details, _ = yield self._join("client_1", "tls") + self.assertEqual(outcome, "left") + sha1 = expected_peer_cert(self.pki.client_certs["client_1"], {}, {})["sha1"] + self.assertEqual(self._aborts(), [f"client certificate {sha1} not accepted"]) + + @inlineCallbacks + def test_tls_auth_no_certificate(self): + outcome, details, _ = yield self._join(None, "tls") + self.assertEqual(outcome, "left") + self.assertEqual(self._aborts(), ["no client certificate presented"]) + self.assertIsNone(self._hello_details().peer_cert) + + # WAMP-cryptosign with TLS channel binding + + @inlineCallbacks + def _check_cryptosign_channel_binding(self, tls_version): + authenticator = AuthCryptoSign( + privkey=binascii.b2a_hex(b"\x01" * 32).decode(), + authextra={"channel_binding": "tls-unique"}, + ) + outcome, details, client_td = yield self._join( + None, "cryptosign", authenticator, tls_version=tls_version + ) + self.assertEqual(self._aborts(), []) + self.assertEqual(outcome, "joined") + self.assertEqual((details.authid, details.authmethod), ("alice", "cryptosign")) + + # both ends of the connection have the same, non-zero channel ID + server_td = self._hello_details() + self.assertTrue(server_td.is_secure) + self.assertEqual( + server_td.channel_id["tls-unique"], client_td.channel_id["tls-unique"] + ) + self.assertNotEqual( + server_td.channel_id["tls-unique"], FakeRouterSession.ZERO_CHANNEL_ID + ) + + def test_cryptosign_channel_binding(self): + return self._check_cryptosign_channel_binding(None) + + def test_cryptosign_channel_binding_tls12(self): + return self._check_cryptosign_channel_binding("1.2") + + def test_cryptosign_channel_binding_tls13(self): + return self._check_cryptosign_channel_binding("1.3") + + @inlineCallbacks + def test_cryptosign_channel_binding_with_client_certificate(self): + """ + Channel binding also works when the client presents a TLS client certificate. + """ + authenticator = AuthCryptoSign( + privkey=binascii.b2a_hex(b"\x01" * 32).decode(), + authextra={"channel_binding": "tls-unique"}, + ) + outcome, details, _ = yield self._join("client_0", "cryptosign", authenticator) + self.assertEqual(self._aborts(), []) + self.assertEqual(outcome, "joined") + + +class TlsWebSocketTests(_TlsWampTests, unittest.TestCase): + transport = "websocket" + + +class TlsRawSocketTests(_TlsWampTests, unittest.TestCase): + transport = "rawsocket" From 723f0c2b066001998b1c95e724799f09f1dc193f Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 20:48:42 +0200 Subject: [PATCH 4/7] #2014: test_tx_choosereactor: patch sys.modules in place patch_modules() rebound sys.modules to a copy and "restored" it to another copy (since 073d33dc, 2015), so after the first test of test_tx_choosereactor the sys.modules dict was no longer the one the import system uses. Any module first imported lazily later in the same trial process then fails. On GitHub CI (all CPython jobs) this hit the new #2014 tests: pyOpenSSL's X509.has_expired() calls datetime.strptime(), whose lazy import of _strptime raised "KeyError: '_strptime'", which hid the actual failure. Locally _strptime happened to be imported earlier, so it did not show. Reproduced locally by running test_tx_choosereactor before test_tx_tls.ExtractPeerCertificateTests in one trial process (2x KeyError: '_strptime'); with mock.patch.dict(sys.modules), which patches and restores the dict in place, the same run fails with the expected AttributeError ('X509' object has no attribute 'get_extension'). Correction to the previous (red) commit message cf8edbc7: the asyncio test module has 22 tests (14 fail, 8 pass), not 26. Expected CI result: still RED, in the same 14 Twisted tests, now with the expected errors on CPython too (locally cpy311: 349 ok, 12 failures, 2 errors, no _strptime errors); test_tx_choosereactor passes. Note: This work was completed with AI assistance (Claude Code). --- .../twisted/test/test_tx_choosereactor.py | 19 ++++++++----------- 1 file changed, 8 insertions(+), 11 deletions(-) diff --git a/src/autobahn/twisted/test/test_tx_choosereactor.py b/src/autobahn/twisted/test/test_tx_choosereactor.py index a1e973d08..3ec00595c 100644 --- a/src/autobahn/twisted/test/test_tx_choosereactor.py +++ b/src/autobahn/twisted/test/test_tx_choosereactor.py @@ -25,7 +25,7 @@ ############################################################################### import sys -from unittest.mock import Mock +from unittest.mock import Mock, patch import twisted.internet from autobahn.twisted import choosereactor @@ -52,16 +52,13 @@ def patch_modules(self): Patch ``sys.modules`` so that Twisted believes there is no installed reactor. """ - old_modules = dict(sys.modules) - - new_modules = dict(sys.modules) - del new_modules["twisted.internet.reactor"] - - def _cleanup(): - sys.modules = old_modules - - self.addCleanup(_cleanup) - sys.modules = new_modules + # patch the sys.modules dict in place: rebinding sys.modules to a copy + # detaches it from the import system, and later lazy imports in the same + # process then fail (e.g. "KeyError: '_strptime'" in datetime.strptime) + patcher = patch.dict(sys.modules) + patcher.start() + self.addCleanup(patcher.stop) + del sys.modules["twisted.internet.reactor"] def test_unknown(self): """ From daf6c67d4153f18aa34403e6b3e667c48b31e36b Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 20:48:42 +0200 Subject: [PATCH 5/7] #2014: CI: run the asyncio unit tests also when the Twisted tests fail In the "Unit Tests" matrix jobs the asyncio step was skipped whenever the Twisted step failed, so the red run of #2014 showed only the 14 Twisted failures and never the 14 asyncio ones. The asyncio step now runs unless the job was cancelled (if: !cancelled()), so both backends always report. Expected CI result: RED. Unit Tests jobs: Twisted fails in the 14 #2014 tests (12 failures, 2 errors), and asyncio now runs and fails in the 14 #2014 tests (locally cpy311: 320 passed, 14 failed). ruff and ty 0.0.89 clean. Note: This work was completed with AI assistance (Claude Code). --- .github/workflows/main.yml | 2 ++ 1 file changed, 2 insertions(+) diff --git a/.github/workflows/main.yml b/.github/workflows/main.yml index 3b57ebe33..f052cba67 100644 --- a/.github/workflows/main.yml +++ b/.github/workflows/main.yml @@ -193,7 +193,9 @@ jobs: - name: Unit test - Twisted / trial run: just test-twisted ${{ matrix.python-env }} + # run on a Twisted failure too, so that both backends always report - name: Unit test - asyncio / pytest + if: ${{ !cancelled() }} run: just test-asyncio ${{ matrix.python-env }} # Last: its "install" dependency replaces the editable install. From 9c48ea85803ce4654415f18d0e4add0c45408067 Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 21:39:56 +0200 Subject: [PATCH 6/7] #2014: red: tls-unique on Twisted must follow RFC 5929, like asyncio tls-unique (RFC 5929) is the first TLS Finished message of the handshake, in a full handshake the client's; Python's ssl module, which autobahn uses on asyncio, returns exactly that. Checked on one connection (TLS 1.2 and 1.3): Twisted's transport_channel_id() gives sha256(server Finished) when is_server is set correctly (WebSocket), and with is_server=None (RawSocket) sha256(client Finished) on the server but sha256(server Finished) on the client. So WAMP-cryptosign channel binding fails between a Twisted router (Crossbar.io) and asyncio clients over WebSocket, and between Twisted RawSocket peers; it works Twisted<->Twisted over WebSocket (both sides use the server's Finished) and Crossbar.io RawSocket<->asyncio (both use the client's). New tests in TlsUniqueTests (autobahn/twisted/test/test_tx_tls.py), 5 FAIL: - a Twisted TLS server and a Python ssl client, TLS 1.2 and TLS 1.3: the server's channel ID must equal the client's - a Twisted TLS client and a Python ssl server, TLS 1.2 and TLS 1.3: same - the WAMP WebSocket and RawSocket server / client protocols set is_server (RawSocket: None) Expected CI result: RED. Twisted trial now fails in 19 #2014 tests (locally cpy311: 349 ok, 17 failures, 2 errors), asyncio unchanged in the 14 #2014 tests (320 passed, 14 failed); ruff and ty 0.0.89 clean. Note: This work was completed with AI assistance (Claude Code). --- src/autobahn/twisted/test/test_tx_tls.py | 157 ++++++++++++++++++++++- 1 file changed, 155 insertions(+), 2 deletions(-) diff --git a/src/autobahn/twisted/test/test_tx_tls.py b/src/autobahn/twisted/test/test_tx_tls.py index c72af8755..cac395594 100644 --- a/src/autobahn/twisted/test/test_tx_tls.py +++ b/src/autobahn/twisted/test/test_tx_tls.py @@ -14,6 +14,9 @@ """ import binascii +import hashlib +import socket +import ssl as stdlib_ssl import txaio @@ -23,7 +26,9 @@ from twisted.internet import reactor, ssl from twisted.internet.defer import Deferred, inlineCallbacks from twisted.internet.endpoints import SSL4ClientEndpoint, SSL4ServerEndpoint -from twisted.internet.interfaces import ISSLTransport +from twisted.internet.interfaces import IHandshakeListener, ISSLTransport +from twisted.internet.protocol import Factory, Protocol +from twisted.internet.threads import deferToThread from twisted.trial import unittest from zope.interface import implementer @@ -35,13 +40,17 @@ ) from autobahn.twisted.rawsocket import ( WampRawSocketClientFactory, + WampRawSocketClientProtocol, WampRawSocketServerFactory, + WampRawSocketServerProtocol, ) -from autobahn.twisted.util import extract_peer_certificate +from autobahn.twisted.util import extract_peer_certificate, transport_channel_id from autobahn.twisted.wamp import ApplicationSession from autobahn.twisted.websocket import ( WampWebSocketClientFactory, + WampWebSocketClientProtocol, WampWebSocketServerFactory, + WampWebSocketServerProtocol, ) from autobahn.wamp import cryptosign from autobahn.wamp.auth import AuthCryptoSign @@ -361,3 +370,147 @@ class TlsWebSocketTests(_TlsWampTests, unittest.TestCase): class TlsRawSocketTests(_TlsWampTests, unittest.TestCase): transport = "rawsocket" + + +_STDLIB_TLS_VERSIONS = { + "1.2": stdlib_ssl.TLSVersion.TLSv1_2, + "1.3": stdlib_ssl.TLSVersion.TLSv1_3, +} + + +def _stdlib_channel_id(tls_socket): + # what autobahn uses on asyncio: Python's ssl module, which implements RFC 5929 + return hashlib.sha256(tls_socket.get_channel_binding("tls-unique")).digest() + + +class TlsUniqueTests(unittest.TestCase): + """ + The TLS channel ID ``tls-unique`` (RFC 5929) is the first TLS Finished message of the + handshake, in a full handshake the client's. On both sides of a connection, Twisted must + compute the same channel ID as Python's ``ssl`` module (which autobahn uses on asyncio), + so that WAMP-cryptosign with channel binding works between Twisted and asyncio peers. + """ + + timeout = 20 + + def setUp(self): + self.pki = ExamplePKI(self.mktemp()) + + @inlineCallbacks + def _check_twisted_server(self, tls_version): + handshake = Deferred() + closed = Deferred() + + @implementer(IHandshakeListener) + class Server(Protocol): + def handshakeCompleted(self_): + handshake.callback( + transport_channel_id(self_.transport, True, "tls-unique") + ) + + def connectionLost(self_, reason): + closed.callback(None) + + server = ssl.PrivateCertificate.loadPEM( + self.pki.pem("server", "crt") + self.pki.pem("server", "key") + ) + version = _TLS_VERSIONS[tls_version] + options = ssl.CertificateOptions( + privateKey=server.privateKey.original, + certificate=server.original, + raiseMinimumTo=version, + lowerMaximumSecurityTo=version, + ) + port = yield SSL4ServerEndpoint( + reactor, 0, options, interface="127.0.0.1" + ).listen(Factory.forProtocol(Server)) + self.addCleanup(port.stopListening) + + def stdlib_client(): + ctx = stdlib_ssl.SSLContext(stdlib_ssl.PROTOCOL_TLS_CLIENT) + ctx.load_verify_locations(self.pki.path("ca", "crt")) + ctx.minimum_version = ctx.maximum_version = _STDLIB_TLS_VERSIONS[ + tls_version + ] + with socket.create_connection(("127.0.0.1", port.getHost().port)) as sock: + with ctx.wrap_socket(sock, server_hostname="localhost") as tls_socket: + # TLS 1.3: the server completes the handshake on the client's Finished + tls_socket.sendall(b"x") + return _stdlib_channel_id(tls_socket) + + client_id = yield deferToThread(stdlib_client) + server_id = yield handshake + yield closed + self.assertEqual(server_id, client_id) + + def test_twisted_server_tls12(self): + return self._check_twisted_server("1.2") + + def test_twisted_server_tls13(self): + return self._check_twisted_server("1.3") + + @inlineCallbacks + def _check_twisted_client(self, tls_version): + handshake = Deferred() + closed = Deferred() + + @implementer(IHandshakeListener) + class Client(Protocol): + def handshakeCompleted(self_): + handshake.callback( + transport_channel_id(self_.transport, False, "tls-unique") + ) + self_.transport.loseConnection() + + def connectionLost(self_, reason): + closed.callback(None) + + ctx = stdlib_ssl.SSLContext(stdlib_ssl.PROTOCOL_TLS_SERVER) + ctx.load_cert_chain( + self.pki.path("server", "crt"), self.pki.path("server", "key") + ) + ctx.minimum_version = ctx.maximum_version = _STDLIB_TLS_VERSIONS[tls_version] + listener = socket.create_server(("127.0.0.1", 0)) + self.addCleanup(listener.close) + + def stdlib_server(): + conn, _ = listener.accept() + with ctx.wrap_socket(conn, server_side=True) as tls_socket: + channel_id = _stdlib_channel_id(tls_socket) + # wait for the client to close the connection + while True: + try: + if not tls_socket.recv(1): + break + except (OSError, stdlib_ssl.SSLError): + break + return channel_id + + d_server = deferToThread(stdlib_server) + options = ssl.optionsForClientTLS( + "localhost", trustRoot=ssl.Certificate.loadPEM(self.pki.pem("ca", "crt")) + ) + endpoint = SSL4ClientEndpoint( + reactor, "127.0.0.1", listener.getsockname()[1], options + ) + yield endpoint.connect(Factory.forProtocol(Client)) + client_id = yield handshake + server_id = yield d_server + yield closed + self.assertEqual(client_id, server_id) + + def test_twisted_client_tls12(self): + return self._check_twisted_client("1.2") + + def test_twisted_client_tls13(self): + return self._check_twisted_client("1.3") + + def test_wamp_protocols_know_their_side(self): + """ + The channel ID depends on the side of the connection, which the transport details take + from the protocol's ``is_server``. + """ + self.assertIs(WampWebSocketServerProtocol.is_server, True) + self.assertIs(WampWebSocketClientProtocol.is_server, False) + self.assertIs(WampRawSocketServerProtocol.is_server, True) + self.assertIs(WampRawSocketClientProtocol.is_server, False) From 37b3d91ebfd6358de1f96b971b2c6d313bd017f9 Mon Sep 17 00:00:00 2001 From: Tobias Oberstein Date: Tue, 20 Oct 2026 22:40:01 +0200 Subject: [PATCH 7/7] #2014: green: TLS peer certificates and tls-unique on Twisted and asyncio Peer certificate: - new autobahn.util.peer_certificate_details(): the peer_cert dict, read with cryptography (pyOpenSSL removed X509.get_extension in 26.2.0 and deprecates the rest of its X509 API). Same keys and formats as before; extensions[].value: a subject / issuer alternative name as OpenSSL prints it (DNS:localhost, IP Address:..), other extensions as cryptography prints them; signature_algorithm "ed25519" instead of "ED25519". Never raises on certificate content: unreadable extensions / subject / issuer are empty - Twisted extract_peer_certificate() uses it (to_cryptography(); hash from pyOpenSSL's subject_name_hash(), which is not deprecated) - Twisted WebSocket (_onOpen) and RawSocket (_on_handshake_complete) read peer_cert after the TLS handshake, next to the channel ID; it was read in connectionMade(), before the handshake, and so was always None - asyncio: is_secure from the SSL object (was from "peercert", so a server without client certificate was not secure and had no channel ID), and new autobahn.asyncio.util.extract_peer_certificate() fills peer_cert (was always None) tls-unique (RFC 5929, as the user decided): - Twisted transport_channel_id() uses the first Finished message of the handshake (the client's): server: get_peer_finished(), client: get_finished(), as CPython's ssl module does. It used the server's - Twisted RawSocket server / client protocols set is_server (was None, so the client side took the wrong message); asyncio WebSocket protocols too - PyPy workaround: PyPy's ssl get_channel_binding("tls-unique") misses the negation CPython has (reused ^ socket_type instead of ^ !socket_type) and returns the server's Finished message; its tls_unique_cb() is correct. New autobahn.util.tls_unique() uses that on PyPy; asyncio uses tls_unique(), and TlsUniqueTests compare Twisted against it, so they pass on PyPy too (they failed there against the plain ssl module, as on the pypy311 job of the previous red push). PyPy issue drafted. Changelog: 26.10.2 section, with the extension value format, the tls-unique change (Twisted peers on <= 26.10.1 and >= 26.10.2 no longer agree over WebSocket when they use channel binding) and the PyPy workaround. This is substantially new code (peer_certificate_details, tls_unique and the asyncio extraction) written by AI on request; it needs human review. Expected CI result: GREEN. Locally: Twisted trial cpy311 368 ok / pypy311 367 ok, asyncio pytest cpy311 334 passed / pypy311 333 passed, all #2014 tests pass on both; import smoke test 35 passed; ruff and ty 0.0.89 clean; docs build ok, no new warnings. Note: This work was completed with AI assistance (Claude Code). --- docs/changelog.rst | 18 ++++ src/autobahn/asyncio/util.py | 41 ++++++-- src/autobahn/asyncio/websocket.py | 4 + src/autobahn/twisted/rawsocket.py | 17 ++- src/autobahn/twisted/test/test_tx_tls.py | 4 +- src/autobahn/twisted/util.py | 80 +++++--------- src/autobahn/twisted/websocket.py | 9 +- src/autobahn/util.py | 126 +++++++++++++++++++++++ 8 files changed, 235 insertions(+), 64 deletions(-) diff --git a/docs/changelog.rst b/docs/changelog.rst index 7891e4b19..054429207 100644 --- a/docs/changelog.rst +++ b/docs/changelog.rst @@ -5,6 +5,24 @@ Changelog ========= +26.10.2 +------- + +**WAMP over TLS** + +* Fix reading the TLS peer certificate (``TransportDetails.peer_cert``) with pyOpenSSL 26.2.0 and later: ``autobahn.twisted.util.extract_peer_certificate()`` raised ``AttributeError: 'X509' object has no attribute 'get_extension'`` for every certificate with extensions, as pyOpenSSL removed its X.509 extension API (and deprecates the rest of its ``X509`` API). The certificate is now read with ``cryptography``, in the new backend-neutral ``autobahn.util.peer_certificate_details()``, which also never raises on certificate content it cannot parse (it is peer data, read before authentication): extensions, or a subject or issuer, that cannot be read are left empty. The ``peer_cert`` dict keeps its keys and formats, except: **⚠** ``extensions[].value`` was OpenSSL's text; a subject or issuer alternative name is still given as OpenSSL prints it (``DNS:localhost, IP Address:127.0.0.1``), other extensions now as ``cryptography`` prints them; and ``signature_algorithm`` is ``ed25519`` instead of ``ED25519`` for Ed25519 certificates (#2014) +* Fix the TLS peer certificate never being available on Twisted WebSocket and RawSocket connections: it was read when the TCP connection was made, before the TLS handshake, and so was always ``None``. It is now read when the TLS handshake is complete, together with the TLS channel ID. With this, TLS client certificate authentication (Crossbar.io authmethod ``tls``, and dynamic authenticators reading ``transport.peer_cert``) works again (#2014) +* Fix asyncio: fill ``TransportDetails.peer_cert`` (it was always ``None``), and treat a TLS server connection without a client certificate as secure, with a TLS channel ID: ``is_secure`` was derived from the peer certificate, so WAMP-cryptosign with channel binding failed on asyncio servers (#2014) +* **⚠ behaviour change** — the TLS channel ID ``tls-unique`` on Twisted now follows RFC 5929 (the first TLS Finished message of the handshake, in a full handshake the client's), as Python's ``ssl`` module, and so autobahn on asyncio, already did. Twisted used the server's Finished message, so WAMP-cryptosign with channel binding failed between Twisted and asyncio peers over WebSocket, and between Twisted RawSocket peers (whose protocols did not set ``is_server``, so the client side used the wrong message). Twisted peers on autobahn 26.10.1 or older and on 26.10.2 no longer agree on the channel ID over WebSocket: upgrade both sides of connections that use channel binding (#2014) +* Work around a PyPy bug: PyPy's ``ssl`` module (up to at least PyPy 7.3.23) returns the server's TLS Finished message from ``get_channel_binding("tls-unique")`` instead of the first (the client's), so autobahn on asyncio on PyPy disagreed with all other peers on the channel ID. The new ``autobahn.util.tls_unique()`` uses PyPy's own ``tls_unique_cb()`` on PyPy, which returns the right message (#2014) +* ``TransportDetails.is_server`` is now set on Twisted RawSocket and asyncio WebSocket connections (it was ``None``) (#2014) + +**Tests & CI/CD** + +* Add tests for WAMP over TLS on Twisted and asyncio, over WebSocket and RawSocket, on TLS 1.2 and 1.3, with certificates created per test: the peer certificate, TLS client certificate authentication, WAMP-cryptosign with channel binding, and ``tls-unique`` against Python's ``ssl`` module (#2014) +* Fix ``test_tx_choosereactor`` replacing ``sys.modules`` with a copy, which broke later lazy imports in the same test process (``KeyError: '_strptime'``) (#2014) +* CI runs the asyncio unit tests also when the Twisted unit tests fail, so that both backends always report (#2014) + 26.10.1 ------- diff --git a/src/autobahn/asyncio/util.py b/src/autobahn/asyncio/util.py index b923e4c85..c54166072 100644 --- a/src/autobahn/asyncio/util.py +++ b/src/autobahn/asyncio/util.py @@ -29,8 +29,9 @@ import warnings from asyncio import sleep # noqa from subprocess import Popen -from typing import Optional +from typing import Any, Optional +from autobahn.util import peer_certificate_details, tls_unique from autobahn.wamp.types import TransportDetails @@ -70,6 +71,7 @@ def get_or_create_event_loop() -> asyncio.AbstractEventLoop: "peer2str", "transport_channel_id", "create_transport_details", + "extract_peer_certificate", ) @@ -109,9 +111,9 @@ def transport_channel_id( "TLS transport channel_id for tls-unique requested, but get_channel_binding not found on ssl_obj" ) - # https://python.readthedocs.io/en/latest/library/ssl.html#ssl.SSLSocket.get_channel_binding - # https://tools.ietf.org/html/rfc5929.html - tls_finished_msg: bytes = ssl_obj.get_channel_binding(cb_type="tls-unique") + # https://docs.python.org/3/library/ssl.html#ssl.SSLSocket.get_channel_binding + # https://www.rfc-editor.org/rfc/rfc5929.html (and see tls_unique() on PyPy) + tls_finished_msg = tls_unique(ssl_obj) if type(tls_finished_msg) != bytes: return b"\x00" * 32 @@ -175,18 +177,43 @@ def get_serializers(): return serializers +def extract_peer_certificate(ssl_obj) -> dict[str, Any] | None: + """ + Extract information on the TLS peer certificate (if the peer presented one) from the + SSL object of an asyncio TLS transport (see :func:`autobahn.util.peer_certificate_details`). + + :param ssl_obj: The ``ssl.SSLObject`` of the TLS transport. + :returns: The certificate information, or ``None`` if the peer presented no certificate. + """ + der = ssl_obj.getpeercert(binary_form=True) + if not der or not isinstance(der, bytes): + return None + from cryptography import x509 + + try: + cert = x509.load_der_x509_certificate(der) + except ValueError: + # OpenSSL accepted the certificate in the TLS handshake, but cryptography cannot parse it + return None + return peer_certificate_details(cert) + + def create_transport_details(transport, is_server: bool) -> TransportDetails: # Internal helper. Base class calls this to create a TransportDetails peer = peer2str(transport) - # https://docs.python.org/3.9/library/asyncio-protocol.html?highlight=get_extra_info#asyncio.BaseTransport.get_extra_info - is_secure = transport.get_extra_info("peercert", None) is not None + # https://docs.python.org/3/library/asyncio-protocol.html#asyncio.BaseTransport.get_extra_info + # a TLS connection has an SSL object, whether or not the peer presented a certificate + # ("peercert" is only set for a peer certificate, so a server without client certificate + # was not "secure" and had no channel ID, #2014) + ssl_obj = transport.get_extra_info("ssl_object", None) + is_secure = ssl_obj is not None if is_secure: channel_id = { "tls-unique": transport_channel_id(transport, is_server, "tls-unique"), } channel_type = TransportDetails.CHANNEL_TYPE_TLS - peer_cert = None + peer_cert = extract_peer_certificate(ssl_obj) else: channel_id = {} channel_type = TransportDetails.CHANNEL_TYPE_TCP diff --git a/src/autobahn/asyncio/websocket.py b/src/autobahn/asyncio/websocket.py index 9d0ad0d85..a0c5188be 100644 --- a/src/autobahn/asyncio/websocket.py +++ b/src/autobahn/asyncio/websocket.py @@ -231,6 +231,8 @@ class WebSocketServerProtocol( log = txaio.make_logger() + is_server = True + @public class WebSocketClientProtocol( @@ -246,6 +248,8 @@ class WebSocketClientProtocol( log = txaio.make_logger() + is_server = False + def _onConnect(self, response): res = self.onConnect(response) self.log.debug("{func}: {res}", func=hltype(self._onConnect), res=res) diff --git a/src/autobahn/twisted/rawsocket.py b/src/autobahn/twisted/rawsocket.py index 05766e883..b98db516f 100644 --- a/src/autobahn/twisted/rawsocket.py +++ b/src/autobahn/twisted/rawsocket.py @@ -31,7 +31,11 @@ import txaio from autobahn.exception import PayloadExceededError -from autobahn.twisted.util import create_transport_details, transport_channel_id +from autobahn.twisted.util import ( + create_transport_details, + extract_peer_certificate, + transport_channel_id, +) from autobahn.util import _LazyHexFormatter, public from autobahn.wamp.exception import ( InvalidUriError, @@ -127,13 +131,16 @@ def _on_handshake_complete(self): try: if self._transport_details.is_secure: # now that the TLS opening handshake is complete, the actual TLS channel ID - # will be available. make sure to set it! + # and the peer certificate will be available. make sure to set them! channel_id = { "tls-unique": transport_channel_id( self.transport, self._transport_details.is_server, "tls-unique" ), } self._transport_details.channel_id = channel_id + self._transport_details.peer_cert = extract_peer_certificate( + self.transport + ) self._session = self.factory._factory() self.log.debug( @@ -320,6 +327,9 @@ class WampRawSocketServerProtocol(WampRawSocketProtocol): * :class:`autobahn.wamp.interfaces.ITransport` """ + # the TLS channel ID (tls-unique) depends on the side of the connection (#2014) + is_server = True + def dataReceived(self, data): if self._handshake_complete: WampRawSocketProtocol.dataReceived(self, data) @@ -423,6 +433,9 @@ class WampRawSocketClientProtocol(WampRawSocketProtocol): * :class:`autobahn.wamp.interfaces.ITransport` """ + # the TLS channel ID (tls-unique) depends on the side of the connection (#2014) + is_server = False + def connectionMade(self): WampRawSocketProtocol.connectionMade(self) self._serializer = copy.copy(self.factory._serializer) diff --git a/src/autobahn/twisted/test/test_tx_tls.py b/src/autobahn/twisted/test/test_tx_tls.py index cac395594..add1b8d16 100644 --- a/src/autobahn/twisted/test/test_tx_tls.py +++ b/src/autobahn/twisted/test/test_tx_tls.py @@ -52,6 +52,7 @@ WampWebSocketServerFactory, WampWebSocketServerProtocol, ) +from autobahn.util import tls_unique from autobahn.wamp import cryptosign from autobahn.wamp.auth import AuthCryptoSign from autobahn.wamp.types import ComponentConfig @@ -380,7 +381,8 @@ class TlsRawSocketTests(_TlsWampTests, unittest.TestCase): def _stdlib_channel_id(tls_socket): # what autobahn uses on asyncio: Python's ssl module, which implements RFC 5929 - return hashlib.sha256(tls_socket.get_channel_binding("tls-unique")).digest() + # (on PyPy with a workaround, see autobahn.util.tls_unique()) + return hashlib.sha256(tls_unique(tls_socket)).digest() class TlsUniqueTests(unittest.TestCase): diff --git a/src/autobahn/twisted/util.py b/src/autobahn/twisted/util.py index 297108d9e..7141a27ad 100644 --- a/src/autobahn/twisted/util.py +++ b/src/autobahn/twisted/util.py @@ -31,6 +31,7 @@ import threading from typing import Any +from autobahn.util import peer_certificate_details from autobahn.wamp.types import TransportDetails from twisted.internet.address import IPv4Address, UNIXAddress from twisted.internet.defer import Deferred @@ -183,21 +184,24 @@ def transport_channel_id( connection: Connection = transport.getHandle() assert connection and isinstance(connection, Connection) - # Obtain latest TLS Finished message that we expected from peer, or None if handshake is not completed. + # tls-unique (RFC 5929, section 3.1) is the first TLS Finished message of the latest + # handshake: the client's in a full handshake, the server's in an abbreviated one + # (session resumption). This is what Python's ssl module returns (which autobahn uses + # on asyncio), so both backends agree (#2014). Twisted does not resume TLS sessions + # by default, and pyOpenSSL has no API to tell, so this assumes a full handshake. # http://www.pyopenssl.org/en/stable/api/ssl.html#OpenSSL.SSL.Connection.get_peer_finished - is_not_resumed = True + is_resumed = False if channel_id_type == "tls-unique": # see also: https://bugs.python.org/file22646/tls_channel_binding.patch - if is_server != is_not_resumed: - # for routers (=servers) XOR new sessions, the channel ID is based on the TLS Finished message we - # expected to receive from the client: contents of the message or None if the TLS handshake has - # not yet completed. - tls_finished_msg = connection.get_peer_finished() - else: - # for clients XOR resumed sessions, the channel ID is based on the TLS Finished message we sent - # to the router (=server): contents of the message or None if the TLS handshake has not yet completed. + if is_resumed != (not is_server): + # a client in a full handshake, or a server in a resumed one: the TLS Finished + # message we sent, or None if the TLS handshake has not yet completed tls_finished_msg = connection.get_finished() + else: + # a server (router) in a full handshake, or a client in a resumed one: the TLS + # Finished message we received, or None if the TLS handshake has not yet completed + tls_finished_msg = connection.get_peer_finished() if tls_finished_msg is None: # this can occur when: @@ -237,7 +241,8 @@ def extract_peer_certificate( TLS client certificate). :param transport: The secure transport from which to extract the peer certificate (if present). - :returns: If the peer provided a certificate, the parsed certificate information set. + :returns: If the peer provided a certificate, the parsed certificate information set + (see :func:`autobahn.util.peer_certificate_details`). """ # check if the Twisted transport is a TLSMemoryBIOProtocol if not ( @@ -248,47 +253,18 @@ def extract_peer_certificate( cert = transport.getPeerCertificate() if cert: - # extract x509 name components from an OpenSSL X509Name object - def maybe_bytes(_value): - if isinstance(_value, bytes): - return _value.decode("utf8") - else: - return _value - - result = { - "md5": "{}".format(maybe_bytes(cert.digest("md5"))).upper(), - "sha1": "{}".format(maybe_bytes(cert.digest("sha1"))).upper(), - "sha256": "{}".format(maybe_bytes(cert.digest("sha256"))).upper(), - "expired": bool(cert.has_expired()), - "hash": maybe_bytes(cert.subject_name_hash()), - "serial": int(cert.get_serial_number()), - "signature_algorithm": maybe_bytes(cert.get_signature_algorithm()), - "version": int(cert.get_version()), - "not_before": maybe_bytes(cert.get_notBefore()), - "not_after": maybe_bytes(cert.get_notAfter()), - "extensions": [], - } - - for i in range(cert.get_extension_count()): - ext = cert.get_extension(i) - ext_info = { - "name": f"{maybe_bytes(ext.get_short_name())}", - "value": f"{maybe_bytes(ext)}", - "critical": ext.get_critical() != 0, - } - result["extensions"].append(ext_info) - - for entity, name in [ - ("subject", cert.get_subject()), - ("issuer", cert.get_issuer()), - ]: - result[entity] = {} - for key, value in name.get_components(): - key = maybe_bytes(key) - value = maybe_bytes(value) - result[entity][f"{key}".lower()] = f"{value}" - - return result + # pyOpenSSL's X509 certificate API is deprecated, and partly removed + # (pyOpenSSL 26.2.0), so read the certificate with cryptography (#2014) + try: + crypto_cert = cert.to_cryptography() + except ValueError: + # OpenSSL accepted the certificate in the TLS handshake, but cryptography + # cannot parse it + return None + return peer_certificate_details( + crypto_cert, subject_name_hash=cert.subject_name_hash() + ) + return None def create_transport_details( diff --git a/src/autobahn/twisted/websocket.py b/src/autobahn/twisted/websocket.py index 125f9f726..87dd95929 100644 --- a/src/autobahn/twisted/websocket.py +++ b/src/autobahn/twisted/websocket.py @@ -35,7 +35,11 @@ txaio.use_twisted() import twisted.internet.protocol -from autobahn.twisted.util import create_transport_details, transport_channel_id +from autobahn.twisted.util import ( + create_transport_details, + extract_peer_certificate, + transport_channel_id, +) from autobahn.util import _is_tls_error, _maybe_tls_reason, hltype, hlval, public from autobahn.wamp import websocket from autobahn.wamp.types import TransportDetails @@ -373,13 +377,14 @@ def _closeConnection(self, abort: bool=False) -> None: def _onOpen(self) -> None: if self._transport_details.is_secure: # now that the TLS opening handshake is complete, the actual TLS channel ID - # will be available. make sure to set it! + # and the peer certificate will be available. make sure to set them! channel_id = { "tls-unique": transport_channel_id( self.transport, self._transport_details.is_server, "tls-unique" ), } self._transport_details.channel_id = channel_id + self._transport_details.peer_cert = extract_peer_certificate(self.transport) self.onOpen() diff --git a/src/autobahn/util.py b/src/autobahn/util.py index ff175ef9e..fcb6e895d 100644 --- a/src/autobahn/util.py +++ b/src/autobahn/util.py @@ -39,6 +39,7 @@ from collections import OrderedDict from datetime import datetime, timedelta from pprint import pformat +from typing import Any import txaio @@ -70,7 +71,9 @@ "machine_id", "newid", "parse_keyfile", + "peer_certificate_details", "public", + "tls_unique", "rid", "rtime", "utcnow", @@ -1079,3 +1082,126 @@ def parse_keyfile(key_path: str, private: bool = True) -> OrderedDict: ) tags[tag] = value return tags + + +def tls_unique(ssl_object) -> bytes | None: + """ + The TLS channel binding ``tls-unique`` (RFC 5929) of a connection made with Python's + ``ssl`` module: the first TLS Finished message of the latest handshake, that is the + client's in a full handshake. + + PyPy's ``ssl`` module (at least up to PyPy 7.3.23) returns the other Finished message + from ``get_channel_binding("tls-unique")`` (the server's in a full handshake), while its + SSL objects' own ``tls_unique_cb()`` returns the right one; on PyPy, this uses that. + + :param ssl_object: The ``ssl.SSLObject`` (asyncio) or ``ssl.SSLSocket``. + :returns: The TLS Finished message, or ``None`` if the TLS handshake is not complete. + """ + if sys.implementation.name == "pypy": + tls_unique_cb = getattr(getattr(ssl_object, "_sslobj", None), "tls_unique_cb", None) + if tls_unique_cb is not None: + return tls_unique_cb() + return ssl_object.get_channel_binding("tls-unique") + + +def _oid_name(oid) -> str: + # OpenSSL's short name, or the dotted OID for an OID cryptography does not know + name = oid._name + return oid.dotted_string if name == "Unknown OID" else name + + +def _x509_name(name) -> dict[str, str]: + # as pyOpenSSL's X509Name.get_components() gave them: OpenSSL short names, lower case + result = {} + for attr in name: + key = attr.rfc4514_attribute_name + if key[:1].isdigit(): + # no RFC 4514 short name, e.g. emailAddress + key = _oid_name(attr.oid) + value = attr.value + if isinstance(value, bytes): + value = value.decode("utf8", errors="replace") + result[key.lower()] = value + return result + + +def _general_name(gn) -> str: + # as OpenSSL prints a subject / issuer alternative name + from cryptography import x509 + + if isinstance(gn, x509.DNSName): + return f"DNS:{gn.value}" + if isinstance(gn, x509.RFC822Name): + return f"email:{gn.value}" + if isinstance(gn, x509.UniformResourceIdentifier): + return f"URI:{gn.value}" + if isinstance(gn, x509.IPAddress): + return f"IP Address:{gn.value}" + if isinstance(gn, x509.RegisteredID): + return f"Registered ID:{gn.value.dotted_string}" + if isinstance(gn, x509.DirectoryName): + return f"DirName:{gn.value.rfc4514_string()}" + return "othername:" + + +def _x509_extension(ext) -> dict[str, Any]: + from cryptography import x509 + + if isinstance(ext.value, (x509.SubjectAlternativeName, x509.IssuerAlternativeName)): + value = ", ".join(_general_name(gn) for gn in ext.value) + else: + value = str(ext.value) + return {"name": _oid_name(ext.oid), "value": value, "critical": ext.critical} + + +def peer_certificate_details( + cert, subject_name_hash: int | None = None +) -> dict[str, Any]: + """ + Information on a TLS peer certificate, as provided in + :attr:`autobahn.wamp.types.TransportDetails.peer_cert`. + + The certificate is data the peer controls, read before authentication, so this + does not raise on certificate content that cannot be parsed: extensions, or a + subject or issuer, that cannot be read are left empty. + + :param cert: The peer certificate (``cryptography.x509.Certificate``). + :param subject_name_hash: OpenSSL's hash of the subject name, when available. + :returns: Fingerprints (``md5``, ``sha1``, ``sha256``, as upper case hex with colons), + ``expired``, ``hash``, ``serial``, ``signature_algorithm``, ``version``, + ``not_before`` and ``not_after`` (``YYYYMMDDhhmmssZ``), ``subject`` and ``issuer`` + (OpenSSL short names in lower case, e.g. ``cn``), and ``extensions``, a list of + ``name``, ``value`` and ``critical``. The value of a subject or issuer alternative + name is given as OpenSSL prints it (e.g. ``DNS:localhost, IP Address:127.0.0.1``), + that of other extensions as ``cryptography`` prints them. + """ + from datetime import UTC + + from cryptography.hazmat.primitives import hashes + + def fingerprint(algorithm) -> str: + return ":".join(f"{b:02X}" for b in cert.fingerprint(algorithm)) + + result: dict[str, Any] = { + "md5": fingerprint(hashes.MD5()), # noqa: S303 - a certificate ID, not for security + "sha1": fingerprint(hashes.SHA1()), # noqa: S303 - a certificate ID, not for security + "sha256": fingerprint(hashes.SHA256()), + "expired": datetime.now(UTC) > cert.not_valid_after_utc, + "hash": subject_name_hash, + "serial": cert.serial_number, + "signature_algorithm": _oid_name(cert.signature_algorithm_oid), + "version": cert.version.value, + "not_before": cert.not_valid_before_utc.strftime("%Y%m%d%H%M%SZ"), + "not_after": cert.not_valid_after_utc.strftime("%Y%m%d%H%M%SZ"), + } + for entity in ["subject", "issuer"]: + try: + result[entity] = _x509_name(getattr(cert, entity)) + except ValueError: + result[entity] = {} + try: + result["extensions"] = [_x509_extension(ext) for ext in cert.extensions] + except Exception: + # e.g. a duplicate extension (cryptography.x509.DuplicateExtension) + result["extensions"] = [] + return result