use rustpython_vm::{PyRef, VirtualMachine, builtins::PyOSError};

#[derive(Copy, Clone)]

pub(crate) enum SockWaitKind {

Read,

Write,

#[allow(dead_code, reason = "kept in lockstep with the native wait kinds")]

Connect,

}

pub(crate) fn timeout_error_msg(vm: &VirtualMachine, msg: String) -> PyRef<PyOSError> {

vm.new_os_subtype_error(vm.ctx.exceptions.timeout_error.to_owned(), None, msg)

}

use crate::{

common::{

hash::PyHash,

lock::{PyMutex, PyRwLock},

},

vm::{

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

VirtualMachine,

builtins::{

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

PyUtf8StrRef, PyWeak,

},

convert::IntoPyException,

function::{

ArgBytesLike, ArgMemoryBuffer, Either, FsPath, FuncArgs, OptionalArg,

PyComparisonValue, PySetterValue,

},

stdlib::_warnings,

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

},

};

#[cfg(target_arch = "wasm32")]

use super::sock_wait_stub::{SockWaitKind, timeout_error_msg};

#[cfg(not(target_arch = "wasm32"))]

use crate::socket::{PySocket, SockWaitKind, sock_wait, timeout_error_msg};

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

use super::error::{

PySSLError, PySSLWantReadError, PySSLWantWriteError, 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},

hint::cold_path,

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

time::Duration,

};

use rustpython_vm::exceptions;

use std::{

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

time::SystemTime,

};

// Rustls imports

use parking_lot::RwLock as ParkingRwLock;

use pem_rfc7468::{LineEnding, encode_string};

use rustls::{

ClientConnection, Connection, HandshakeKind, RootCertStore,

crypto::SupportedKxGroup,

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

server::{Accepted, ResolvesServerCert},

sign::CertifiedKey,

};

use rustpython_host_env::ssl::{

CapturingClientSessionStore, ClientSessionKind, RecordCursor, SessionCache, SessionData,

TlsConnection,

};

/// Caps unsent TLS so BIO writes stay inside asyncio's 64 KiB

/// socket-transport high-water mark when SO_SNDBUF is small.

const TLS_IO_BUFFER_LIMIT: Option<usize> = Some(4 * 1024);

// Import certificate operations module

use super::cert;

use rustpython_host_env::ssl::{

chain::{self, VerifiedChainBuilder},

cipher, oid,

providers::CryptoExt,

};

// Import compat module (OpenSSL compatibility layer)

use super::compat::{

ClientConfigOptions, MultiCertResolver, ProtocolSettings, ServerConfigOptions, SslError,

create_client_config, create_server_config, curve_name_to_kx_group, is_blocking_io_error,

ssl_do_handshake,

};

use super::handshake::TlsState;

// Type aliases for better readability

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

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

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

type CapathStamp = (String, Option<SystemTime>);

type CapathCache = (Vec<CapathStamp>, Arc<Vec<Vec<u8>>>);

#[derive(Debug, Default)]

struct CapathState {

directories: Vec<String>,

cache: Option<CapathCache>,

}

// Constants matching Python ssl module

// SSL/TLS Protocol versions

#[pyattr]

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

#[pyattr]

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

#[pyattr]

const PROTOCOL_TLS_CLIENT: i32 = rustpython_host_env::ssl::PROTOCOL_TLS_CLIENT;

#[pyattr]

const PROTOCOL_TLS_SERVER: i32 = rustpython_host_env::ssl::PROTOCOL_TLS_SERVER;

// 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 = rustpython_host_env::ssl::PROTOCOL_TLSV1;

#[pyattr]

const PROTOCOL_TLSv1_1: i32 = rustpython_host_env::ssl::PROTOCOL_TLSV1_1;

#[pyattr]

const PROTOCOL_TLSv1_2: i32 = rustpython_host_env::ssl::PROTOCOL_TLSV1_2;

#[pyattr]

const PROTOCOL_TLSv1_3: i32 = rustpython_host_env::ssl::PROTOCOL_TLSV1_3;

#[pyattr]

use rustpython_host_env::ssl::{

PROTO_MAXIMUM_SUPPORTED, PROTO_MINIMUM_SUPPORTED, PROTO_SSL3 as PROTO_SSLv3,

PROTO_TLSV1 as PROTO_TLSv1, PROTO_TLSV1_1 as PROTO_TLSv1_1, PROTO_TLSV1_2 as PROTO_TLSv1_2,

PROTO_TLSV1_3 as PROTO_TLSv1_3,

};

// 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;

pub(crate) const SSL3_RT_MAX_PACKET_SIZE: usize =

rustpython_host_env::ssl::SSL3_RT_MAX_PACKET_SIZE;

const SSL_SESSION_CACHE_SIZE: usize = rustpython_host_env::ssl::SESSION_CACHE_SIZE;

// Certificate verification modes

#[pyattr]

const CERT_NONE: i32 = rustpython_host_env::ssl::CERT_NONE;

#[pyattr]

const CERT_OPTIONAL: i32 = rustpython_host_env::ssl::CERT_OPTIONAL;

#[pyattr]

const CERT_REQUIRED: i32 = rustpython_host_env::ssl::CERT_REQUIRED;

// SSL Verification Flags / Certificate requirements

#[pyattr]

const VERIFY_DEFAULT: i32 = rustpython_host_env::ssl::VERIFY_DEFAULT;

#[pyattr]

const VERIFY_CRL_CHECK_LEAF: i32 = rustpython_host_env::ssl::VERIFY_CRL_CHECK_LEAF;

#[pyattr]

const VERIFY_CRL_CHECK_CHAIN: i32 = rustpython_host_env::ssl::VERIFY_CRL_CHECK_CHAIN;

// VERIFY_X509_STRICT flag for RFC 5280 strict compliance

// When set, performs additional validation including AKI extension checks

#[pyattr]

pub(crate) const VERIFY_X509_STRICT: i32 = rustpython_host_env::ssl::VERIFY_X509_STRICT;

#[pyattr]

const VERIFY_ALLOW_PROXY_CERTS: i32 = rustpython_host_env::ssl::VERIFY_ALLOW_PROXY_CERTS;

#[pyattr]

const VERIFY_X509_TRUSTED_FIRST: i32 = rustpython_host_env::ssl::VERIFY_X509_TRUSTED_FIRST;

// VERIFY_X509_PARTIAL_CHAIN flag for partial chain validation

// When set, accept certificates if any certificate in the chain is in the trust store

// (not just root CAs). This matches OpenSSL's X509_V_FLAG_PARTIAL_CHAIN behavior.

#[pyattr]

pub(crate) const VERIFY_X509_PARTIAL_CHAIN: i32 =

rustpython_host_env::ssl::VERIFY_X509_PARTIAL_CHAIN;

#[pyattr]

const HOSTFLAG_NEVER_CHECK_SUBJECT: i32 =

rustpython_host_env::ssl::HOSTFLAG_NEVER_CHECK_SUBJECT;

#[pyattr]

use rustpython_host_env::ssl::{

OP_ALL, OP_CIPHER_SERVER_PREFERENCE, OP_ENABLE_MIDDLEBOX_COMPAT, OP_IGNORE_UNEXPECTED_EOF,

OP_LEGACY_SERVER_CONNECT, OP_NO_COMPRESSION, OP_NO_RENEGOTIATION, OP_NO_SSLv2, OP_NO_SSLv3,

OP_NO_TICKET, OP_NO_TLSV1 as OP_NO_TLSv1, OP_NO_TLSV1_1 as OP_NO_TLSv1_1,

OP_NO_TLSV1_2 as OP_NO_TLSv1_2, OP_NO_TLSV1_3 as OP_NO_TLSv1_3, OP_SINGLE_DH_USE,

OP_SINGLE_ECDH_USE,

};

#[pyattr]

use rustpython_host_env::ssl::{

ALERT_DESCRIPTION_ACCESS_DENIED, ALERT_DESCRIPTION_BAD_CERTIFICATE,

ALERT_DESCRIPTION_BAD_CERTIFICATE_HASH_VALUE,

ALERT_DESCRIPTION_BAD_CERTIFICATE_STATUS_RESPONSE, ALERT_DESCRIPTION_BAD_RECORD_MAC,

ALERT_DESCRIPTION_CERTIFICATE_EXPIRED, ALERT_DESCRIPTION_CERTIFICATE_REQUIRED,

ALERT_DESCRIPTION_CERTIFICATE_REVOKED, ALERT_DESCRIPTION_CERTIFICATE_UNKNOWN,

ALERT_DESCRIPTION_CERTIFICATE_UNOBTAINABLE, ALERT_DESCRIPTION_CLOSE_NOTIFY,

ALERT_DESCRIPTION_DECODE_ERROR, ALERT_DESCRIPTION_DECOMPRESSION_FAILURE,

ALERT_DESCRIPTION_DECRYPT_ERROR, ALERT_DESCRIPTION_DECRYPTION_FAILED,

ALERT_DESCRIPTION_EXPORT_RESTRICTION, ALERT_DESCRIPTION_HANDSHAKE_FAILURE,

ALERT_DESCRIPTION_ILLEGAL_PARAMETER, ALERT_DESCRIPTION_INAPPROPRIATE_FALLBACK,

ALERT_DESCRIPTION_INSUFFICIENT_SECURITY, ALERT_DESCRIPTION_INTERNAL_ERROR,

ALERT_DESCRIPTION_MISSING_EXTENSION, ALERT_DESCRIPTION_NO_APPLICATION_PROTOCOL,

ALERT_DESCRIPTION_NO_CERTIFICATE, ALERT_DESCRIPTION_NO_RENEGOTIATION,

ALERT_DESCRIPTION_PROTOCOL_VERSION, ALERT_DESCRIPTION_RECORD_OVERFLOW,

ALERT_DESCRIPTION_UNEXPECTED_MESSAGE, ALERT_DESCRIPTION_UNKNOWN_CA,

ALERT_DESCRIPTION_UNKNOWN_PSK_IDENTITY, ALERT_DESCRIPTION_UNRECOGNIZED_NAME,

ALERT_DESCRIPTION_UNSUPPORTED_CERTIFICATE, ALERT_DESCRIPTION_UNSUPPORTED_EXTENSION,

ALERT_DESCRIPTION_USER_CANCELLED,

};

// `ssl.py` still requires OpenSSL-shaped numeric compatibility fields even

// for non-OpenSSL TLS providers. Keep them in the supported 3.x ABI range,

// but report the actual rustls/AWS-LC backend in the human-readable string.

#[pyattr]

const OPENSSL_VERSION_NUMBER: i32 = 0x30300000;

#[pyattr]

const OPENSSL_VERSION: &str = "OpenSSL 3.3.0-compatible (AWS-LC/rustls 0.23)";

#[pyattr]

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

#[pyattr]

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

#[pyattr(once)]

fn _DEFAULT_CIPHERS(_vm: &VirtualMachine) -> String {

cipher::default_cipher_string()

}

// Has features

#[pyattr]

const HAS_SNI: bool = true;

#[pyattr]

const HAS_TLS_UNIQUE: bool = true;

#[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

#[pyattr]

use rustpython_host_env::ssl::{ENCODING_DER, ENCODING_PEM, ENCODING_PEM_AUX};

/// Validate server hostname for TLS SNI.

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

use rustpython_host_env::ssl::HostnameError;

match rustpython_host_env::ssl::validate_hostname(hostname) {

Ok(()) => Ok(()),

Err(HostnameError::EmbeddedNul) => {

cold_path();

Err(exceptions::nul_char_type_error(vm))

}

Err(error) => Err(vm.new_value_error(error.message().to_owned())),

}

}

/// Parse `[len][proto]...` the way `ssl.py` hands it to `_set_alpn_protocols`.

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

rustpython_host_env::ssl::parse_length_prefixed_alpn(bytes)

.map_err(|error| vm.new_value_error(error.0))

}

// SSLContext - manages TLS configuration

#[pyattr]

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

#[derive(Debug, PyPayload)]

struct PySSLContext {

#[pytraverse(skip)]

context_identity: Arc<()>,

#[pytraverse(skip)]

protocol: i32,

#[pytraverse(skip)]

check_hostname: PyRwLock<bool>,

#[pytraverse(skip)]

host_flags: PyRwLock<i32>,

#[pytraverse(skip)]

verify_mode: PyRwLock<i32>,

#[pytraverse(skip)]

verify_flags: PyRwLock<i32>,

// Rustls configuration (built lazily)

#[pytraverse(skip)]

server_config: PyRwLock<Option<chain::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>>>,

// OpenSSL-style hashed CA directories, materialized when a connection

// needs an immutable rustls configuration.

#[pytraverse(skip)]

capath_state: PyRwLock<CapathState>,

// 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

#[pytraverse(skip)]

options: PyRwLock<i32>,

// ALPN protocols

#[pytraverse(skip)]

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

// 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>>,

keylog_filename: PyRwLock<Option<PyObjectRef>>,

#[pytraverse(skip)]

key_log: Arc<super::keylog::KeyLog>,

// 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 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>>>,

/// Key exchange groups a SUITEB* cipher string pinned (RFC 6460)

#[pytraverse(skip)]

suite_b_kx_groups: PyRwLock<Option<Vec<&'static dyn SupportedKxGroup>>>,

}

#[derive(FromArgs)]

struct WrapSocketArgs {

sock: PyObjectRef,

server_side: bool,

#[pyarg(any, optional)]

server_hostname: Option<PyUtf8StrRef>,

#[pyarg(named, optional)]

owner: Option<PyObjectRef>,

#[pyarg(named, optional)]

session: Option<PyObjectRef>,

}

#[derive(FromArgs)]

struct WrapBioArgs {

incoming: PyRef<PyMemoryBIO>,

outgoing: PyRef<PyMemoryBIO>,

server_side: bool,

#[pyarg(any, optional)]

server_hostname: Option<PyUtf8StrRef>,

#[pyarg(named, optional)]

owner: Option<PyObjectRef>,

#[pyarg(named, optional)]

session: Option<PyObjectRef>,

}

#[derive(FromArgs)]

struct LoadVerifyLocationsArgs {

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

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

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

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

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

cadata: 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: Option<Either<PyStrRef, ArgBytesLike>>,

#[pyarg(any, optional)]

password: Option<PyObjectRef>,

}

#[derive(FromArgs)]

struct GetCertArgs {

#[pyarg(positional, default)]

der: bool,

}

#[derive(FromArgs)]

struct GetCaCertsArgs {

#[pyarg(any, default)]

binary_form: bool,

}

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

impl PySSLContext {

fn warn_deprecated_tls_version(version: i32, vm: &VirtualMachine) -> PyResult<()> {

let version_name = match version {

PROTO_SSLv3 => Some("SSLv3"),

PROTO_TLSv1 => Some("TLSv1"),

PROTO_TLSv1_1 => Some("TLSv1_1"),

_ => None,

};

if let Some(version_name) = version_name {

_warnings::warn(

vm.ctx.exceptions.deprecation_warning,

format!("ssl.TLSVersion.{version_name} is deprecated"),

2,

vm,

)?;

}

Ok(())

}

// 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)

}

fn add_verify_dir(&self, directory: String) {

let mut capath_state = self.capath_state.write();

if capath_state

.directories

.iter()

.any(|known| known == &directory)

{

return;

}

capath_state.directories.push(directory);

capath_state.cache = None;

drop(capath_state);

*self.server_config.write() = None;

}

fn capath_certificates(&self) -> Arc<Vec<Vec<u8>>> {

let directories = self.capath_state.read().directories.clone();

if directories.is_empty() {

return Arc::new(Vec::new());

}

let stamps = directories

.iter()

.map(|directory| {

let modified = rustpython_host_env::fs::metadata(directory)

.and_then(|metadata| metadata.modified())

.ok();

(directory.clone(), modified)

})

.collect::<Vec<_>>();

if let Some((cached_stamps, certificates)) = self.capath_state.read().cache.as_ref()

&& cached_stamps == &stamps

{

return certificates.clone();

}

let mut store = RootCertStore::empty();

let mut certificates = Vec::new();

let mut loader = cert::CertLoader::new(&mut store, &mut certificates);

for directory in &directories {

let _ = loader.load_from_dir(directory);

}

let certificates = Arc::new(certificates);

let mut capath_state = self.capath_state.write();

if capath_state.directories == directories {

capath_state.cache = Some((stamps, certificates.clone()));

}

certificates

}

fn verification_roots(&self) -> (RootCertStore, Vec<Vec<u8>>) {

let mut root_store = self.root_certs.read().clone();

let mut ca_certs_der = self.ca_certs_der.read().clone();

for certificate in self.capath_certificates().iter() {

let _ = root_store.add(certificate.clone().into());

if !ca_certs_der.iter().any(|known| known == certificate) {

ca_certs_der.push(certificate.clone());

}

}

(root_store, ca_certs_der)

}

#[pygetset]

fn check_hostname(zelf: &Py<Self>) -> bool {

*zelf.check_hostname.read()

}

#[pygetset(setter)]

fn set_check_hostname(zelf: &Py<Self>, value: bool) {

*zelf.check_hostname.write() = value;

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

if value {

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

if current_verify_mode == CERT_NONE {

*zelf.verify_mode.write() = CERT_REQUIRED;

}

}

}

#[pygetset]

fn _host_flags(zelf: &Py<Self>) -> i32 {

*zelf.host_flags.read()

}

#[pygetset(setter)]

fn set__host_flags(zelf: &Py<Self>, value: i32) {

*zelf.host_flags.write() = value;

}

#[pygetset]

fn verify_mode(zelf: &Py<Self>) -> i32 {

*zelf.verify_mode.read()

}

#[pygetset(setter)]

fn set_verify_mode(zelf: &Py<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 && *zelf.check_hostname.read() {

return Err(vm.new_value_error(

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

));

}

*zelf.verify_mode.write() = mode;

Ok(())

}

#[pygetset]

fn protocol(zelf: &Py<Self>) -> i32 {

zelf.protocol

}

#[pygetset]

fn verify_flags(zelf: &Py<Self>) -> i32 {

*zelf.verify_flags.read()

}

#[pygetset(setter)]

fn set_verify_flags(zelf: &Py<Self>, value: i32) {

*zelf.verify_flags.write() = value;

}

#[pygetset]

fn post_handshake_auth(zelf: &Py<Self>) -> bool {

*zelf.post_handshake_auth.read()

}

#[pygetset(setter)]

fn set_post_handshake_auth(zelf: &Py<Self>, value: bool) {

*zelf.post_handshake_auth.write() = value;

}

#[pygetset]

fn num_tickets(zelf: &Py<Self>) -> i32 {

*zelf.num_tickets.read()

}

#[pygetset(setter)]

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

if value < 0 {

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

}

if zelf.protocol != PROTOCOL_TLS_SERVER {

return Err(

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

);

}

*zelf.num_tickets.write() = value;

Ok(())

}

#[pygetset]

fn options(zelf: &Py<Self>) -> i32 {

*zelf.options.read()

}

#[pygetset(setter)]

fn set_options(zelf: &Py<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 = *zelf.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,

)?;

}

*zelf.options.write() = value;

Ok(())

}

#[pygetset]

fn minimum_version(zelf: &Py<Self>) -> i32 {

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

// return MINIMUM_SUPPORTED if value is 0

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

}

#[pygetset(setter)]

fn set_minimum_version(zelf: &Py<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

&& !(PROTO_SSLv3..=PROTO_TLSv1_3).contains(&value)

{

return Err(vm.new_value_error(format!("invalid protocol version: {value}")));

}

Self::warn_deprecated_tls_version(value, vm)?;

// Convert special values to rustls actual supported versions

// MINIMUM_SUPPORTED (-2) -> 0 (auto-negotiate)

// MAXIMUM_SUPPORTED (-1) -> MAXIMUM_VERSION (TLSv1.3)

let normalized_value = match value {

PROTO_MINIMUM_SUPPORTED => 0, // Auto-negotiate

PROTO_MAXIMUM_SUPPORTED => MAXIMUM_VERSION, // TLSv1.3

_ => value,

};

*zelf.minimum_version.write() = normalized_value;

Ok(())

}

#[pygetset]

fn maximum_version(zelf: &Py<Self>) -> i32 {

let v = *zelf.maximum_version.read();

// return MAXIMUM_SUPPORTED if value is 0

if v == 0 { PROTO_MAXIMUM_SUPPORTED } else { v }

}

#[pygetset(setter)]

fn set_maximum_version(zelf: &Py<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

&& !(PROTO_SSLv3..=PROTO_TLSv1_3).contains(&value)

{

return Err(vm.new_value_error(format!("invalid protocol version: {value}")));

}

Self::warn_deprecated_tls_version(value, vm)?;

// Convert special values to rustls actual supported versions

// MAXIMUM_SUPPORTED (-1) -> 0 (auto-negotiate)

// MINIMUM_SUPPORTED (-2) -> MINIMUM_VERSION (TLSv1.2)

let normalized_value = match value {

PROTO_MAXIMUM_SUPPORTED => 0, // Auto-negotiate

PROTO_MINIMUM_SUPPORTED => MINIMUM_VERSION, // TLSv1.2

_ => value,

};

*zelf.maximum_version.write() = normalized_value;

Ok(())

}

#[pymethod]

fn load_cert_chain(

zelf: &Py<Self>,

args: LoadCertChainArgs,

vm: &VirtualMachine,

) -> PyResult<()> {

let crypto_ext = CryptoExt::get_ext();

// Parse certfile argument (str or bytes) to path

let cert_path = Self::parse_path_arg(&args.certfile, vm)?;

// Parse keyfile argument (default to certfile if not provided)

let key_path = match args.keyfile {

Some(ref k) => Self::parse_path_arg(k, vm)?,

None => cert_path.clone(),

};

// Parse password argument (str, bytes-like, or callable)

// Callable passwords are NOT invoked immediately (lazy evaluation)

let (password_str, password_callable) =

Self::parse_password_argument(args.password.as_ref(), vm)?;

// Validate immediate password length (limit: PEM_BUFSIZE = 1024 bytes)

if let Some(ref pwd) = password_str

&& pwd.len() > PEM_BUFSIZE

{

return Err(vm.new_value_error(format!(

"password cannot be longer than {PEM_BUFSIZE} bytes",

)));

}

// First attempt: Load with immediate password (or None if callable)

let mut result =

cert::load_cert_chain_from_file(&cert_path, &key_path, password_str.as_deref());

// If failed and callable exists, invoke it and retry

// This implements lazy evaluation: callable only invoked if password is actually needed

if result.is_err()

&& let Some(callable) = password_callable

{

// Invoke callable - exceptions propagate naturally

let pwd_result = callable.call((), vm)?;

// Convert callable result to string

let password_from_callable =

if let Ok(pwd_str) = PyUtf8StrRef::try_from_object(vm, pwd_result.clone()) {

pwd_str.as_str().to_owned()

} else if pwd_result.check_buffer() {

let pwd_bytes_like = ArgBytesLike::try_from_object(vm, pwd_result)?;

String::from_utf8(pwd_bytes_like.borrow_buf().to_vec()).map_err(|_| {

vm.new_type_error("password callback returned invalid UTF-8 bytes")

})?

} else {

return Err(

vm.new_type_error("password callback must return a string or bytes")

);

};

// Validate callable password length

if password_from_callable.len() > PEM_BUFSIZE {

return Err(vm.new_value_error(format!(

"password cannot be longer than {PEM_BUFSIZE} bytes",

)));

}

// Retry with callable password

result = cert::load_cert_chain_from_file(

&cert_path,

&key_path,

Some(&password_from_callable),

);

}

// Process result

let (certs, key) = result.map_err(|e| {

// Try to downcast to io::Error to preserve errno information

if let Ok(io_err) = e.downcast::<std::io::Error>() {

match io_err.kind() {

// File access errors (NotFound, PermissionDenied) - preserve errno

std::io::ErrorKind::NotFound | std::io::ErrorKind::PermissionDenied => {

io_err.into_pyexception(vm)

}

// Other io::Error types

std::io::ErrorKind::Other => {

let msg = io_err.to_string();

if msg.contains("Failed to decrypt") || msg.contains("wrong password") {

// Wrong password error

vm.new_os_subtype_error(

PySSLError::class(&vm.ctx).to_owned(),

None,

msg,

)

.upcast()

} else {

// [SSL] PEM lib

super::compat::SslError::create_ssl_error_with_reason(

vm,

Some("SSL"),

"",

"PEM lib",

)

}

}

// PEM parsing errors - [SSL] PEM lib

_ => super::compat::SslError::create_ssl_error_with_reason(

vm,

Some("SSL"),

"",

"PEM lib",

),

}

} else {

// Unknown error type - [SSL] PEM lib

super::compat::SslError::create_ssl_error_with_reason(

vm,

Some("SSL"),

"",

"PEM lib",

)

}

})?;

// Validate certificate and key match

cert::validate_cert_key_match(&certs, &key).map_err(|e| {

let msg = if e.contains("key values mismatch") {

"[SSL: KEY_VALUES_MISMATCH] key values mismatch".to_owned()

} else {

e

};

vm.new_os_subtype_error(PySSLError::class(&vm.ctx).to_owned(), Some(0), msg)