// spell-checker: ignore ssleof aesccm aesgcm capath getblocking setblocking ENDTLS TLSEXT

//! Pure Rust SSL/TLS implementation using rustls

//!

//! This module provides SSL/TLS support without requiring C dependencies.

//! It implements the Python ssl module API using:

//! - rustls: TLS protocol implementation

//! - x509-parser/x509-cert: Certificate parsing

//! - ring: Cryptographic primitives

//! - rustls-platform-verifier: Platform-native certificate verification

//!

//! DO NOT add openssl dependency here.

//!

//! Warning: This library contains AI-generated code and comments. Do not trust any code or comment without verification. Please have a qualified expert review the code and remove this notice after review.

// OID (Object Identifier) management module

mod oid;

// Certificate operations module (parsing, validation, conversion)

mod cert;

// OpenSSL compatibility layer (abstracts rustls operations)

mod compat;

// SSL exception types (shared with openssl backend)

mod error;

pub(crate) use _ssl::module_def;

#[allow(non_snake_case)]

#[allow(non_upper_case_globals)]

#[pymodule(with(error::ssl_error))]

mod _ssl {

use crate::{

common::{

hash::PyHash,

lock::{PyMutex, PyRwLock},

},

socket::{PySocket, SelectKind, sock_select, timeout_error_msg},

vm::{

AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, TryFromObject,

VirtualMachine,

builtins::{

PyBaseExceptionRef, PyBytesRef, PyListRef, PyStrRef, PyType, PyTypeRef,

PyUtf8StrRef,

},

convert::IntoPyException,

function::{

ArgBytesLike, ArgMemoryBuffer, Either, FuncArgs, OptionalArg, PyComparisonValue,

},

stdlib::_warnings,

types::{Comparable, Constructor, Hashable, PyComparisonOp, Representable},

},

};

// Import error types used in this module (others are exposed via pymodule(with(...)))

use super::error::{

PySSLError, create_ssl_eof_error, create_ssl_want_read_error, create_ssl_want_write_error,

create_ssl_zero_return_error,

};

use alloc::sync::Arc;

use core::{

hash::{Hash, Hasher},

sync::atomic::{AtomicUsize, Ordering},

time::Duration,

};

use rustls::crypto::aws_lc_rs::ALL_CIPHER_SUITES;

use std::{

collections::{HashMap, hash_map::DefaultHasher},

io::BufRead,

time::SystemTime,

};

// Rustls imports

use parking_lot::{Mutex as ParkingMutex, RwLock as ParkingRwLock};

use pem_rfc7468::{LineEnding, encode_string};

use rustls::{

ClientConfig, ClientConnection, RootCertStore, ServerConfig, ServerConnection,

client::{ClientSessionMemoryCache, ClientSessionStore},

crypto::SupportedKxGroup,

pki_types::{CertificateDer, CertificateRevocationListDer, PrivateKeyDer, ServerName},

server::{ClientHello, ResolvesServerCert},

sign::CertifiedKey,

version::{TLS12, TLS13},

};

use sha2::{Digest, Sha256};

// Import certificate operations module

use super::cert;

// Import OID module

use super::oid;

// Import compat module (OpenSSL compatibility layer)

use super::compat::{

ClientConfigOptions, MultiCertResolver, ProtocolSettings, ServerConfigOptions, SslError,

TlsConnection, create_client_config, create_server_config, curve_name_to_kx_group,

extract_cipher_info, get_cipher_encryption_desc, is_blocking_io_error,

normalize_cipher_name, ssl_do_handshake,

};

// Type aliases for better readability

// Additional type alias for certificate/key pairs (SessionCache and SniCertName defined below)

/// Certificate and private key pair used in SSL contexts

type CertKeyPair = (Arc<CertifiedKey>, PrivateKeyDer<'static>);

// Constants matching Python ssl module

// SSL/TLS Protocol versions

#[pyattr]

const PROTOCOL_TLS: i32 = 2; // Auto-negotiate best version

#[pyattr]

const PROTOCOL_SSLv23: i32 = PROTOCOL_TLS; // Alias for PROTOCOL_TLS

#[pyattr]

const PROTOCOL_TLS_CLIENT: i32 = 16;

#[pyattr]

const PROTOCOL_TLS_SERVER: i32 = 17;

// Note: rustls doesn't support TLS 1.0/1.1 for security reasons

// These are defined for API compatibility but will raise errors if used

#[pyattr]

const PROTOCOL_TLSv1: i32 = 3;

#[pyattr]

const PROTOCOL_TLSv1_1: i32 = 4;

#[pyattr]

const PROTOCOL_TLSv1_2: i32 = 5;

#[pyattr]

const PROTOCOL_TLSv1_3: i32 = 6;

// Protocol version constants for TLSVersion enum

#[pyattr]

const PROTO_SSLv3: i32 = 0x0300;

#[pyattr]

const PROTO_TLSv1: i32 = 0x0301;

#[pyattr]

const PROTO_TLSv1_1: i32 = 0x0302;

#[pyattr]

const PROTO_TLSv1_2: i32 = 0x0303;

#[pyattr]

const PROTO_TLSv1_3: i32 = 0x0304;

// Minimum and maximum supported protocol versions for rustls

// Use special values -2 and -1 to avoid enum name conflicts

#[pyattr]

const PROTO_MINIMUM_SUPPORTED: i32 = -2; // special value

#[pyattr]

const PROTO_MAXIMUM_SUPPORTED: i32 = -1; // special value

// Internal constants for rustls actual supported versions

// rustls only supports TLS 1.2 and TLS 1.3

const MINIMUM_VERSION: i32 = PROTO_TLSv1_2; // 0x0303

const MAXIMUM_VERSION: i32 = PROTO_TLSv1_3; // 0x0304

// Buffer sizes and limits (OpenSSL/CPython compatibility)

const PEM_BUFSIZE: usize = 1024;

// OpenSSL: ssl/ssl_local.h

const SSL3_RT_MAX_PLAIN_LENGTH: usize = 16384;

// SSL session cache size (common practice, similar to OpenSSL defaults)

const SSL_SESSION_CACHE_SIZE: usize = 256;

// Certificate verification modes

#[pyattr]

const CERT_NONE: i32 = 0;

#[pyattr]

const CERT_OPTIONAL: i32 = 1;

#[pyattr]

const CERT_REQUIRED: i32 = 2;

// Certificate requirements

#[pyattr]

const VERIFY_DEFAULT: i32 = 0;

#[pyattr]

const VERIFY_CRL_CHECK_LEAF: i32 = 4;

#[pyattr]

const VERIFY_CRL_CHECK_CHAIN: i32 = 12;

#[pyattr]

const VERIFY_X509_STRICT: i32 = 32;

#[pyattr]

const VERIFY_ALLOW_PROXY_CERTS: i32 = 64;

#[pyattr]

const VERIFY_X509_TRUSTED_FIRST: i32 = 32768;

#[pyattr]

const VERIFY_X509_PARTIAL_CHAIN: i32 = 0x80000;

// Options (OpenSSL-compatible flags, mostly no-op in rustls)

#[pyattr]

const OP_NO_SSLv2: i32 = 0x00000000; // Not supported anyway

#[pyattr]

const OP_NO_SSLv3: i32 = 0x02000000;

#[pyattr]

const OP_NO_TLSv1: i32 = 0x04000000;

#[pyattr]

const OP_NO_TLSv1_1: i32 = 0x10000000;

#[pyattr]

const OP_NO_TLSv1_2: i32 = 0x08000000;

#[pyattr]

const OP_NO_TLSv1_3: i32 = 0x20000000;

#[pyattr]

const OP_NO_COMPRESSION: i32 = 0x00020000;

#[pyattr]

const OP_CIPHER_SERVER_PREFERENCE: i32 = 0x00400000;

#[pyattr]

const OP_SINGLE_DH_USE: i32 = 0x00000000; // No-op in rustls

#[pyattr]

const OP_SINGLE_ECDH_USE: i32 = 0x00000000; // No-op in rustls

#[pyattr]

const OP_NO_TICKET: i32 = 0x00004000;

#[pyattr]

const OP_LEGACY_SERVER_CONNECT: i32 = 0x00000004;

#[pyattr]

const OP_NO_RENEGOTIATION: i32 = 0x40000000;

#[pyattr]

const OP_IGNORE_UNEXPECTED_EOF: i32 = 0x00000080;

#[pyattr]

const OP_ENABLE_MIDDLEBOX_COMPAT: i32 = 0x00100000;

#[pyattr]

const OP_ALL: i32 = 0x00000BFB; // Combined "safe" options (reduced for i32, excluding OP_LEGACY_SERVER_CONNECT for OpenSSL 3.0.0+ compatibility)

// Alert types (matching _TLSAlertType enum)

#[pyattr]

const ALERT_DESCRIPTION_CLOSE_NOTIFY: i32 = 0;

#[pyattr]

const ALERT_DESCRIPTION_UNEXPECTED_MESSAGE: i32 = 10;

#[pyattr]

const ALERT_DESCRIPTION_BAD_RECORD_MAC: i32 = 20;

#[pyattr]

const ALERT_DESCRIPTION_DECRYPTION_FAILED: i32 = 21;

#[pyattr]

const ALERT_DESCRIPTION_RECORD_OVERFLOW: i32 = 22;

#[pyattr]

const ALERT_DESCRIPTION_DECOMPRESSION_FAILURE: i32 = 30;

#[pyattr]

const ALERT_DESCRIPTION_HANDSHAKE_FAILURE: i32 = 40;

#[pyattr]

const ALERT_DESCRIPTION_NO_CERTIFICATE: i32 = 41;

#[pyattr]

const ALERT_DESCRIPTION_BAD_CERTIFICATE: i32 = 42;

#[pyattr]

const ALERT_DESCRIPTION_UNSUPPORTED_CERTIFICATE: i32 = 43;

#[pyattr]

const ALERT_DESCRIPTION_CERTIFICATE_REVOKED: i32 = 44;

#[pyattr]

const ALERT_DESCRIPTION_CERTIFICATE_EXPIRED: i32 = 45;

#[pyattr]

const ALERT_DESCRIPTION_CERTIFICATE_UNKNOWN: i32 = 46;

#[pyattr]

const ALERT_DESCRIPTION_ILLEGAL_PARAMETER: i32 = 47;

#[pyattr]

const ALERT_DESCRIPTION_UNKNOWN_CA: i32 = 48;

#[pyattr]

const ALERT_DESCRIPTION_ACCESS_DENIED: i32 = 49;

#[pyattr]

const ALERT_DESCRIPTION_DECODE_ERROR: i32 = 50;

#[pyattr]

const ALERT_DESCRIPTION_DECRYPT_ERROR: i32 = 51;

#[pyattr]

const ALERT_DESCRIPTION_EXPORT_RESTRICTION: i32 = 60;

#[pyattr]

const ALERT_DESCRIPTION_PROTOCOL_VERSION: i32 = 70;

#[pyattr]

const ALERT_DESCRIPTION_INSUFFICIENT_SECURITY: i32 = 71;

#[pyattr]

const ALERT_DESCRIPTION_INTERNAL_ERROR: i32 = 80;

#[pyattr]

const ALERT_DESCRIPTION_INAPPROPRIATE_FALLBACK: i32 = 86;

#[pyattr]

const ALERT_DESCRIPTION_USER_CANCELLED: i32 = 90;

#[pyattr]

const ALERT_DESCRIPTION_NO_RENEGOTIATION: i32 = 100;

#[pyattr]

const ALERT_DESCRIPTION_MISSING_EXTENSION: i32 = 109;

#[pyattr]

const ALERT_DESCRIPTION_UNSUPPORTED_EXTENSION: i32 = 110;

#[pyattr]

const ALERT_DESCRIPTION_CERTIFICATE_UNOBTAINABLE: i32 = 111;

#[pyattr]

const ALERT_DESCRIPTION_UNRECOGNIZED_NAME: i32 = 112;

#[pyattr]

const ALERT_DESCRIPTION_BAD_CERTIFICATE_STATUS_RESPONSE: i32 = 113;

#[pyattr]

const ALERT_DESCRIPTION_BAD_CERTIFICATE_HASH_VALUE: i32 = 114;

#[pyattr]

const ALERT_DESCRIPTION_UNKNOWN_PSK_IDENTITY: i32 = 115;

#[pyattr]

const ALERT_DESCRIPTION_CERTIFICATE_REQUIRED: i32 = 116;

#[pyattr]

const ALERT_DESCRIPTION_NO_APPLICATION_PROTOCOL: i32 = 120;

// Version info - reporting as OpenSSL 3.3.0 for compatibility

#[pyattr]

const OPENSSL_VERSION_NUMBER: i32 = 0x30300000; // OpenSSL 3.3.0 (808452096)

#[pyattr]

const OPENSSL_VERSION: &str = "OpenSSL 3.3.0 (rustls/0.23)";

#[pyattr]

const OPENSSL_VERSION_INFO: (i32, i32, i32, i32, i32) = (3, 3, 0, 0, 15); // 3.3.0 release

#[pyattr]

const _OPENSSL_API_VERSION: (i32, i32, i32, i32, i32) = (3, 3, 0, 0, 15); // 3.3.0 release

// Default cipher list for rustls - using modern secure ciphers

#[pyattr]

const _DEFAULT_CIPHERS: &str =

"TLS_AES_256_GCM_SHA384:TLS_AES_128_GCM_SHA256:TLS_CHACHA20_POLY1305_SHA256";

// Has features

#[pyattr]

const HAS_SNI: bool = true;

#[pyattr]

const HAS_TLS_UNIQUE: bool = false; // Not supported

#[pyattr]

const HAS_ECDH: bool = true;

#[pyattr]

const HAS_NPN: bool = false; // Deprecated, use ALPN

#[pyattr]

const HAS_ALPN: bool = true;

#[pyattr]

const HAS_PSK: bool = false; // PSK not supported in rustls

#[pyattr]

const HAS_SSLv2: bool = false;

#[pyattr]

const HAS_SSLv3: bool = false;

#[pyattr]

const HAS_TLSv1: bool = false; // Not supported for security

#[pyattr]

const HAS_TLSv1_1: bool = false; // Not supported for security

#[pyattr]

const HAS_TLSv1_2: bool = true; // rustls supports TLS 1.2

#[pyattr]

const HAS_TLSv1_3: bool = true;

#[pyattr]

const HAS_PHA: bool = false; // Post-Handshake Auth not supported in rustls

// Encoding constants (matching OpenSSL)

#[pyattr]

const ENCODING_PEM: i32 = 1;

#[pyattr]

const ENCODING_DER: i32 = 2;

#[pyattr]

const ENCODING_PEM_AUX: i32 = 0x101; // PEM + 0x100

/// Validate server hostname for TLS SNI

///

/// Checks that the hostname:

/// - Is not empty

/// - Does not start with a dot

/// - Is not an IP address (SNI requires DNS names)

/// - Does not contain null bytes

/// - Does not exceed 253 characters (DNS limit)

///

/// Returns Ok(()) if validation passes, or an appropriate error.

fn validate_hostname(hostname: &str, vm: &VirtualMachine) -> PyResult<()> {

if hostname.is_empty() {

return Err(vm.new_value_error("server_hostname cannot be an empty string"));

}

if hostname.starts_with('.') {

return Err(vm.new_value_error("server_hostname cannot start with a dot"));

}

// IP addresses are allowed as server_hostname

// SNI will not be sent for IP addresses

if hostname.contains('\0') {

return Err(vm.new_type_error("embedded null character"));

}

if hostname.len() > 253 {

return Err(vm.new_value_error("server_hostname is too long (maximum 253 characters)"));

}

Ok(())

}

// SNI certificate resolver that uses shared mutable state

// The Python SNI callback updates this state, and resolve() reads from it

#[derive(Debug)]

struct SniCertResolver {

// SNI state: (certificate, server_name)

sni_state: Arc<ParkingMutex<SniCertName>>,

}

impl ResolvesServerCert for SniCertResolver {

fn resolve(&self, client_hello: ClientHello<'_>) -> Option<Arc<CertifiedKey>> {

let mut state = self.sni_state.lock();

// Extract and store SNI from client hello for later use

if let Some(sni) = client_hello.server_name() {

state.1 = Some(sni.to_string());

} else {

state.1 = None;

}

// Return the current certificate (may have been updated by Python callback)

Some(state.0.clone())

}

}

// Session data structure for tracking TLS sessions

#[derive(Debug, Clone)]

struct SessionData {

#[allow(dead_code)]

server_name: String,

session_id: Vec<u8>,

creation_time: SystemTime,

lifetime: u64,

}

// Type alias to simplify complex session cache type

type SessionCache = Arc<ParkingRwLock<HashMap<Vec<u8>, Arc<ParkingMutex<SessionData>>>>>;

// Type alias for SNI state

type SniCertName = (Arc<CertifiedKey>, Option<String>);

// SESSION EMULATION IMPLEMENTATION

//

// IMPORTANT: This is an EMULATION of CPython's SSL session management.

// Rustls 0.23 does NOT expose session data (ticket bytes, session IDs, etc.)

// through public APIs. All session value fields are private.

//

// LIMITATIONS:

// - Session IDs are generated from metadata (server name + timestamp hash)

// NOT actual TLS session IDs

// - Ticket data is not stored (Rustls keeps it internally)

// - Session resumption works (via Rustls's automatic mechanism)

// but we can't access the actual session state

//

// This implementation provides:

// ✓ session.id - synthetic ID based on metadata

// ✓ session.time - creation timestamp

// ✓ session.timeout - default lifetime value

// ✓ session.has_ticket - always True when session exists

// ✓ session_reused - tracked via handshake_kind()

// ✗ Actual TLS session ID/ticket data - NOT ACCESSIBLE

// Generate a synthetic session ID from server name and timestamp

// NOTE: This is NOT the actual TLS session ID, just a unique identifier

fn generate_session_id_from_metadata(server_name: &str, time: &SystemTime) -> Vec<u8> {

let mut hasher = Sha256::new();

hasher.update(server_name.as_bytes());

hasher.update(

time.duration_since(SystemTime::UNIX_EPOCH)

.unwrap()

.as_secs()

.to_le_bytes(),

);

hasher.finalize()[..16].to_vec()

}

// Custom ClientSessionStore that tracks session metadata for Python access

// NOTE: This wraps ClientSessionMemoryCache and records metadata when sessions are stored

#[derive(Debug)]

struct PythonClientSessionStore {

inner: Arc<ClientSessionMemoryCache>,

session_cache: SessionCache,

}

impl ClientSessionStore for PythonClientSessionStore {

fn set_kx_hint(&self, server_name: ServerName<'static>, group: rustls::NamedGroup) {

self.inner.set_kx_hint(server_name, group);

}

fn kx_hint(&self, server_name: &ServerName<'_>) -> Option<rustls::NamedGroup> {

self.inner.kx_hint(server_name)

}

fn set_tls12_session(

&self,

server_name: ServerName<'static>,

value: rustls::client::Tls12ClientSessionValue,

) {

// Store in inner cache for actual resumption (Rustls handles this)

self.inner.set_tls12_session(server_name.clone(), value);

// Record metadata in Python-accessible cache

// NOTE: We can't access value.session_id or value.ticket (private fields)

// So we generate a synthetic ID from metadata

let creation_time = SystemTime::now();

let server_name_str = server_name.to_str();

let session_data = SessionData {

server_name: server_name_str.as_ref().to_string(),

session_id: generate_session_id_from_metadata(

server_name_str.as_ref(),

&creation_time,

),

creation_time,

lifetime: 7200, // TLS 1.2 default session lifetime

};

let key = server_name_str.as_bytes().to_vec();

self.session_cache

.write()

.insert(key, Arc::new(ParkingMutex::new(session_data)));

}

fn tls12_session(

&self,

server_name: &ServerName<'_>,

) -> Option<rustls::client::Tls12ClientSessionValue> {

self.inner.tls12_session(server_name)

}

fn remove_tls12_session(&self, server_name: &ServerName<'static>) {

self.inner.remove_tls12_session(server_name);

// Also remove from Python cache

let key = server_name.to_str().as_bytes().to_vec();

self.session_cache.write().remove(&key);

}

fn insert_tls13_ticket(

&self,

server_name: ServerName<'static>,

value: rustls::client::Tls13ClientSessionValue,

) {

// Store in inner cache for actual resumption (Rustls handles this)

self.inner.insert_tls13_ticket(server_name.clone(), value);

// Record metadata in Python-accessible cache

// NOTE: We can't access value.ticket or value.lifetime_secs (private fields)

// So we use default values

let creation_time = SystemTime::now();

let server_name_str = server_name.to_str();

let session_data = SessionData {

server_name: server_name_str.to_string(),

session_id: generate_session_id_from_metadata(

server_name_str.as_ref(),

&creation_time,

),

creation_time,

lifetime: 7200, // Default TLS 1.3 ticket lifetime (Rustls uses this)

};

let key = server_name_str.as_bytes().to_vec();

self.session_cache

.write()

.insert(key, Arc::new(ParkingMutex::new(session_data)));

}

fn take_tls13_ticket(

&self,

server_name: &ServerName<'static>,

) -> Option<rustls::client::Tls13ClientSessionValue> {

self.inner.take_tls13_ticket(server_name)

}

}

/// Parse length-prefixed ALPN protocol list

///

/// Format: [len1, proto1..., len2, proto2..., ...]

///

/// This is the wire format used by Python's ssl.py when calling _set_alpn_protocols().

/// Each protocol is prefixed with a single byte indicating its length.

///

/// # Arguments

/// * `bytes` - The length-prefixed protocol data

/// * `vm` - VirtualMachine for error creation

///

/// # Returns

/// * `Ok(Vec<Vec<u8>>)` - List of protocol names as byte vectors

/// * `Err(PyBaseExceptionRef)` - ValueError with detailed error message

fn parse_length_prefixed_alpn(bytes: &[u8], vm: &VirtualMachine) -> PyResult<Vec<Vec<u8>>> {

let mut alpn_list = Vec::new();

let mut offset = 0;

while offset < bytes.len() {

// Check if we can read the length byte

if offset + 1 > bytes.len() {

return Err(vm.new_value_error(format!(

"Invalid ALPN protocol data: unexpected end at offset {offset}",

)));

}

let proto_len = bytes[offset] as usize;

offset += 1;

// Validate protocol length

if proto_len == 0 {

return Err(vm.new_value_error(format!(

"Invalid ALPN protocol data: protocol length cannot be 0 at offset {}",

offset - 1

)));

}

// Check if we have enough bytes for the protocol data

if offset + proto_len > bytes.len() {

return Err(vm.new_value_error(format!(

"Invalid ALPN protocol data: expected {} bytes at offset {}, but only {} bytes remain",

proto_len, offset, bytes.len() - offset

)));

}

// Extract protocol bytes

let proto = bytes[offset..offset + proto_len].to_vec();

alpn_list.push(proto);

offset += proto_len;

}

Ok(alpn_list)

}

/// Parse OpenSSL cipher string to rustls SupportedCipherSuite list

///

/// Supports patterns like:

/// - "AES128" → filters for AES_128

/// - "AES256" → filters for AES_256

/// - "AES128:AES256" → both

/// - "ECDHE+AESGCM" → ECDHE AND AESGCM (both conditions must match)

/// - "ALL" or "DEFAULT" → all available

/// - "!MD5" → exclusion (ignored, rustls doesn't support weak ciphers anyway)

fn parse_cipher_string(cipher_str: &str) -> Result<Vec<rustls::SupportedCipherSuite>, String> {

if cipher_str.is_empty() {

return Err("No cipher can be selected".to_string());

}

let all_suites = ALL_CIPHER_SUITES;

let mut selected = Vec::new();

for part in cipher_str.split(':') {

let part = part.trim();

// Skip exclusions (rustls doesn't support these)

if part.starts_with('!') {

continue;

}

// Skip priority markers starting with +

if part.starts_with('+') {

continue;

}

// Match pattern

match part {

"ALL" | "DEFAULT" | "HIGH" => {

// Add all available cipher suites

selected.extend_from_slice(all_suites);

}

_ => {

// Check if this is a compound pattern with + (AND condition)

// e.g., "ECDHE+AESGCM" means ECDHE AND AESGCM

let patterns: Vec<&str> = part.split('+').collect();

let mut found_any = false;

for suite in all_suites {

let name = format!("{:?}", suite.suite());

// Check if all patterns match (AND condition)

let matches = patterns.iter().all(|&pattern| {

// Handle common OpenSSL pattern variations

if pattern.contains("AES128") {

name.contains("AES_128")

} else if pattern.contains("AES256") {

name.contains("AES_256")

} else if pattern == "AESGCM" {

// AESGCM: AES with GCM mode

name.contains("AES") && name.contains("GCM")

} else if pattern == "AESCCM" {

// AESCCM: AES with CCM mode

name.contains("AES") && name.contains("CCM")

} else if pattern == "CHACHA20" {

name.contains("CHACHA20")

} else if pattern == "ECDHE" {

name.contains("ECDHE")

} else if pattern == "DHE" {

// DHE but not ECDHE

name.contains("DHE") && !name.contains("ECDHE")

} else if pattern == "ECDH" {

// ECDH but not ECDHE

name.contains("ECDH") && !name.contains("ECDHE")

} else if pattern == "DH" {

// DH but not DHE or ECDH

name.contains("DH")

&& !name.contains("DHE")

&& !name.contains("ECDH")

} else if pattern == "RSA" {

name.contains("RSA")

} else if pattern == "AES" {

name.contains("AES")

} else if pattern == "ECDSA" {

name.contains("ECDSA")

} else {

// Direct substring match for other patterns

name.contains(pattern)

}

});

if matches {

selected.push(*suite);

found_any = true;

}

}

if !found_any {

// No matching cipher suite found - warn but continue

}

}

}

}

// Remove duplicates

selected.dedup_by_key(|s| s.suite());

if selected.is_empty() {

Err("No cipher can be selected".to_string())

} else {

Ok(selected)

}

}

// SSLContext - manages TLS configuration

#[pyattr]

#[pyclass(name = "_SSLContext", module = "ssl", traverse)]

#[derive(Debug, PyPayload)]

struct PySSLContext {

#[pytraverse(skip)]

protocol: i32,

#[pytraverse(skip)]

check_hostname: PyRwLock<bool>,

#[pytraverse(skip)]

verify_mode: PyRwLock<i32>,

#[pytraverse(skip)]

verify_flags: PyRwLock<i32>,

// Rustls configuration (built lazily)

#[allow(dead_code)]

#[pytraverse(skip)]

client_config: PyRwLock<Option<Arc<ClientConfig>>>,

#[allow(dead_code)]

#[pytraverse(skip)]

server_config: PyRwLock<Option<Arc<ServerConfig>>>,

// Certificate store

#[pytraverse(skip)]

root_certs: PyRwLock<RootCertStore>,

// Store full CA certificates for get_ca_certs()

// RootCertStore only keeps TrustAnchors, not full certificates

#[pytraverse(skip)]

ca_certs_der: PyRwLock<Vec<Vec<u8>>>,

// Store CA certificates from capath for lazy loading simulation

// (CPython only returns these in get_ca_certs() after they're used in handshake)

#[pytraverse(skip)]

capath_certs_der: PyRwLock<Vec<Vec<u8>>>,

// Certificate Revocation Lists for CRL checking

#[pytraverse(skip)]

crls: PyRwLock<Vec<CertificateRevocationListDer<'static>>>,

// Server certificate/key pairs (supports multiple for RSA+ECC dual mode)

// OpenSSL allows multiple cert/key pairs to be loaded, and selects the appropriate

// one based on client capabilities during handshake

// Stored as (CertifiedKey, PrivateKeyDer) to support both server and client usage

#[pytraverse(skip)]

cert_keys: PyRwLock<Vec<CertKeyPair>>,

// Options

#[allow(dead_code)]

#[pytraverse(skip)]

options: PyRwLock<i32>,

// ALPN protocols

#[allow(dead_code)]

#[pytraverse(skip)]

alpn_protocols: PyRwLock<Vec<Vec<u8>>>,

// ALPN strict matching flag

// When false (default), mimics OpenSSL behavior: no ALPN negotiation failure

// When true, requires ALPN match (Rustls default behavior)

#[allow(dead_code)]

#[pytraverse(skip)]

require_alpn_match: PyRwLock<bool>,

// TLS 1.3 features

#[pytraverse(skip)]

post_handshake_auth: PyRwLock<bool>,

#[pytraverse(skip)]

num_tickets: PyRwLock<i32>,

// Protocol version limits

#[pytraverse(skip)]

minimum_version: PyRwLock<i32>,

#[pytraverse(skip)]

maximum_version: PyRwLock<i32>,

// SNI callback for server-side (contains PyObjectRef - needs GC tracking)

sni_callback: PyRwLock<Option<PyObjectRef>>,

// Message callback for debugging (contains PyObjectRef - needs GC tracking)

msg_callback: PyRwLock<Option<PyObjectRef>>,

// ECDH curve name for key exchange

#[pytraverse(skip)]

ecdh_curve: PyRwLock<Option<String>>,

// Certificate statistics for cert_store_stats()

#[pytraverse(skip)]

ca_cert_count: PyRwLock<usize>, // Number of CA certificates

#[pytraverse(skip)]

x509_cert_count: PyRwLock<usize>, // Total number of certificates

// Session management

#[pytraverse(skip)]

client_session_cache: SessionCache,

// Rustls session store for actual TLS session resumption

#[pytraverse(skip)]

rustls_session_store: Arc<PythonClientSessionStore>,

// Rustls server session store for server-side session resumption

#[pytraverse(skip)]

rustls_server_session_store: Arc<rustls::server::ServerSessionMemoryCache>,

// Shared ticketer for TLS 1.2 session tickets

#[pytraverse(skip)]

server_ticketer: Arc<dyn rustls::server::ProducesTickets>,

// Server-side session statistics

#[pytraverse(skip)]

accept_count: AtomicUsize, // Total number of accepts

#[pytraverse(skip)]

session_hits: AtomicUsize, // Number of session reuses

// Cipher suite selection

/// Selected cipher suites (None = use all rustls defaults)

#[pytraverse(skip)]

selected_ciphers: PyRwLock<Option<Vec<rustls::SupportedCipherSuite>>>,

}

#[derive(FromArgs)]

struct WrapSocketArgs {

sock: PyObjectRef,

server_side: bool,

#[pyarg(positional, optional)]

server_hostname: OptionalArg<Option<PyUtf8StrRef>>,

#[pyarg(named, optional)]

owner: OptionalArg<PyObjectRef>,

#[pyarg(named, optional)]

session: OptionalArg<PyObjectRef>,

}

#[derive(FromArgs)]

struct WrapBioArgs {

incoming: PyRef<PyMemoryBIO>,

outgoing: PyRef<PyMemoryBIO>,

#[pyarg(named, optional)]

server_side: OptionalArg<bool>,

#[pyarg(named, optional)]

server_hostname: OptionalArg<Option<PyUtf8StrRef>>,

#[pyarg(named, optional)]

owner: OptionalArg<PyObjectRef>,

#[pyarg(named, optional)]

session: OptionalArg<PyObjectRef>,

}

#[derive(FromArgs)]

struct LoadVerifyLocationsArgs {

#[pyarg(any, optional, error_msg = "path should be a str or bytes")]

cafile: OptionalArg<Option<Either<PyStrRef, ArgBytesLike>>>,

#[pyarg(any, optional, error_msg = "path should be a str or bytes")]

capath: OptionalArg<Option<Either<PyStrRef, ArgBytesLike>>>,

#[pyarg(any, optional, error_msg = "cadata should be a str or bytes")]

cadata: OptionalArg<Option<Either<PyStrRef, ArgBytesLike>>>,

}

#[derive(FromArgs)]

struct LoadCertChainArgs {

#[pyarg(any, error_msg = "path should be a str or bytes")]

certfile: Either<PyStrRef, ArgBytesLike>,

#[pyarg(any, optional, error_msg = "path should be a str or bytes")]

keyfile: OptionalArg<Option<Either<PyStrRef, ArgBytesLike>>>,

#[pyarg(any, optional)]

password: OptionalArg<PyObjectRef>,

}

#[derive(FromArgs)]

struct GetCertArgs {

#[pyarg(any, optional)]

binary_form: OptionalArg<bool>,

}

#[pyclass(with(Constructor, Representable), flags(BASETYPE))]

impl PySSLContext {

// Helper method to convert DER certificate bytes to Python dict

fn cert_der_to_dict(&self, vm: &VirtualMachine, cert_der: &[u8]) -> PyResult<PyObjectRef> {

cert::cert_der_to_dict_helper(vm, cert_der)

}

#[pygetset]

fn check_hostname(&self) -> bool {

*self.check_hostname.read()

}

#[pygetset(setter)]

fn set_check_hostname(&self, value: bool) {

*self.check_hostname.write() = value;

// When check_hostname is enabled, ensure verify_mode is at least CERT_REQUIRED

if value {

let current_verify_mode = *self.verify_mode.read();

if current_verify_mode == CERT_NONE {

*self.verify_mode.write() = CERT_REQUIRED;

}

}

}

#[pygetset]

fn verify_mode(&self) -> i32 {

*self.verify_mode.read()

}

#[pygetset(setter)]

fn set_verify_mode(&self, mode: i32, vm: &VirtualMachine) -> PyResult<()> {

if !(CERT_NONE..=CERT_REQUIRED).contains(&mode) {

return Err(vm.new_value_error("invalid verify mode"));

}

// Cannot set CERT_NONE when check_hostname is enabled

if mode == CERT_NONE && *self.check_hostname.read() {

return Err(vm.new_value_error(

"Cannot set verify_mode to CERT_NONE when check_hostname is enabled",

));

}

*self.verify_mode.write() = mode;

Ok(())

}

#[pygetset]

fn protocol(&self) -> i32 {

self.protocol

}

#[pygetset]

fn verify_flags(&self) -> i32 {

*self.verify_flags.read()

}

#[pygetset(setter)]

fn set_verify_flags(&self, value: i32) {

*self.verify_flags.write() = value;

}

#[pygetset]

fn post_handshake_auth(&self) -> bool {

*self.post_handshake_auth.read()

}

#[pygetset(setter)]

fn set_post_handshake_auth(&self, value: bool) {

*self.post_handshake_auth.write() = value;

}

#[pygetset]

fn num_tickets(&self) -> i32 {

*self.num_tickets.read()

}

#[pygetset(setter)]

fn set_num_tickets(&self, value: i32, vm: &VirtualMachine) -> PyResult<()> {

if value < 0 {

return Err(vm.new_value_error("num_tickets must be a non-negative integer"));

}

if self.protocol != PROTOCOL_TLS_SERVER {

return Err(

vm.new_value_error("num_tickets can only be set on server-side contexts")

);

}

*self.num_tickets.write() = value;

Ok(())

}

#[pygetset]

fn options(&self) -> i32 {

*self.options.read()

}

#[pygetset(setter)]

fn set_options(&self, value: i32, vm: &VirtualMachine) -> PyResult<()> {

// Validate that the value is non-negative

if value < 0 {

return Err(vm.new_value_error("options must be non-negative"));

}

// Deprecated SSL/TLS protocol version options

let opt_no = OP_NO_SSLv2

| OP_NO_SSLv3

| OP_NO_TLSv1

| OP_NO_TLSv1_1

| OP_NO_TLSv1_2

| OP_NO_TLSv1_3;

// Get current options and calculate newly set bits

let old_opts = *self.options.read();

let set = !old_opts & value; // Bits being newly set

// Warn if any deprecated options are being newly set

if (set & opt_no) != 0 {

_warnings::warn(

vm.ctx.exceptions.deprecation_warning,

"ssl.OP_NO_SSL*/ssl.OP_NO_TLS* options are deprecated".to_owned(),

2, // stack_level = 2

vm,

)?;

}

*self.options.write() = value;

Ok(())

}

#[pygetset]

fn minimum_version(&self) -> i32 {

let v = *self.minimum_version.read();

// return MINIMUM_SUPPORTED if value is 0

if v == 0 { PROTO_MINIMUM_SUPPORTED } else { v }

}

#[pygetset(setter)]

fn set_minimum_version(&self, value: i32, vm: &VirtualMachine) -> PyResult<()> {

// Validate that the value is a valid TLS version constant

// Valid values: 0 (default), -2 (MINIMUM_SUPPORTED), -1 (MAXIMUM_SUPPORTED),

// or 0x0300-0x0304 (SSLv3-TLSv1.3)

if value != 0

&& value != -2

&& value != -1