ssl: emit the full peer certificate dict

Use the same certificate dict helper for getpeercert() and Certificate.get_info() so OCSP, caIssuers, and CRL URLs are present. Clear CHANNEL_BINDING_TYPES on the rustls backend, and add a MemoryBIO leftover-plaintext snippet. Assisted-by: Claude
youknowone · Sep 5, 2026 · Sep 5, 2026 · Sep 5, 2026 · Sep 5, 2026 · Sep 5, 2026
commit e939d92e4b38ff2d2cd30ce7bc5cd742f3b116a1
diff --git a/Lib/ssl.py b/Lib/ssl.py
@@ -268,6 +268,9 @@ class _TLSMessageType:
 socket_error = OSError  # keep that public name in module namespace
 
 CHANNEL_BINDING_TYPES = ['tls-unique']
+# TODO: RUSTPYTHON; rustls does not expose TLS Finished messages
+if "rustls" in OPENSSL_VERSION:
+    CHANNEL_BINDING_TYPES = []
 
 HAS_NEVER_CHECK_COMMON_NAME = hasattr(_ssl, 'HOSTFLAG_NEVER_CHECK_SUBJECT')
 

diff --git a/Lib/test/test_asyncio/test_events.py b/Lib/test/test_asyncio/test_events.py
@@ -1289,7 +1289,6 @@ def test_create_unix_server_ssl_verified(self):
         server.close()
         self.loop.run_until_complete(proto.done)
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON; - SSL peer certificate format differs
     @unittest.skipIf(ssl is None, 'No ssl module')
     def test_create_server_ssl_verified(self):
         proto = MyProto(loop=self.loop)

diff --git a/Lib/test/test_ssl.py b/Lib/test/test_ssl.py
@@ -4192,7 +4192,6 @@ def test_default_ecdh_curve(self):
                 s.connect((HOST, server.port))
                 self.assertIn("ECDH", s.cipher()[0])
 
-    @unittest.expectedFailure  # TODO: RUSTPYTHON
     @unittest.skipUnless("tls-unique" in ssl.CHANNEL_BINDING_TYPES,
                          "'tls-unique' channel binding not available")
     def test_tls_unique_channel_binding(self):

diff --git a/crates/stdlib/src/ssl.rs b/crates/stdlib/src/ssl.rs
@@ -3723,11 +3723,7 @@ mod _ssl {
                 return Ok(Some(vm.ctx.new_dict().into()));
             }
 
-            // Parse DER certificate and convert to dict (outside lock)
-            let (_, cert) = x509_parser::parse_x509_certificate(&der_bytes)
-                .map_err(|e| vm.new_value_error(format!("Failed to parse certificate: {e}")))?;
-
-            cert::cert_to_dict(vm, &cert).map(Some)
+            cert::cert_der_to_dict_helper(vm, &der_bytes).map(Some)
         }
 
         #[pymethod]
@@ -4643,7 +4639,6 @@ mod _ssl {
     }
 
     impl PySSLCertificate {
-        // Parse the certificate lazily
         fn parse(&self) -> Result<x509_parser::certificate::X509Certificate<'_>, String> {
             match x509_parser::parse_x509_certificate(&self.der_bytes) {
                 Ok((_, cert)) => Ok(cert),
@@ -4682,8 +4677,7 @@ mod _ssl {
 
         #[pymethod]
         fn get_info(&self, vm: &VirtualMachine) -> PyResult {
-            let cert = self.parse().map_err(|e| vm.new_value_error(e))?;
-            cert::cert_to_dict(vm, &cert)
+            cert::cert_der_to_dict_helper(vm, &self.der_bytes)
         }
     }
 

diff --git a/crates/stdlib/src/ssl/cert.rs b/crates/stdlib/src/ssl/cert.rs
@@ -228,45 +228,6 @@ fn normalize_wildcard_hostname(hostname: &str) -> &str {
     hostname.strip_prefix("*.").unwrap_or(hostname)
 }
 
-/// Process Subject Alternative Name (SAN) general names into Python tuples
-///
-/// Converts X.509 GeneralName entries into Python tuple format.
-/// Returns a vector of PyObjectRef tuples in the format: (type, value)
-fn process_san_general_names(
-    vm: &VirtualMachine,
-    general_names: &[GeneralName<'_>],
-) -> Vec<PyObjectRef> {
-    general_names
-        .iter()
-        .filter_map(|name| match name {
-            GeneralName::DNSName(dns) => Some(vm.new_tuple(("DNS", *dns)).into()),
-            GeneralName::IPAddress(ip) => {
-                let ip_str = format_ip_address(ip);
-                Some(vm.new_tuple(("IP Address", ip_str)).into())
-            }
-            GeneralName::RFC822Name(email) => Some(vm.new_tuple(("email", *email)).into()),
-            GeneralName::URI(uri) => Some(vm.new_tuple(("URI", *uri)).into()),
-            GeneralName::DirectoryName(dn) => {
-                let dn_str = format!("{dn}");
-                Some(vm.new_tuple(("DirName", dn_str)).into())
-            }
-            GeneralName::RegisteredID(oid) => {
-                let oid_str = oid.to_string();
-                Some(vm.new_tuple(("Registered ID", oid_str)).into())
-            }
-            GeneralName::OtherName(oid, value) => {
-                let oid_str = oid.to_string();
-                let value_str = format!("{value:?}");
-                Some(
-                    vm.new_tuple(("othername", format!("{oid_str}:{value_str}")))
-                        .into(),
-                )
-            }
-            _ => None,
-        })
-        .collect()
-}
-
 // Certificate Validation and Parsing
 
 /// Check if a certificate is a CA certificate by examining the Basic Constraints extension
@@ -294,87 +255,10 @@ pub(super) fn is_ca_certificate(cert_der: &[u8]) -> bool {
     cert.version().0 == 0 && cert.subject() == cert.issuer()
 }
 
-/// Convert an X509Name to Python nested tuple format for SSL certificate dicts
-///
-/// Format: ((('CN', 'example.com'),), (('O', 'Example Org'),), ...)
-fn name_to_py(vm: &VirtualMachine, name: &x509_parser::x509::X509Name<'_>) -> PyObjectRef {
-    let list = name
-        .iter()
-        .flat_map(|rdn| {
-            // Each RDN can have multiple attributes
-            rdn.iter()
-                .map(|attr| {
-                    let oid_str = attr.attr_type().to_id_string();
-                    let value_str = attr.attr_value().as_str().unwrap_or("").to_string();
-                    let key = oid_to_attribute_name(&oid_str);
-
-                    vm.new_tuple((vm.new_tuple((vm.ctx.new_str(key), vm.ctx.new_str(value_str))),))
-                        .into()
-                })
-                .collect::<Vec<_>>()
-        })
-        .collect::<Vec<PyObjectRef>>();
-
-    vm.ctx.new_tuple(list).into()
-}
-
-/// Convert DER-encoded certificate to Python dict (for getpeercert with binary_form=False)
-///
-/// Returns a dict with fields: subject, issuer, version, serialNumber,
-/// notBefore, notAfter, subjectAltName (if present)
-pub(super) fn cert_to_dict(
-    vm: &VirtualMachine,
-    cert: &x509_parser::certificate::X509Certificate<'_>,
-) -> PyResult {
-    let dict = vm.ctx.new_dict();
-
-    // Subject and Issuer
-    dict.set_item("subject", name_to_py(vm, cert.subject()), vm)?;
-    dict.set_item("issuer", name_to_py(vm, cert.issuer()), vm)?;
-
-    // Version (X.509 v3 = version 2 in the cert, but Python uses 3)
-    dict.set_item(
-        "version",
-        vm.ctx.new_int(cert.version().0 as i32 + 1).into(),
-        vm,
-    )?;
-
-    // Serial number - hex format with even length
-    let serial = format_serial_number(&cert.serial);
-    dict.set_item("serialNumber", vm.ctx.new_str(serial).into(), vm)?;
-
-    // Validity dates - format with GMT using jiff
-    dict.set_item(
-        "notBefore",
-        vm.ctx
-            .new_str(format_asn1_time(&cert.validity().not_before))
-            .into(),
-        vm,
-    )?;
-    dict.set_item(
-        "notAfter",
-        vm.ctx
-            .new_str(format_asn1_time(&cert.validity().not_after))
-            .into(),
-        vm,
-    )?;
-
-    // Subject Alternative Names (if present)
-    if let Ok(Some(san_ext)) = cert.subject_alternative_name() {
-        let san_list = process_san_general_names(vm, &san_ext.value.general_names);
-
-        if !san_list.is_empty() {
-            dict.set_item("subjectAltName", vm.ctx.new_tuple(san_list).into(), vm)?;
-        }
-    }
-
-    Ok(dict.into())
-}
-
-/// Convert DER-encoded certificate to Python dict (for get_ca_certs)
+/// Convert DER-encoded certificate to Python dict.
 ///
-/// Similar to cert_to_dict but includes additional fields like crlDistributionPoints
-/// and uses CPython's specific ordering: issuer, notAfter, notBefore, serialNumber, subject, version
+/// Includes OCSP, caIssuers, and crlDistributionPoints when present, and uses
+/// the field order: issuer, notAfter, notBefore, serialNumber, subject, version
 pub(super) fn cert_der_to_dict_helper(
     vm: &VirtualMachine,
     cert_der: &[u8],

diff --git a/extra_tests/snippets/stdlib_ssl.py b/extra_tests/snippets/stdlib_ssl.py
@@ -0,0 +1,46 @@
+import pathlib
+import ssl
+
+CERT = (
+    pathlib.Path(__file__).resolve().parent.parent.parent
+    / "Lib"
+    / "test"
+    / "certdata"
+    / "keycert.pem"
+)
+
+TAIL = b"tail"
+
+server_context = ssl.SSLContext(ssl.PROTOCOL_TLS_SERVER)
+server_context.load_cert_chain(str(CERT))
+client_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
+client_context.check_hostname = False
+client_context.verify_mode = ssl.CERT_NONE
+server_context.maximum_version = client_context.maximum_version = ssl.TLSVersion.TLSv1_2
+
+client_in, client_out = ssl.MemoryBIO(), ssl.MemoryBIO()
+server_in, server_out = ssl.MemoryBIO(), ssl.MemoryBIO()
+client = client_context.wrap_bio(client_in, client_out)
+server = server_context.wrap_bio(server_in, server_out, server_side=True)
+
+for _ in range(5):
+    try:
+        client.do_handshake()
+    except ssl.SSLWantReadError:
+        pass
+    server_in.write(client_out.read())
+    try:
+        server.do_handshake()
+    except ssl.SSLWantReadError:
+        pass
+    client_in.write(server_out.read())
+client.do_handshake()
+server.do_handshake()
+
+try:
+    server.unwrap()
+except ssl.SSLWantReadError:
+    pass
+client_in.write(server_out.read() + TAIL)
+client.unwrap()
+assert client_in.read() == TAIL