From 51e886c2249245f8c9082ce285a820f92957f1bc Mon Sep 17 00:00:00 2001 From: youknowdot Date: Sat, 17 Oct 2026 07:36:27 -0700 Subject: [PATCH] Fix optional OpenSSL duplex I/O on Unix (#9173) * Fix optional OpenSSL duplex I/O on Unix Release connection locks before readiness waits while serializing native SSL attempts and preserving owned write retries. Wake waiters on native progress and socket identity invalidation, and construct callback/native exceptions after unlocking. Add regression coverage to the existing SSL snippet and run the optional OpenSSL backend in the existing Unix CI matrix. Windows and AIX retain the legacy transport pending a separate compatibility-safe driver. Assisted-by: OpenAI Codex:model version unavailable * Consolidate OpenSSL runtime checks into the existing CI step Keep the minimal system and vendored OpenSSL build checks, and run the Unix SSL regressions in the existing system OpenSSL step. Preserve the runtime feature set and enable its macOS dependency in that job. Assisted-by: OpenAI Codex:model version unavailable --- .github/workflows/ci.yaml | 11 +- crates/stdlib/src/openssl.rs | 1167 ++++++++++++++++++++++++--- crates/stdlib/src/openssl/driver.rs | 264 ++++++ crates/stdlib/src/openssl/waiter.rs | 285 +++++++ crates/stdlib/src/socket.rs | 300 ++++++- extra_tests/snippets/stdlib_ssl.py | 305 +++++++ 6 files changed, 2205 insertions(+), 127 deletions(-) create mode 100644 crates/stdlib/src/openssl/driver.rs create mode 100644 crates/stdlib/src/openssl/waiter.rs diff --git a/.github/workflows/ci.yaml b/.github/workflows/ci.yaml index 704df8b2a..ea8f546de 100644 --- a/.github/workflows/ci.yaml +++ b/.github/workflows/ci.yaml @@ -275,6 +275,8 @@ jobs: - name: Install macOS dependencies if: env.RUN_RUST_TESTS == 'true' uses: ./.github/actions/install-macos-deps + with: + openssl: true - name: Build rust tests if: env.RUN_RUST_TESTS == 'true' @@ -327,8 +329,13 @@ jobs: target/debug/rustpython extra_tests/snippets/stdlib_re.py - name: Test openssl build - if: env.RUN_RUST_TESTS == 'true' && runner.os == 'Linux' - run: cargo build --locked --no-default-features --features ssl-openssl + if: env.RUN_RUST_TESTS == 'true' && runner.os != 'Windows' + run: | + cargo build --locked --no-default-features --features ssl-openssl + cargo build --locked --release --features threading,jit,ssl-openssl ${{ env.CARGO_ARGS_NO_SSL }} + target/release/rustpython -B extra_tests/snippets/stdlib_ssl.py + target/release/rustpython -B -m test -v --single-process --timeout=300 test_ssl + timeout-minutes: 30 - name: Test vendored OpenSSL build if: env.RUN_RUST_TESTS == 'true' && runner.os == 'Linux' diff --git a/crates/stdlib/src/openssl.rs b/crates/stdlib/src/openssl.rs index e9393bd44..25f7422ae 100644 --- a/crates/stdlib/src/openssl.rs +++ b/crates/stdlib/src/openssl.rs @@ -1,6 +1,10 @@ // spell-checker:disable mod cert; +#[cfg(all(unix, not(target_os = "aix")))] +mod driver; +#[cfg(all(unix, not(target_os = "aix")))] +mod waiter; // SSL exception types (shared with rustls backend) #[allow( @@ -42,6 +46,17 @@ mod _ssl { use core::hint::cold_path; use super::bio; + #[cfg(all(unix, not(target_os = "aix")))] + use super::{ + driver::{Driver, Needs, Permit, WriteOwner}, + waiter::{ProgressRegistry, ProgressWaiter}, + }; + #[cfg(not(all(unix, not(target_os = "aix"))))] + use crate::socket::{SockWaitKind, sock_wait}; + #[cfg(all(unix, not(target_os = "aix")))] + use alloc::sync::Arc; + #[cfg(all(unix, not(target_os = "aix")))] + use std::os::fd::AsRawFd; // Import error types and helpers used in this module (others are exposed via pymodule(with(...))) use super::ssl_error::{ @@ -53,7 +68,7 @@ mod _ssl { LazyLock, PyDetachingRwLock, PyMappedRwLockReadGuard, PyMutex, PyRwLock, PyRwLockReadGuard, PyRwLockWriteGuard, }, - socket::{self, PySocket, SockWaitKind, sock_wait}, + socket::{self, PySocket}, vm::{ AsObject, Py, PyObject, PyObjectRef, PyPayload, PyRef, PyResult, VirtualMachine, builtins::{ @@ -479,6 +494,8 @@ mod _ssl { // SSL pointer during handshake - needed because connection lock is held during handshake // and callbacks may need to access SSL without acquiring the lock static HANDSHAKE_SSL_PTR: core::cell::Cell> = const { core::cell::Cell::new(None) }; + #[cfg(all(unix, not(target_os = "aix")))] + static SSL_IO_CALLS: core::cell::RefCell> = const { core::cell::RefCell::new(Vec::new()) }; static HANDSHAKE_CONNECTIONS: core::cell::RefCell> = const { core::cell::RefCell::new(Vec::new()) }; } @@ -2194,19 +2211,26 @@ mod _ssl { Self::new_py_ssl_socket(zelf.clone(), args.server_side, args.server_hostname, vm)?; // Create SslStream with socket - let stream = ssl::SslStream::new(ssl, SocketStream(args.sock.clone())) - .map_err(|e| convert_openssl_error(vm, e))?; + let stream = ssl::SslStream::new( + ssl, + SocketStream::new(args.sock.clone()).map_err(|e| e.to_pyexception(vm))?, + ) + .map_err(|e| convert_openssl_error(vm, e))?; let py_ssl_socket = PySslSocket { ctx: PyRwLock::new(zelf.clone()), - connection: SslConnectionLock(PyDetachingRwLock::new(SslConnection::Socket( - stream, + connection: SslConnectionLock(PyDetachingRwLock::new(SslConnection::new( + SslTransport::Socket(stream), ))), socket_type, server_hostname, got_eof_error: AtomicCell::new(false), shutdown_seen_zero: AtomicCell::new(false), callback_error: PyMutex::new(None), + #[cfg(all(unix, not(target_os = "aix")))] + info_progress: AtomicCell::new(0), + #[cfg(all(unix, not(target_os = "aix")))] + previous_info_callback: AtomicCell::new(None), owner: PyRwLock::new(Some( args.owner .unwrap_or_else(|| args.sock.clone().into()) @@ -2228,6 +2252,9 @@ mod _ssl { ); } + #[cfg(all(unix, not(target_os = "aix")))] + py_ref.install_progress_callback(); + // Set session if provided if let Some(session) = args.session && !vm.is_none(&session) @@ -2252,6 +2279,8 @@ mod _ssl { let bio_stream = BioStream { inbio: args.incoming, outbio: args.outgoing, + #[cfg(all(unix, not(target_os = "aix")))] + progress: 0, }; // Create SslStream with BioStream @@ -2260,12 +2289,18 @@ mod _ssl { let py_ssl_socket = PySslSocket { ctx: PyRwLock::new(zelf.clone()), - connection: SslConnectionLock(PyDetachingRwLock::new(SslConnection::Bio(stream))), + connection: SslConnectionLock(PyDetachingRwLock::new(SslConnection::new( + SslTransport::Bio(stream), + ))), socket_type, server_hostname, got_eof_error: AtomicCell::new(false), shutdown_seen_zero: AtomicCell::new(false), callback_error: PyMutex::new(None), + #[cfg(all(unix, not(target_os = "aix")))] + info_progress: AtomicCell::new(0), + #[cfg(all(unix, not(target_os = "aix")))] + previous_info_callback: AtomicCell::new(None), owner: PyRwLock::new(args.owner.map(|o| o.downgrade(None, vm)).transpose()?), }; @@ -2283,6 +2318,9 @@ mod _ssl { ); } + #[cfg(all(unix, not(target_os = "aix")))] + py_ref.install_progress_callback(); + // Set session if provided if let Some(session) = args.session && !vm.is_none(&session) @@ -2354,6 +2392,7 @@ mod _ssl { // Err is true if the socket is blocking type SocketDeadline = Result; + #[cfg(not(all(unix, not(target_os = "aix"))))] enum SelectRet { Nonblocking, TimedOut, @@ -2361,26 +2400,47 @@ mod _ssl { Ok, } + #[cfg(not(all(unix, not(target_os = "aix"))))] #[derive(Clone, Copy)] enum SslNeeds { Read, Write, } - struct SocketStream(PyRef); + struct SocketStream { + socket: PyRef, + #[cfg(all(unix, not(target_os = "aix")))] + identity: Arc, + #[cfg(all(unix, not(target_os = "aix")))] + progress: u64, + #[cfg(all(unix, not(target_os = "aix")))] + interrupted: bool, + } impl SocketStream { + fn new(socket: PyRef) -> std::io::Result { + Ok(Self { + #[cfg(all(unix, not(target_os = "aix")))] + identity: socket.ssl_identity()?, + socket, + #[cfg(all(unix, not(target_os = "aix")))] + progress: 0, + #[cfg(all(unix, not(target_os = "aix")))] + interrupted: false, + }) + } fn timeout_deadline(&self) -> SocketDeadline { - self.0.get_timeout().map(|d| Instant::now() + d) + self.socket.get_timeout().map(|d| Instant::now() + d) } + #[cfg(not(all(unix, not(target_os = "aix"))))] fn select( &self, needs: SslNeeds, deadline: &SocketDeadline, vm: &VirtualMachine, ) -> PyResult { - let Some(sock) = self.0.sock_opt() else { + let Some(sock) = self.socket.sock_opt() else { return Ok(SelectRet::Closed); }; // For blocking sockets without timeout, call sock_select with None timeout @@ -2406,6 +2466,7 @@ mod _ssl { }) } + #[cfg(not(all(unix, not(target_os = "aix"))))] fn socket_needs( &self, err: &ssl::Error, @@ -2431,6 +2492,8 @@ mod _ssl { struct BioStream { inbio: PyRef, outbio: PyRef, + #[cfg(all(unix, not(target_os = "aix")))] + progress: u64, } impl Read for BioStream { @@ -2450,6 +2513,10 @@ mod _ssl { "BIO has no data available", )) } else { + #[cfg(all(unix, not(target_os = "aix")))] + { + self.progress = self.progress.wrapping_add(nbytes as u64); + } Ok(nbytes as usize) } } @@ -2468,6 +2535,10 @@ mod _ssl { if nbytes < 0 { return Err(std::io::Error::other("BIO write failed")); } + #[cfg(all(unix, not(target_os = "aix")))] + { + self.progress = self.progress.wrapping_add(nbytes as u64); + } Ok(nbytes as usize) } } @@ -2479,12 +2550,12 @@ mod _ssl { } // Enum to represent different SSL connection modes - enum SslConnection { + enum SslTransport { Socket(ssl::SslStream), Bio(ssl::SslStream), } - impl Drop for SslConnection { + impl Drop for SslTransport { fn drop(&mut self) { // Match CPython: freeing a socket without unwrap() must not invalidate // a session retained by Python. This marks state without doing I/O. @@ -2496,26 +2567,27 @@ mod _ssl { } } - impl SslConnection { + impl SslTransport { // Get a reference to the SSL object fn ssl(&self) -> &ssl::SslRef { match self { - SslConnection::Socket(stream) => stream.ssl(), - SslConnection::Bio(stream) => stream.ssl(), + SslTransport::Socket(stream) => stream.ssl(), + SslTransport::Bio(stream) => stream.ssl(), } } // Get underlying socket stream reference (only for socket mode) fn get_ref(&self) -> Option<&SocketStream> { match self { - SslConnection::Socket(stream) => Some(stream.get_ref()), - SslConnection::Bio(_) => None, + SslTransport::Socket(stream) => Some(stream.get_ref()), + SslTransport::Bio(_) => None, } } - // Check if this is in BIO mode + // Check if this is in BIO mode (legacy blocking transport). + #[cfg(not(all(unix, not(target_os = "aix"))))] fn is_bio(&self) -> bool { - matches!(self, SslConnection::Bio(_)) + matches!(self, SslTransport::Bio(_)) } // Perform SSL handshake @@ -2523,8 +2595,8 @@ mod _ssl { // SSL_get_error requires an empty queue before this native attempt. unsafe { sys::ERR_clear_error() }; match self { - SslConnection::Socket(stream) => stream.do_handshake(), - SslConnection::Bio(stream) => stream.do_handshake(), + SslTransport::Socket(stream) => stream.do_handshake(), + SslTransport::Bio(stream) => stream.do_handshake(), } } @@ -2532,8 +2604,8 @@ mod _ssl { fn ssl_write(&mut self, buf: &[u8]) -> Result { unsafe { sys::ERR_clear_error() }; match self { - SslConnection::Socket(stream) => stream.ssl_write(buf), - SslConnection::Bio(stream) => stream.ssl_write(buf), + SslTransport::Socket(stream) => stream.ssl_write(buf), + SslTransport::Bio(stream) => stream.ssl_write(buf), } } @@ -2541,36 +2613,90 @@ mod _ssl { fn ssl_read(&mut self, buf: &mut [u8]) -> Result { unsafe { sys::ERR_clear_error() }; match self { - SslConnection::Socket(stream) => stream.ssl_read(buf), - SslConnection::Bio(stream) => stream.ssl_read(buf), + SslTransport::Socket(stream) => stream.ssl_read(buf), + SslTransport::Bio(stream) => stream.ssl_read(buf), } } fn shutdown(&mut self) -> Result { unsafe { sys::ERR_clear_error() }; match self { - SslConnection::Socket(stream) => stream.shutdown(), - SslConnection::Bio(stream) => stream.shutdown(), + SslTransport::Socket(stream) => stream.shutdown(), + SslTransport::Bio(stream) => stream.shutdown(), } } // Get SSL shutdown state fn get_shutdown(&mut self) -> ssl::ShutdownState { match self { - SslConnection::Socket(stream) => stream.get_shutdown(), - SslConnection::Bio(stream) => stream.get_shutdown(), + SslTransport::Socket(stream) => stream.get_shutdown(), + SslTransport::Bio(stream) => stream.get_shutdown(), } } // Check if incoming BIO has EOF (for BIO mode only) fn is_bio_eof(&self) -> bool { match self { - SslConnection::Socket(_) => false, - SslConnection::Bio(stream) => stream.get_ref().inbio.eof_written.load(), + SslTransport::Socket(_) => false, + SslTransport::Bio(stream) => stream.get_ref().inbio.eof_written.load(), } } } + struct SslConnection { + transport: SslTransport, + #[cfg(all(unix, not(target_os = "aix")))] + driver: Driver, + #[cfg(all(unix, not(target_os = "aix")))] + progress: Arc, + } + + impl SslConnection { + fn new(transport: SslTransport) -> Self { + Self { + transport, + #[cfg(all(unix, not(target_os = "aix")))] + driver: Driver::default(), + #[cfg(all(unix, not(target_os = "aix")))] + progress: Arc::new(ProgressRegistry::default()), + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + fn notify_progress(&self) { + self.progress.notify(); + } + + #[cfg(all(unix, not(target_os = "aix")))] + fn io_progress(&self) -> u64 { + match &self.transport { + SslTransport::Socket(stream) => stream.get_ref().progress, + SslTransport::Bio(stream) => stream.get_ref().progress, + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + fn take_interrupted(&mut self) -> bool { + match &mut self.transport { + SslTransport::Socket(stream) => core::mem::take(&mut stream.get_mut().interrupted), + SslTransport::Bio(_) => false, + } + } + } + + impl core::ops::Deref for SslConnection { + type Target = SslTransport; + fn deref(&self) -> &Self::Target { + &self.transport + } + } + + impl core::ops::DerefMut for SslConnection { + fn deref_mut(&mut self) -> &mut Self::Target { + &mut self.transport + } + } + struct SslConnectionLock(PyDetachingRwLock); impl core::ops::Deref for SslConnectionLock { @@ -2585,9 +2711,9 @@ mod _ssl { unsafe impl Traverse for SslConnectionLock { fn traverse(&self, traverse_fn: &mut TraverseFn<'_>) { if let Some(connection) = self.try_read() { - match &*connection { - SslConnection::Socket(stream) => stream.get_ref().0.traverse(traverse_fn), - SslConnection::Bio(stream) => { + match &connection.transport { + SslTransport::Socket(stream) => stream.get_ref().socket.traverse(traverse_fn), + SslTransport::Bio(stream) => { stream.get_ref().inbio.traverse(traverse_fn); stream.get_ref().outbio.traverse(traverse_fn); } @@ -2611,6 +2737,12 @@ mod _ssl { #[pytraverse(skip)] shutdown_seen_zero: AtomicCell, callback_error: PyMutex>, + #[cfg(all(unix, not(target_os = "aix")))] + #[pytraverse(skip)] + info_progress: AtomicCell, + #[cfg(all(unix, not(target_os = "aix")))] + #[pytraverse(skip)] + previous_info_callback: AtomicCell, } impl fmt::Debug for PySslSocket { @@ -2619,8 +2751,715 @@ mod _ssl { } } + #[cfg(all(unix, not(target_os = "aix")))] + struct IoScope(*const PySslSocket); + + #[cfg(all(unix, not(target_os = "aix")))] + impl IoScope { + fn enter(socket: &PySslSocket, vm: &VirtualMachine) -> PyResult { + let pointer = socket as *const _; + let entered = SSL_IO_CALLS.with(|calls| { + let mut calls = calls.borrow_mut(); + if calls.contains(&pointer) { + false + } else { + calls.push(pointer); + true + } + }); + // Exception allocation can run Python finalizers. Do not retain a + // RefCell borrow while they may attempt another SSL invocation. + if entered { + Ok(Self(pointer)) + } else { + Err(vm.new_runtime_error("Reentrant I/O on the same SSL connection")) + } + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + impl Drop for IoScope { + fn drop(&mut self) { + SSL_IO_CALLS.with(|calls| { + let pointer = calls.borrow_mut().pop(); + debug_assert_eq!(pointer, Some(self.0)); + }); + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + enum IoOperation<'a> { + Handshake, + Read(&'a mut [u8]), + Write(Option>), + Shutdown, + #[cfg(ossl111)] + PostHandshake, + } + + #[cfg(all(unix, not(target_os = "aix")))] + impl IoOperation<'_> { + fn timeout_message(&self, needs: Option) -> &'static str { + match self { + Self::Handshake => "The handshake operation timed out", + Self::Read(_) => "The read operation timed out", + Self::Write(_) => "The write operation timed out", + Self::Shutdown if needs == Some(Needs::Read) => "The read operation timed out", + Self::Shutdown => "The write operation timed out", + #[cfg(ossl111)] + Self::PostHandshake => "The handshake operation timed out", + } + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + enum IoOutput { + Done, + Count(usize), + Shutdown(ssl::ShutdownResult), + Unwrapped(Option>), + } + + #[cfg(all(unix, not(target_os = "aix")))] + enum IoStep { + Done(IoOutput), + // Preserve the pre-existing clean-read-EOF/PHA signal boundary. + Immediate(IoOutput), + Again, + Wait(Option, Needs), + } + + // Owned failures leave the native guard before exception allocation or + // attribute writes, either of which can execute Python (including GC). + #[cfg(all(unix, not(target_os = "aix")))] + enum IoError { + Eof, + Closed, + Timeout(&'static str), + WantWrite, + Message(&'static str), + Io(std::io::Error), + Callback(PyBaseExceptionRef), + Native(SslErrorSnapshot, Option), + #[cfg(ossl111)] + Stack(ErrorStack, Option), + } + + #[cfg(all(unix, not(target_os = "aix")))] + impl IoError { + fn into_pyexception(self, vm: &VirtualMachine) -> PyBaseExceptionRef { + match self { + Self::Eof => create_ssl_eof_error(vm).upcast(), + Self::Closed => socket_closed_error(vm), + Self::Timeout(message) => { + socket::timeout_error_msg(vm, message.to_owned()).upcast() + } + Self::WantWrite => create_ssl_want_write_error(vm).upcast(), + Self::Message(message) => new_ssl_error(vm, message), + Self::Io(error) => error.to_pyexception(vm), + Self::Callback(error) => error, + Self::Native(error, callback) => { + let error = error.into_pyexception(vm); + if let Some(callback) = callback { + error.set_context(Some(callback)); + } + error + } + #[cfg(ossl111)] + Self::Stack(error, callback) => { + let error = convert_openssl_error(vm, error); + if let Some(callback) = callback { + error.set_context(Some(callback)); + } + error + } + } + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + struct IoCall<'a> { + _ssl: &'a PySslSocket, + owner: Option, + owns_pending: bool, + last_needs: Option, + progress: Arc, + socket: Option<(PyRef, Arc, SocketDeadline)>, + waiter: Option>, + // Keep this installed through waits and signal callbacks, but never + // expose a stale native SSL pointer while its guard is released. + _scope: IoScope, + } + + #[cfg(all(unix, not(target_os = "aix")))] + impl<'a> IoCall<'a> { + fn new(ssl: &'a PySslSocket, writing: bool, vm: &VirtualMachine) -> PyResult { + let scope = IoScope::enter(ssl, vm)?; + let connection = ssl.connection.read(); + let progress = connection.progress.clone(); + let socket = connection.get_ref().map(|stream| { + ( + stream.socket.clone(), + stream.identity.clone(), + stream.timeout_deadline(), + ) + }); + Ok(Self { + _ssl: ssl, + owner: writing.then(WriteOwner::new), + owns_pending: false, + last_needs: None, + progress, + socket, + waiter: None, + _scope: scope, + }) + } + + fn prepare_wait( + &mut self, + observed: u64, + needs: Option, + error: IoError, + timeout_message: &'static str, + ) -> Result { + let Some((socket, identity, deadline)) = &self.socket else { + return Err(error); + }; + match deadline { + Err(false) => return Err(error), + Ok(deadline) if *deadline <= Instant::now() => { + return Err(IoError::Timeout(timeout_message)); + } + _ => {} + } + if self.waiter.is_none() { + self.waiter = Some(Arc::new(ProgressWaiter::new().map_err(IoError::Io)?)); + } + let waiter = self.waiter.as_ref().expect("waiter initialized"); + // Close does not need the SSL lock. Drain BEFORE checking identity + // and subscribing, or a concurrent close notification could be lost. + match waiter.drain() { + Ok(()) => {} + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => { + return Ok(IoStep::Again); + } + Err(error) => return Err(IoError::Io(error)), + } + let fd = socket + .sock_for_ssl_identity(identity) + .map_err(|_| IoError::Closed)? + .as_raw_fd(); + if !identity.subscribe(waiter.clone()) { + return Err(IoError::Closed); + } + self.progress.register(waiter, observed); + // Exclusive-writer contenders must not poll socket writability: it + // can remain ready while they are forbidden to execute native I/O. + Ok(IoStep::Wait( + needs.map(|_| fd), + needs.unwrap_or(Needs::Write), + )) + } + + fn wait( + &self, + fd: Option, + needs: Needs, + timeout_message: &'static str, + vm: &VirtualMachine, + ) -> Result<(), IoError> { + let waiter = self.waiter.as_ref().expect("registered waiter"); + let deadline = self + .socket + .as_ref() + .and_then(|(_, _, deadline)| deadline.as_ref().ok().copied()); + loop { + let result = vm.allow_threads(|| waiter.wait(fd, needs == Needs::Write, deadline)); + // No connection or socket lock is held while Python signals run. + vm.check_signals().map_err(IoError::Callback)?; + match result { + Ok(false) => return Ok(()), + Ok(true) => { + return Err(IoError::Timeout(timeout_message)); + } + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {} + Err(error) => { + if self + .socket + .as_ref() + .is_some_and(|(_, identity, _)| identity.is_closed()) + { + return Err(IoError::Closed); + } + return Err(IoError::Io(error)); + } + } + } + } + + fn abandon_pending(&mut self) { + if self.owns_pending { + self.owner.as_ref().expect("pending write owner").release(); + self.owns_pending = false; + self.progress.notify(); + } + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + impl Drop for IoCall<'_> { + fn drop(&mut self) { + // Cleanup must never acquire the SSL lock: a callback on another + // thread may be waiting for this invocation to finish. Publish an + // orphan before waking generation-checked registrations instead. + self.abandon_pending(); + if let Some(waiter) = &self.waiter { + self.progress.unregister(waiter); + if let Some((_, identity, _)) = &self.socket { + identity.unsubscribe(waiter.clone()); + } + } + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + impl PySslSocket { + fn install_progress_callback(&self) { + let connection = self.connection.write(); + let pointer = connection.ssl().as_ptr(); + unsafe { + self.previous_info_callback + .store(SSL_get_info_callback(pointer)); + SSL_set_info_callback(pointer, Some(ssl_info_progress)); + } + } + + fn drive(&self, mut operation: IoOperation<'_>, vm: &VirtualMachine) -> PyResult { + self.check_eof(vm)?; + let mut call = IoCall::new(self, matches!(operation, IoOperation::Write(_)), vm)?; + let mut queued = false; + let mut shutdown_zeros = 0; + let mut attempted = false; + let mut waited_for_ownership = false; + loop { + let step = (|| { + let mut connection = self.connection.write(); + if self.got_eof_error.load() { + return Err(IoError::Eof); + } + if connection.driver.terminal { + return Err(IoError::Message( + "TLS connection is unusable after a fatal error", + )); + } + if call + .socket + .as_ref() + .is_some_and(|(_, identity, _)| identity.is_closed()) + { + return Err(IoError::Closed); + } + // Buffered TLS data can satisfy even a very short positive + // timeout. Attempt native I/O once before timing retries. + if (attempted || waited_for_ownership) + && call.socket.as_ref().is_some_and(|(_, _, deadline)| { + deadline + .as_ref() + .is_ok_and(|deadline| *deadline <= Instant::now()) + }) + { + return Err(IoError::Timeout(operation.timeout_message(call.last_needs))); + } + // Abandonment is deliberately independent of the SSL lock. + // Observe its generation before checking the predicate. + let observed = call.progress.generation(); + let permit = match &mut operation { + IoOperation::Write(bytes) => connection.driver.write( + call.owner.as_ref().unwrap(), + bytes, + &mut queued, + ), + _ => connection.driver.other_io(), + }; + match permit { + Permit::RetryOriginalWrite => Err(IoError::WantWrite), + Permit::InvalidRetry => Err(IoError::Message( + "The pending SSL write must be retried with the same data in a new call", + )), + Permit::Wait => { + waited_for_ownership = true; + call.prepare_wait( + observed, + None, + IoError::WantWrite, + operation.timeout_message(call.last_needs), + ) + } + Permit::Ready => { + attempted = true; + call.owns_pending = call.owner.is_some(); + let before = ( + connection.io_progress(), + self.info_progress.load(), + connection.get_shutdown(), + ); + let permission_before = connection.driver.other_io(); + let pointer = connection.ssl().as_ptr(); + let result = { + let _native_scope = + HandshakeVmGuard::new(vm, &self.connection, pointer); + match &mut operation { + IoOperation::Handshake => vm + .allow_threads(|| connection.do_handshake()) + .map(|()| IoOutput::Done), + IoOperation::Read(buffer) => vm + .allow_threads(|| connection.ssl_read(buffer)) + .map(IoOutput::Count), + IoOperation::Write(_) => { + let SslConnection { + transport, driver, .. + } = &mut *connection; + vm.allow_threads(|| { + transport.ssl_write( + driver.bytes(call.owner.as_ref().unwrap()), + ) + }) + .map(IoOutput::Count) + } + IoOperation::Shutdown => { + if self.shutdown_seen_zero.load() { + // Leave plaintext after close_notify for unwrap(). + unsafe { + sys::SSL_ctrl( + pointer, + sys::SSL_CTRL_SET_READ_AHEAD, + 0, + core::ptr::null_mut(), + ); + } + } + vm.allow_threads(|| connection.shutdown()) + .map(IoOutput::Shutdown) + } + #[cfg(ossl111)] + IoOperation::PostHandshake => { + unsafe { sys::ERR_clear_error() }; + let result = vm.allow_threads(|| unsafe { + SSL_verify_client_post_handshake(pointer) + }); + let error = (result == 0).then(ErrorStack::get); + let callback = self.callback_error.lock().take(); + connection.notify_progress(); + return match (error, callback) { + (Some(error), callback) => { + Err(IoError::Stack(error, callback)) + } + (None, Some(callback)) => { + Err(IoError::Callback(callback)) + } + (None, None) => Ok(IoStep::Immediate(IoOutput::Done)), + }; + } + } + }; + let needs = + result.as_ref().err().and_then(|error| match error.code() { + ssl::ErrorCode::WANT_READ => Some(Needs::Read), + ssl::ErrorCode::WANT_WRITE => Some(Needs::Write), + _ => None, + }); + call.last_needs = needs; + let terminal = result.as_ref().err().is_some_and(|error| { + matches!( + error.code(), + ssl::ErrorCode::SSL | ssl::ErrorCode::SYSCALL + ) + }); + // Settle native ownership BEFORE a callback/signal can + // return an exception. Success must never be replayed. + if let Some(owner) = &call.owner { + call.owns_pending = connection.driver.settle_write(owner, needs); + } + if terminal { + connection.driver.fail(); + } + if result.as_ref().err().is_some_and(is_unexpected_eof) { + self.got_eof_error.store(true); + } + let after = ( + connection.io_progress(), + self.info_progress.load(), + connection.get_shutdown(), + ); + if before != after + || result.is_ok() + || terminal + || permission_before != connection.driver.other_io() + { + connection.notify_progress(); + } + let interrupted = connection.take_interrupted(); + // Snapshot exactly once, while this operation still + // owns native state. No unlocked helper drains it. + let callback = self.callback_error.lock().take(); + if let Some(callback) = callback { + // A fully blocking socket could not expose this + // intermediate WANT before the nonblocking driver. + // Timeout/nonblocking sockets and MemoryBIO retain + // native-error-first exception chaining. + if needs.is_some() + && call.socket.as_ref().is_some_and(|(_, _, deadline)| { + matches!(deadline, Err(true)) + }) + { + return Err(IoError::Callback(callback)); + } + return Err(match result { + Ok(_) => IoError::Callback(callback), + Err(error) => IoError::Native( + SslErrorSnapshot::capture(error, pointer), + Some(callback), + ), + }); + } + match result { + Ok(IoOutput::Shutdown(ssl::ShutdownResult::Sent)) => { + shutdown_zeros += 1; + self.shutdown_seen_zero.store(true); + if shutdown_zeros > 1 { + Ok(IoStep::Done(IoOutput::Unwrapped( + call.socket + .as_ref() + .map(|(socket, _, _)| socket.clone()), + ))) + } else { + Ok(IoStep::Again) + } + } + Ok(IoOutput::Shutdown(ssl::ShutdownResult::Received)) => { + Ok(IoStep::Done(IoOutput::Unwrapped( + call.socket.as_ref().map(|(socket, _, _)| socket.clone()), + ))) + } + Ok(output) => Ok(IoStep::Done(output)), + Err(error) => { + // Callback errors take precedence over these + // successful/EOF read reinterpretations above. + if matches!(operation, IoOperation::Read(_)) { + if error.code() == ssl::ErrorCode::ZERO_RETURN { + return Ok(IoStep::Immediate(IoOutput::Count(0))); + } + if error.code() == ssl::ErrorCode::WANT_READ + && connection.is_bio_eof() + { + self.got_eof_error.store(true); + connection.driver.fail(); + connection.notify_progress(); + return Err(IoError::Eof); + } + } + let error = IoError::Native( + SslErrorSnapshot::capture(error, pointer), + None, + ); + if let Some(needs) = needs { + if interrupted { + Ok(IoStep::Again) + } else { + call.prepare_wait( + observed, + Some(needs), + error, + operation.timeout_message(call.last_needs), + ) + } + } else { + Err(error) + } + } + } + } + } + })(); + // Allocation/attribute writes may run arbitrary Python. The + // SSL guard and callback pointer are gone, and a known-exiting + // write no longer blocks contenders as an active owner. + let step = match step { + Ok(IoStep::Immediate(output)) => return Ok(output), + Ok(step) => step, + Err(error) => { + call.abandon_pending(); + return Err(error.into_pyexception(vm)); + } + }; + if let Err(error) = vm.check_signals() { + call.abandon_pending(); + return Err(error); + } + match step { + IoStep::Immediate(_) => unreachable!("immediate result already returned"), + IoStep::Done(output) => return Ok(output), + IoStep::Again => {} + IoStep::Wait(fd, needs) => { + if let Err(error) = + call.wait(fd, needs, operation.timeout_message(call.last_needs), vm) + { + call.abandon_pending(); + return Err(error.into_pyexception(vm)); + } + } + } + } + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + type SslInfoCallback = + Option; + + #[cfg(all(unix, not(target_os = "aix")))] + unsafe extern "C" { + fn SSL_get_info_callback(ssl: *const sys::SSL) -> SslInfoCallback; + fn SSL_set_info_callback(ssl: *mut sys::SSL, callback: SslInfoCallback); + fn SSL_CTX_get_info_callback(ctx: *const sys::SSL_CTX) -> SslInfoCallback; + } + + #[cfg(all(unix, not(target_os = "aix")))] + unsafe extern "C" fn ssl_info_progress( + ssl: *const sys::SSL, + where_: core::ffi::c_int, + value: core::ffi::c_int, + ) { + // These public callback bits are present in OpenSSL 1.1.1 as well as 3.x. + const SSL_CB_LOOP: core::ffi::c_int = 0x01; + const SSL_CB_HANDSHAKE_DONE: core::ffi::c_int = 0x20; + const SSL_CB_ALERT: core::ffi::c_int = 0x4000; + unsafe { + let socket = + sys::SSL_get_ex_data(ssl, get_socket_ex_data_index()) as *const Py; + if socket.is_null() { + return; + } + let socket = &*socket; + if where_ & (SSL_CB_LOOP | SSL_CB_HANDSHAKE_DONE | SSL_CB_ALERT) != 0 { + socket + .info_progress + .store(socket.info_progress.load().wrapping_add(1)); + } + // A NULL per-SSL callback delegates to the CURRENT context, including + // a context installed by SNI. Preserve that fallback dynamically. + let callback = socket + .previous_info_callback + .load() + .or_else(|| SSL_CTX_get_info_callback(sys::SSL_get_SSL_CTX(ssl))); + if let Some(callback) = callback { + callback(ssl, where_, value); + } + } + } + #[pyclass(flags(IMMUTABLETYPE))] impl PySslSocket { + #[cfg(all(unix, not(target_os = "aix")))] + #[pymethod] + fn do_handshake(zelf: &Py, vm: &VirtualMachine) -> PyResult<()> { + zelf.drive(IoOperation::Handshake, vm).map(|_| ()) + } + + #[cfg(all(unix, not(target_os = "aix")))] + #[pymethod] + fn shutdown(zelf: &Py, vm: &VirtualMachine) -> PyResult>> { + // Do not reacquire the connection after completion: another native + // callback may already be waiting for this Python call to return. + match zelf.drive(IoOperation::Shutdown, vm)? { + IoOutput::Unwrapped(socket) => Ok(socket), + _ => unreachable!("shutdown returns its underlying socket"), + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + #[pymethod] + fn verify_client_post_handshake(zelf: &Py, vm: &VirtualMachine) -> PyResult<()> { + cfg_select! { + ossl111 => zelf.drive(IoOperation::PostHandshake, vm).map(|_| ()), + _ => Err(vm.new_not_implemented_error("Post-handshake auth is not supported by your OpenSSL version.")), + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + #[pymethod] + fn write(zelf: &Py, data: ArgBytesLike, vm: &VirtualMachine) -> PyResult { + zelf.check_eof(vm)?; + // OpenSSL retains the write across WANT, including after this + // Python invocation returns. Own the bytes, not an exporter lock. + let bytes = { + let source = data.borrow_buf(); + let mut owned = Vec::new(); + owned + .try_reserve_exact(source.len()) + .map_err(|_| vm.no_memory_error())?; + owned.extend_from_slice(&source); + owned.into_boxed_slice() + }; + match zelf.drive(IoOperation::Write(Some(bytes)), vm)? { + IoOutput::Count(count) => Ok(count), + _ => unreachable!("write returns a byte count"), + } + } + + #[cfg(all(unix, not(target_os = "aix")))] + #[pymethod] + fn read( + zelf: &Py, + n: isize, + buffer: OptionalArg, + vm: &VirtualMachine, + ) -> PyResult { + let read_len = match &buffer { + OptionalArg::Present(buffer) => { + let len = buffer.borrow_buf_mut().len(); + if n <= 0 || n as usize > len { + len + } else { + n as usize + } + } + OptionalArg::Missing => { + if n < 0 { + return Err(vm.new_value_error("size should not be negative")); + } + n as usize + } + }; + zelf.check_eof(vm)?; + // As in CPython, a zero-length read need not enter native SSL. + if read_len == 0 { + return Ok(if buffer.is_present() { + vm.ctx.new_int(0).into() + } else { + vm.ctx.new_bytes(vec![]).into() + }); + } + // Do not keep the caller's buffer locked while waiting for a peer. + let mut scratch = vm.new_zeroed_bytes(read_len)?; + let count = match zelf.drive(IoOperation::Read(&mut scratch), vm)? { + IoOutput::Count(count) => count, + _ => unreachable!("read returns a byte count"), + }; + Ok(match &buffer { + OptionalArg::Present(buffer) => { + buffer.borrow_buf_mut()[..count].copy_from_slice(&scratch[..count]); + vm.ctx.new_int(count).into() + } + OptionalArg::Missing => { + scratch.truncate(count); + scratch.shrink_to_fit(); + vm.ctx.new_bytes(scratch).into() + } + }) + } + fn callback_owner(zelf: &Py) -> Option { if let Some(owner) = zelf.owner.read().as_ref() { return owner.upgrade(); @@ -2637,23 +3476,32 @@ mod _ssl { } } - fn ssl_error(&self, error: ssl::Error, vm: &VirtualMachine) -> PyBaseExceptionRef { + #[cfg(not(all(unix, not(target_os = "aix"))))] + fn ssl_error( + &self, + error: ssl::Error, + ssl_ptr: *const sys::SSL, + vm: &VirtualMachine, + ) -> PyBaseExceptionRef { if is_unexpected_eof(&error) { self.got_eof_error.store(true); } - convert_ssl_error(vm, error) + SslErrorSnapshot::capture(error, ssl_ptr).into_pyexception(vm) } + #[cfg(not(all(unix, not(target_os = "aix"))))] fn finish_ssl_operation( &self, result: Result, + ssl_ptr: *const sys::SSL, vm: &VirtualMachine, ) -> PyResult> { - if let Some(callback_error) = self.callback_error.lock().take() { + let callback_error = self.callback_error.lock().take(); + if let Some(callback_error) = callback_error { return Err(match result { Ok(_) => callback_error, Err(error) => { - let error = self.ssl_error(error, vm); + let error = self.ssl_error(error, ssl_ptr, vm); error.set_context(Some(callback_error)); error } @@ -3044,6 +3892,7 @@ mod _ssl { }) } + #[cfg(not(all(unix, not(target_os = "aix"))))] #[pymethod] fn verify_client_post_handshake(zelf: &Py, vm: &VirtualMachine) -> PyResult<()> { cfg_select! { @@ -3063,6 +3912,7 @@ mod _ssl { } } + #[cfg(not(all(unix, not(target_os = "aix"))))] #[pymethod] fn shutdown(zelf: &Py, vm: &VirtualMachine) -> PyResult>> { let mut stream = zelf.connection.write(); @@ -3082,21 +3932,24 @@ mod _ssl { ); } } - let error = - match zelf.finish_ssl_operation(vm.allow_threads(|| stream.shutdown()), vm)? { - Ok(ssl::ShutdownResult::Received) => break, - Ok(ssl::ShutdownResult::Sent) => { - zeros += 1; - if zeros > 1 { - break; - } - zelf.shutdown_seen_zero.store(true); - // Retry immediately: the peer's notification may already - // be buffered, even when the transport is not readable. - continue; + let error = match zelf.finish_ssl_operation( + vm.allow_threads(|| stream.shutdown()), + ssl_ptr, + vm, + )? { + Ok(ssl::ShutdownResult::Received) => break, + Ok(ssl::ShutdownResult::Sent) => { + zeros += 1; + if zeros > 1 { + break; } - Err(error) => error, - }; + zelf.shutdown_seen_zero.store(true); + // Retry immediately: the peer's notification may already + // be buffered, even when the transport is not readable. + continue; + } + Err(error) => error, + }; if let Some(socket) = stream.get_ref() { let (needs, state) = socket.socket_needs( &error, @@ -3117,9 +3970,9 @@ mod _ssl { _ => {} } } - return Err(zelf.ssl_error(error, vm)); + return Err(zelf.ssl_error(error, ssl_ptr, vm)); } - Ok(stream.get_ref().map(|socket| socket.0.clone())) + Ok(stream.get_ref().map(|socket| socket.socket.clone())) } #[cfg(osslconf = "OPENSSL_NO_COMP")] @@ -3144,6 +3997,7 @@ mod _ssl { }) } + #[cfg(not(all(unix, not(target_os = "aix"))))] #[pymethod] fn do_handshake(zelf: &Py, vm: &VirtualMachine) -> PyResult<()> { zelf.check_eof(vm)?; @@ -3157,15 +4011,8 @@ mod _ssl { // BIO mode: no timeout/select logic, just do handshake if stream.is_bio() { let result = zelf - .finish_ssl_operation(vm.allow_threads(|| stream.do_handshake()), vm)? - .map_err(|e| { - let exc = zelf.ssl_error(e, vm); - // If it's a cert verification error, set verify info - if exc.class().is(PySSLCertVerificationError::class(&vm.ctx)) { - set_verify_error_info(&exc, ssl_ptr, vm); - } - exc - }); + .finish_ssl_operation(vm.allow_threads(|| stream.do_handshake()), ssl_ptr, vm)? + .map_err(|e| zelf.ssl_error(e, ssl_ptr, vm)); return result; } @@ -3181,9 +4028,11 @@ mod _ssl { // same reason. The connection lock stays held across it, which // is why it is a detaching lock: a thread reaching the same // socket gives up its interpreter rather than wait attached. - let err = match zelf - .finish_ssl_operation(vm.allow_threads(|| stream.do_handshake()), vm)? - { + let err = match zelf.finish_ssl_operation( + vm.allow_threads(|| stream.do_handshake()), + ssl_ptr, + vm, + )? { Ok(()) => { return Ok(()); } @@ -3213,28 +4062,25 @@ mod _ssl { } } } - let exc = zelf.ssl_error(err, vm); - // If it's a cert verification error, set verify info - if exc.class().is(PySSLCertVerificationError::class(&vm.ctx)) { - set_verify_error_info(&exc, ssl_ptr, vm); - } - return Err(exc); + return Err(zelf.ssl_error(err, ssl_ptr, vm)); } } + #[cfg(not(all(unix, not(target_os = "aix"))))] #[pymethod] fn write(zelf: &Py, data: ArgBytesLike, vm: &VirtualMachine) -> PyResult { zelf.check_eof(vm)?; let mut stream = zelf.connection.write(); - let _vm_guard = HandshakeVmGuard::new(vm, &zelf.connection, stream.ssl().as_ptr()); + let ssl_ptr = stream.ssl().as_ptr(); + let _vm_guard = HandshakeVmGuard::new(vm, &zelf.connection, ssl_ptr); let data = data.borrow_buf(); let data = &*data; // BIO mode: no timeout/select logic if stream.is_bio() { return zelf - .finish_ssl_operation(vm.allow_threads(|| stream.ssl_write(data)), vm)? - .map_err(|e| zelf.ssl_error(e, vm)); + .finish_ssl_operation(vm.allow_threads(|| stream.ssl_write(data)), ssl_ptr, vm)? + .map_err(|e| zelf.ssl_error(e, ssl_ptr, vm)); } // Socket mode: handle timeout and blocking @@ -3258,9 +4104,11 @@ mod _ssl { // Sending waits for the peer to make room, which it need not // ever do; `SSL_write_ex` is released around for the same // reason. - let err = match zelf - .finish_ssl_operation(vm.allow_threads(|| stream.ssl_write(data)), vm)? - { + let err = match zelf.finish_ssl_operation( + vm.allow_threads(|| stream.ssl_write(data)), + ssl_ptr, + vm, + )? { Ok(len) => return Ok(len), Err(e) => e, }; @@ -3286,7 +4134,7 @@ mod _ssl { } } } - return Err(zelf.ssl_error(err, vm)); + return Err(zelf.ssl_error(err, ssl_ptr, vm)); } } @@ -3347,6 +4195,7 @@ mod _ssl { sys::SSL_session_reused(ssl_ptr) != 0 }) } + #[cfg(not(all(unix, not(target_os = "aix"))))] #[pymethod] fn read( zelf: &Py, @@ -3387,7 +4236,8 @@ mod _ssl { } let mut stream = zelf.connection.write(); - let _vm_guard = HandshakeVmGuard::new(vm, &zelf.connection, stream.ssl().as_ptr()); + let ssl_ptr = stream.ssl().as_ptr(); + let _vm_guard = HandshakeVmGuard::new(vm, &zelf.connection, ssl_ptr); // The read below answers when the peer writes, which may be never, // and reaching the caller's buffer takes a lock that every other // thread touching the same object waits on. Read aside and take @@ -3397,7 +4247,11 @@ mod _ssl { // BIO mode: no timeout/select logic let count = if stream.is_bio() { - match zelf.finish_ssl_operation(vm.allow_threads(|| stream.ssl_read(buf)), vm)? { + match zelf.finish_ssl_operation( + vm.allow_threads(|| stream.ssl_read(buf)), + ssl_ptr, + vm, + )? { Ok(count) => count, Err(e) => { // close_notify is an orderly TLS EOF. @@ -3414,7 +4268,7 @@ mod _ssl { zelf.got_eof_error.store(true); return Err(create_ssl_eof_error(vm).upcast()); } - return Err(zelf.ssl_error(e, vm)); + return Err(zelf.ssl_error(e, ssl_ptr, vm)); } } } else { @@ -3427,9 +4281,11 @@ mod _ssl { // This is the wait the whole method is shaped around: it // ends when the peer writes. `SSL_read_ex` is released // around for the same reason. - let err = match zelf - .finish_ssl_operation(vm.allow_threads(|| stream.ssl_read(buf)), vm)? - { + let err = match zelf.finish_ssl_operation( + vm.allow_threads(|| stream.ssl_read(buf)), + ssl_ptr, + vm, + )? { Ok(count) => break count, Err(e) => e, }; @@ -3460,7 +4316,7 @@ mod _ssl { } } } - return Err(zelf.ssl_error(err, vm)); + return Err(zelf.ssl_error(err, ssl_ptr, vm)); } }; let ret = match &buffer { @@ -4126,31 +4982,63 @@ mod _ssl { } } - // Helper function to set verify_code and verify_message on SSLCertVerificationError - fn set_verify_error_info( - exc: &Py, - ssl_ptr: *const sys::SSL, - vm: &VirtualMachine, - ) { - // Get verify result - let verify_code = unsafe { sys::SSL_get_verify_result(ssl_ptr) }; - let verify_code_obj = vm.ctx.new_int(verify_code); - - // Get verify message - let verify_message = unsafe { - let verify_str = sys::X509_verify_cert_error_string(verify_code); - if verify_str.is_null() { - vm.ctx.none() - } else { - let c_str = core::ffi::CStr::from_ptr(verify_str); - vm.ctx.new_str(c_str.to_string_lossy()).into() - } - }; - - let exc_obj = exc.as_object(); - let _ = exc_obj.set_attr("verify_code", verify_code_obj, vm); - let _ = exc_obj.set_attr("verify_message", verify_message, vm); + struct VerifyErrorInfo { + code: core::ffi::c_long, + message: Option, } + + // Only owned metadata may outlive the native SSL guard. In particular, + // exception normalization must never consult a later operation's SSL state. + struct SslErrorSnapshot { + error: ssl::Error, + verify: Option, + } + + impl SslErrorSnapshot { + fn capture(error: ssl::Error, ssl_ptr: *const sys::SSL) -> Self { + let is_cert_verify = + matches!(error.code(), ssl::ErrorCode::SSL | ssl::ErrorCode::SYSCALL) + && error.ssl_error().is_some_and(|stack| { + stack.errors().last().is_some_and(|error| { + sys::ERR_GET_LIB(error.code()) == 20 + && sys::ERR_GET_REASON(error.code()) == 134 + }) + }); + let verify = is_cert_verify.then(|| { + // The caller holds this connection's native guard throughout + // capture. Neither the pointer nor borrowed C data is retained. + let code = unsafe { sys::SSL_get_verify_result(ssl_ptr) }; + let message = unsafe { + let pointer = sys::X509_verify_cert_error_string(code); + (!pointer.is_null()).then(|| { + core::ffi::CStr::from_ptr(pointer) + .to_string_lossy() + .into_owned() + }) + }; + VerifyErrorInfo { code, message } + }); + Self { error, verify } + } + + fn into_pyexception(self, vm: &VirtualMachine) -> PyBaseExceptionRef { + let exception = convert_ssl_error(vm, self.error); + if let Some(verify) = self.verify { + let _ = + exception + .as_object() + .set_attr("verify_code", vm.ctx.new_int(verify.code), vm); + let message = verify + .message + .map_or_else(|| vm.ctx.none(), |message| vm.ctx.new_str(message).into()); + let _ = exception + .as_object() + .set_attr("verify_message", message, vm); + } + exception + } + } + fn is_unexpected_eof(error: &ssl::Error) -> bool { (error.code() == ssl::ErrorCode::SYSCALL && error.io_error().is_none() @@ -4277,21 +5165,70 @@ mod _ssl { } } + #[cfg(all(unix, not(target_os = "aix")))] + impl SocketStream { + fn finish_transport(&mut self, result: std::io::Result) -> std::io::Result { + match result { + Ok(count) => { + self.progress = self.progress.wrapping_add(count as u64); + Ok(count) + } + Err(error) if error.kind() == std::io::ErrorKind::Interrupted => { + // openssl's custom BIO treats only WouldBlock/NotConnected + // as retryable. Passing EINTR through would become a fatal + // SSL_ERROR_SYSCALL. Check Python signals after settlement. + self.interrupted = true; + Err(std::io::Error::from(std::io::ErrorKind::WouldBlock)) + } + Err(error) => Err(error), + } + } + } + impl Read for SocketStream { fn read(&mut self, buf: &mut [u8]) -> std::io::Result { - let mut socket: &PySocket = &self.0; - socket.read(buf) + #[cfg(all(unix, not(target_os = "aix")))] + { + let result = { + let socket = self.socket.sock_for_ssl_identity(&self.identity)?; + // recv initializes only the reported prefix; this caller's + // buffer was already initialized and is valid throughout. + let buffer = unsafe { + core::slice::from_raw_parts_mut( + buf.as_mut_ptr().cast::>(), + buf.len(), + ) + }; + socket.recv_with_flags(buffer, libc::MSG_DONTWAIT) + }; + self.finish_transport(result) + } + #[cfg(not(all(unix, not(target_os = "aix"))))] + { + let mut socket: &PySocket = &self.socket; + socket.read(buf) + } } } impl Write for SocketStream { fn write(&mut self, buf: &[u8]) -> std::io::Result { - let mut socket: &PySocket = &self.0; - socket.write(buf) + #[cfg(all(unix, not(target_os = "aix")))] + { + let result = { + let socket = self.socket.sock_for_ssl_identity(&self.identity)?; + socket.send_with_flags(buf, libc::MSG_DONTWAIT) + }; + self.finish_transport(result) + } + #[cfg(not(all(unix, not(target_os = "aix"))))] + { + let mut socket: &PySocket = &self.socket; + socket.write(buf) + } } fn flush(&mut self) -> std::io::Result<()> { - let mut socket: &PySocket = &self.0; - socket.flush() + Ok(()) } } diff --git a/crates/stdlib/src/openssl/driver.rs b/crates/stdlib/src/openssl/driver.rs new file mode 100644 index 000000000..b1ff3a4f7 --- /dev/null +++ b/crates/stdlib/src/openssl/driver.rs @@ -0,0 +1,264 @@ +//! Ownership rules for nonblocking OpenSSL retries. +//! +//! In particular, SSL_write returning WANT_WRITE forbids *any* other I/O on +//! that SSL until the same write is retried. Its buffer must survive Python +//! timeout, signal and nonblocking returns as well as ordinary OS waits. +use alloc::sync::Arc; +use core::sync::atomic::{AtomicBool, Ordering}; + +#[derive(Clone, Debug)] +pub(super) struct WriteOwner(Arc); + +impl WriteOwner { + pub(super) fn new() -> Self { + Self(Arc::new(AtomicBool::new(true))) + } + + pub(super) fn release(&self) { + self.0.store(false, Ordering::Release); + } + + fn is_active(&self) -> bool { + self.0.load(Ordering::Acquire) + } + + fn is(&self, other: &Self) -> bool { + Arc::ptr_eq(&self.0, &other.0) + } +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Needs { + Read, + Write, +} + +#[derive(Debug)] +struct PendingWrite { + bytes: Box<[u8]>, + owner: WriteOwner, + needs: Option, +} + +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub(super) enum Permit { + Ready, + Wait, + RetryOriginalWrite, + InvalidRetry, +} + +#[derive(Default, Debug)] +pub(super) struct Driver { + pending: Option, + pub(super) terminal: bool, +} + +impl Driver { + pub(super) fn other_io(&self) -> Permit { + match &self.pending { + Some(pending) if pending.needs == Some(Needs::Write) => { + if pending.owner.is_active() { + Permit::Wait + } else { + Permit::RetryOriginalWrite + } + } + _ => Permit::Ready, + } + } + + pub(super) fn write( + &mut self, + owner: &WriteOwner, + bytes: &mut Option>, + queued: &mut bool, + ) -> Permit { + // An invocation can fail before entering native SSL. There is no + // continuation in that case, so its abandoned buffer is disposable. + if self + .pending + .as_ref() + .is_some_and(|pending| pending.needs.is_none() && !pending.owner.is_active()) + { + self.pending = None; + } + if let Some(pending) = &mut self.pending { + if pending.owner.is_active() { + if pending.owner.is(owner) { + return Permit::Ready; + } + *queued = true; + return Permit::Wait; + } + // A write already queued behind another call is a distinct write, + // even when its contents happen to match the abandoned operation. + if *queued || bytes.as_deref() != Some(pending.bytes.as_ref()) { + return Permit::InvalidRetry; + } + pending.owner = owner.clone(); + // Retain the original allocation, not the new invocation's copy. + *bytes = None; + return Permit::Ready; + } + self.pending = Some(PendingWrite { + bytes: bytes.take().expect("new write owns its buffer"), + owner: owner.clone(), + needs: None, + }); + Permit::Ready + } + + pub(super) fn bytes(&self, owner: &WriteOwner) -> &[u8] { + let pending = self.pending.as_ref().expect("active write"); + assert!(pending.owner.is(owner) && pending.owner.is_active()); + &pending.bytes + } + + pub(super) fn settle_write(&mut self, owner: &WriteOwner, needs: Option) -> bool { + let pending = self.pending.as_mut().expect("active write"); + assert!(pending.owner.is(owner) && pending.owner.is_active()); + if let Some(needs) = needs { + pending.needs = Some(needs); + } else { + self.pending = None; + } + needs.is_some() + } + + pub(super) fn fail(&mut self) { + self.terminal = true; + self.pending = None; + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn payload() -> Box<[u8]> { + Box::from(&b"stable bytes"[..]) + } + + #[test] + fn want_write_is_exclusive_and_keeps_original_allocation() { + let mut driver = Driver::default(); + let owner = WriteOwner::new(); + let mut bytes = Some(payload()); + let ptr = bytes.as_ref().unwrap().as_ptr(); + assert_eq!(driver.write(&owner, &mut bytes, &mut false), Permit::Ready); + driver.settle_write(&owner, Some(Needs::Write)); + assert_eq!(driver.other_io(), Permit::Wait); + assert_eq!(driver.bytes(&owner).as_ptr(), ptr); + owner.release(); + assert_eq!(driver.other_io(), Permit::RetryOriginalWrite); + let next = WriteOwner::new(); + assert_eq!( + driver.write(&next, &mut Some(payload()), &mut false), + Permit::Ready + ); + assert_eq!(driver.bytes(&next).as_ptr(), ptr); + assert!( + !driver.settle_write(&next, None), + "completion disarms cleanup" + ); + assert_eq!(driver.other_io(), Permit::Ready); + } + + #[test] + fn want_read_allows_progress_but_not_a_different_write() { + let mut driver = Driver::default(); + let owner = WriteOwner::new(); + driver.write(&owner, &mut Some(payload()), &mut false); + driver.settle_write(&owner, Some(Needs::Read)); + assert_eq!(driver.other_io(), Permit::Ready); + owner.release(); + assert_eq!(driver.other_io(), Permit::Ready); + let other = WriteOwner::new(); + let mut different = Some(Box::from(&b"different"[..])); + assert_eq!( + driver.write(&other, &mut different, &mut false), + Permit::InvalidRetry + ); + assert_eq!( + driver.write(&other, &mut Some(payload()), &mut false), + Permit::Ready + ); + assert_eq!(driver.bytes(&other), b"stable bytes"); + } + + #[test] + fn concurrent_equal_writes_are_distinct() { + let mut driver = Driver::default(); + let first = WriteOwner::new(); + let second = WriteOwner::new(); + driver.write(&first, &mut Some(payload()), &mut false); + driver.settle_write(&first, Some(Needs::Write)); + let mut bytes = Some(payload()); + let ptr = bytes.as_ref().unwrap().as_ptr(); + let mut queued = false; + assert_eq!(driver.write(&second, &mut bytes, &mut queued), Permit::Wait); + driver.settle_write(&first, None); + assert_eq!( + driver.write(&second, &mut bytes, &mut queued), + Permit::Ready + ); + assert_eq!(driver.bytes(&second).as_ptr(), ptr); + } + + #[test] + fn queued_writer_cannot_acknowledge_an_abandoned_write() { + let mut driver = Driver::default(); + let first = WriteOwner::new(); + let second = WriteOwner::new(); + driver.write(&first, &mut Some(payload()), &mut false); + driver.settle_write(&first, Some(Needs::Read)); + let mut bytes = Some(payload()); + let mut queued = false; + assert_eq!(driver.write(&second, &mut bytes, &mut queued), Permit::Wait); + first.release(); + assert_eq!( + driver.write(&second, &mut bytes, &mut queued), + Permit::InvalidRetry + ); + assert!(driver.pending.is_some()); + } + + #[test] + fn stale_release_cannot_orphan_an_adopted_writer() { + let mut driver = Driver::default(); + let first = WriteOwner::new(); + driver.write(&first, &mut Some(payload()), &mut false); + assert!(driver.settle_write(&first, Some(Needs::Write))); + first.release(); + let next = WriteOwner::new(); + assert_eq!( + driver.write(&next, &mut Some(payload()), &mut false), + Permit::Ready + ); + first.release(); + assert_eq!(driver.other_io(), Permit::Wait); + assert_eq!(driver.bytes(&next), b"stable bytes"); + assert!(!driver.settle_write(&next, None)); + } + + #[test] + fn release_before_attempt_and_terminal_error_leave_no_retry() { + let mut driver = Driver::default(); + let owner = WriteOwner::new(); + driver.write(&owner, &mut Some(payload()), &mut false); + owner.release(); + let owner = WriteOwner::new(); + let mut different = Some(Box::from(&b"new operation"[..])); + assert_eq!( + driver.write(&owner, &mut different, &mut false), + Permit::Ready + ); + assert_eq!(driver.bytes(&owner), b"new operation"); + driver.settle_write(&owner, Some(Needs::Write)); + driver.fail(); + assert!(driver.terminal); + owner.release(); + assert!(driver.pending.is_none()); + } +} diff --git a/crates/stdlib/src/openssl/waiter.rs b/crates/stdlib/src/openssl/waiter.rs new file mode 100644 index 000000000..fa58e33e2 --- /dev/null +++ b/crates/stdlib/src/openssl/waiter.rs @@ -0,0 +1,285 @@ +use alloc::sync::{Arc, Weak}; +use rustpython_host_env::{ + io as host_io, + select::{self, POLLIN, POLLNVAL, POLLOUT, PollFd}, +}; +use std::{ + io, + os::fd::{AsFd, AsRawFd, OwnedFd, RawFd}, + time::Instant, +}; + +/// A persistent, native-only notification for one blocked SSL operation. +#[derive(Debug)] +pub(super) struct ProgressWaiter { + read_fd: OwnedFd, + write_fd: OwnedFd, +} + +impl ProgressWaiter { + pub(super) fn new() -> io::Result { + let (read_fd, write_fd) = wake_pipe()?; + Ok(Self { read_fd, write_fd }) + } + + pub(super) fn notify(&self) { + loop { + match host_io::write_once(self.write_fd.as_fd().into(), &[1]) { + Ok(_) => return, + Err(error) if error.kind() == io::ErrorKind::Interrupted => continue, + // A full pipe already carries a persistent notification. + Err(error) if error.kind() == io::ErrorKind::WouldBlock => return, + // Both endpoints stay owned for this entire call, so neither + // EBADF nor EPIPE is possible without external fd corruption. + Err(_) => return, + } + } + } + + /// Re-arm under the connection guard, before the final identity check and + /// close-observer subscription. Draining after subscribing can lose close. + pub(super) fn drain(&self) -> io::Result<()> { + let mut buffer = [0; 256]; + loop { + match host_io::read_once(self.read_fd.as_fd().into(), &mut buffer) { + Ok(0) => { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "SSL progress notification pipe closed", + )); + } + Ok(_) => {} + Err(error) if error.kind() == io::ErrorKind::WouldBlock => return Ok(()), + // In particular, return EINTR so the caller can check signals. + Err(error) => return Err(error), + } + } + } + + /// Wait without a connection or socket guard. A missing socket means that + /// write ownership permits only progress notifications to resume this call. + /// Returns true on timeout; readiness never drains the notification pipe. + pub(super) fn wait( + &self, + socket_fd: Option, + write: bool, + deadline: Option, + ) -> io::Result { + if socket_fd.is_some_and(|fd| fd < 0) { + return Err(io::Error::from_raw_os_error(libc::EBADF)); + } + let mut fds = [ + PollFd { + fd: self.read_fd.as_raw_fd(), + events: POLLIN, + revents: 0, + }, + PollFd { + // poll ignores a negative fd when only progress is allowed. + fd: socket_fd.unwrap_or(-1), + // TLS consumes normal stream data, never MSG_OOB. Watching + // POLLPRI could spin on urgent data that this BIO cannot read. + events: if write { POLLOUT } else { POLLIN }, + revents: 0, + }, + ]; + loop { + let (timeout, capped) = match deadline { + None => (-1, false), + Some(deadline) => { + let remaining = deadline.saturating_duration_since(Instant::now()); + match select::duration_as_millis_ceiling(remaining) { + Some(timeout) => (timeout, false), + None => (i32::MAX, true), + } + } + }; + // Even an expired deadline gets a zero-time poll, as in wait_fd. + // EINTR is deliberately propagated for Python signal checking. + match select::poll_fds(&mut fds, timeout)? { + 0 if capped => {} + 0 => return Ok(true), + _ if fds.iter().any(|fd| fd.revents & POLLNVAL != 0) => { + return Err(io::Error::from_raw_os_error(libc::EBADF)); + } + _ => return Ok(false), + } + } + } +} + +impl crate::socket::SocketCloseObserver for ProgressWaiter { + fn socket_closed(&self) { + self.notify(); + } +} + +fn wake_pipe() -> io::Result<(OwnedFd, OwnedFd)> { + cfg_select! { + any( + target_os = "dragonfly", + target_os = "freebsd", + target_os = "linux", + target_os = "android", + target_os = "netbsd", + target_os = "openbsd" + ) => { + rustpython_host_env::posix::pipe2(libc::O_CLOEXEC | libc::O_NONBLOCK) + } + target_os = "redox" => { + use std::os::fd::FromRawFd; + + // Redox provides pipe2, but the host helper excludes this target. + let mut fds = [-1; 2]; + // SAFETY: fds has storage for the two descriptors pipe2 writes. + if unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_CLOEXEC | libc::O_NONBLOCK) } < 0 { + return Err(io::Error::last_os_error()); + } + // SAFETY: successful pipe2 transfers ownership of two new fds. + Ok(unsafe { (OwnedFd::from_raw_fd(fds[0]), OwnedFd::from_raw_fd(fds[1])) }) + } + _ => { + // The host helper sets CLOEXEC and owns both fds immediately, so + // any subsequent fcntl failure also closes both endpoints. + let (read_fd, write_fd) = rustpython_host_env::posix::pipe()?; + rustpython_host_env::fcntl::set_blocking(read_fd.as_fd(), false)?; + rustpython_host_env::fcntl::set_blocking(write_fd.as_fd(), false)?; + Ok((read_fd, write_fd)) + } + } +} + +/// Implementations are native-only and must not reenter the registry. +/// Production uses only ProgressWaiter's nonblocking pipe notification. +pub(super) trait ProgressNotify { + fn notify(&self); +} + +impl ProgressNotify for ProgressWaiter { + fn notify(&self) { + self.notify(); + } +} + +struct Registrations { + generation: u64, + waiters: Vec>, +} + +/// Ownership can be abandoned without acquiring the native SSL lock. Compare +/// generations when registering so that abandonment between predicate-check +/// and registration cannot strand a waiter. +pub(super) struct ProgressRegistry { + state: std::sync::Mutex>, +} + +impl Default for ProgressRegistry { + fn default() -> Self { + Self { + state: std::sync::Mutex::new(Registrations { + generation: 0, + waiters: Vec::new(), + }), + } + } +} + +impl ProgressRegistry { + pub(super) fn generation(&self) -> u64 { + self.state.lock().unwrap().generation + } + + pub(super) fn register(&self, waiter: &Arc, observed: u64) { + let mut state = self.state.lock().unwrap(); + let weak = Arc::downgrade(waiter); + state.waiters.retain(|waiter| waiter.strong_count() != 0); + if !state + .waiters + .iter() + .any(|registered| registered.ptr_eq(&weak)) + { + state.waiters.push(weak); + } + // Comparison and queuing share the notify mutex. A concurrent release + // either changed this generation or will find this registered waiter. + if state.generation != observed { + waiter.notify(); + } + } + + pub(super) fn unregister(&self, waiter: &Arc) { + let weak = Arc::downgrade(waiter); + self.state + .lock() + .unwrap() + .waiters + .retain(|registered| !registered.ptr_eq(&weak)); + } + + pub(super) fn notify(&self) { + let mut state = self.state.lock().unwrap(); + state.generation = state.generation.wrapping_add(1); + state.waiters.retain(|waiter| { + if let Some(waiter) = waiter.upgrade() { + waiter.notify(); + true + } else { + false + } + }); + } +} + +#[cfg(test)] +mod tests { + use super::*; + use core::sync::atomic::{AtomicUsize, Ordering}; + + #[derive(Default)] + struct Counter(AtomicUsize); + + impl ProgressNotify for Counter { + fn notify(&self) { + self.0.fetch_add(1, Ordering::Relaxed); + } + } + + #[test] + fn release_before_registration_is_not_lost() { + let registry = ProgressRegistry::default(); + let observed = registry.generation(); + let waiter = Arc::new(Counter::default()); + registry.notify(); + registry.register(&waiter, observed); + assert_eq!(waiter.0.load(Ordering::Relaxed), 1); + } + + #[test] + fn release_after_registration_notifies_and_unregister_removes_it() { + let registry = ProgressRegistry::default(); + let waiter = Arc::new(Counter::default()); + registry.register(&waiter, registry.generation()); + assert_eq!(waiter.0.load(Ordering::Relaxed), 0); + registry.notify(); + assert_eq!(waiter.0.load(Ordering::Relaxed), 1); + registry.unregister(&waiter); + registry.notify(); + assert_eq!(waiter.0.load(Ordering::Relaxed), 1); + } + + #[test] + fn unchanged_retry_does_not_self_wake_or_duplicate_registration() { + let registry = ProgressRegistry::default(); + let waiter = Arc::new(Counter::default()); + for _ in 0..10 { + registry.register(&waiter, registry.generation()); + } + assert_eq!(waiter.0.load(Ordering::Relaxed), 0); + assert_eq!(registry.state.lock().unwrap().waiters.len(), 1); + registry.notify(); + assert_eq!(waiter.0.load(Ordering::Relaxed), 1); + let fresh = registry.generation(); + registry.register(&waiter, fresh); + assert_eq!(waiter.0.load(Ordering::Relaxed), 1); + } +} diff --git a/crates/stdlib/src/socket.rs b/crates/stdlib/src/socket.rs index db30bc138..b3b2d10dd 100644 --- a/crates/stdlib/src/socket.rs +++ b/crates/stdlib/src/socket.rs @@ -3,7 +3,14 @@ pub(crate) use _socket::module_def; #[cfg(feature = "ssl")] -pub(super) use _socket::{PySocket, SockWaitKind, sock_wait, timeout_error_msg}; +pub(super) use _socket::{PySocket, timeout_error_msg}; +#[cfg(any( + feature = "__ssl-rustls", + all(feature = "ssl-openssl", not(all(unix, not(target_os = "aix")))) +))] +pub(super) use _socket::{SockWaitKind, sock_wait}; +#[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] +pub(crate) use _socket::{SocketCloseObserver, SocketIdentity}; #[pymodule] mod _socket { @@ -44,6 +51,10 @@ mod _socket { Ok(()) } + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + use alloc::sync::{Arc, Weak}; + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + use core::sync::atomic::AtomicBool; use core::sync::atomic::{AtomicI32, Ordering}; use core::{ mem::ManuallyDrop, @@ -53,6 +64,8 @@ mod _socket { use crossbeam_utils::atomic::AtomicCell; use host_socket::raw::Socket; use num_traits::ToPrimitive; + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + use std::sync::Mutex; use std::{ ffi, io::{self, Read, Write}, @@ -1058,6 +1071,66 @@ mod _socket { flags: i32, } + /// A native-only wakeup. Implementations must not execute Python. + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + pub(crate) trait SocketCloseObserver: Send + Sync { + fn socket_closed(&self); + } + + /// Identifies one initialization, even when a later socket reuses its fd. + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + #[derive(Debug, Default)] + pub(crate) struct SocketIdentity { + closed: AtomicBool, + observers: Mutex>>, + } + + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + impl SocketIdentity { + pub(crate) fn is_closed(&self) -> bool { + self.closed.load(Ordering::Acquire) + } + + /// Register atomically with the closed check, so close-before-wait + /// cannot lose a wakeup. The caller retains the strong observer. + pub(crate) fn subscribe(&self, observer: Arc) -> bool { + let mut observers = self.observers.lock().unwrap(); + if self.is_closed() { + return false; + } + observers.retain(|observer| observer.strong_count() != 0); + let observer = Arc::downgrade(&observer); + if !observers.iter().any(|current| current.ptr_eq(&observer)) { + observers.push(observer); + } + true + } + + pub(crate) fn unsubscribe(&self, observer: Arc) { + let observer = Arc::downgrade(&observer); + self.observers + .lock() + .unwrap() + .retain(|current| !current.ptr_eq(&observer)); + } + + // The socket write lock is held by the caller. Return weak observers + // rather than invoking them while either lock is held. + fn invalidate(&self) -> Vec> { + let mut observers = self.observers.lock().unwrap(); + self.closed.store(true, Ordering::Release); + core::mem::take(&mut *observers) + } + + fn notify_closed(observers: Vec>) { + for observer in observers { + if let Some(observer) = observer.upgrade() { + observer.socket_closed(); + } + } + } + } + #[pyattr(name = "socket")] #[pyattr(name = "SocketType")] #[pyclass(name = "socket")] @@ -1071,6 +1144,10 @@ mod _socket { proto: AtomicI32, pub(crate) timeout: AtomicCell, sock: PyRwLock>, + // Shared access takes `sock` first, in socket -> identity-slot order. + // Ordinary sockets never allocate an identity or observer list. + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + ssl_identity: Mutex>>, } const _: () = assert!(core::mem::size_of::>() == core::mem::size_of::()); @@ -1083,6 +1160,19 @@ mod _socket { proto: AtomicI32::new(0), timeout: AtomicCell::new(-1.0), sock: PyRwLock::new(None), + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + ssl_identity: Mutex::new(None), + } + } + } + + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + impl Drop for PySocket { + fn drop(&mut self) { + // Rust destruction must also notify when Python __del__ did not run. + // Keep ordinary socket destruction free of additional locking. + if self.ssl_identity.get_mut().unwrap().is_some() { + drop(self.take_sock()); } } } @@ -1119,6 +1209,64 @@ mod _socket { .ok_or_else(|| io::Error::from_raw_os_error(CLOSED_ERR)) } + /// Capture the current initialization under the socket read lock. + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + pub(crate) fn ssl_identity(&self) -> io::Result> { + let _sock = self.sock()?; + let mut identity = self.ssl_identity.lock().unwrap(); + Ok(identity + .get_or_insert_with(|| Arc::new(SocketIdentity::default())) + .clone()) + } + + /// Hold this guard only around a truly nonblocking transport syscall, + /// never around SSL processing, Python callbacks, or readiness waits. + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + pub(crate) fn sock_for_ssl_identity( + &self, + identity: &Arc, + ) -> io::Result> { + let sock = self.sock()?; + let current = self.ssl_identity.lock().unwrap(); + if !current + .as_ref() + .is_some_and(|current| Arc::ptr_eq(current, identity) && !identity.is_closed()) + { + return Err(io::Error::from_raw_os_error(CLOSED_ERR)); + } + Ok(sock) + } + + // Caller holds the socket write lock. Notifications must be delivered + // only after releasing that lock, including on initialization failure. + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + fn invalidate_ssl_identity_locked(&self) -> Vec> { + self.ssl_identity + .lock() + .unwrap() + .take() + .map_or_else(Vec::new, |identity| identity.invalidate()) + } + + fn take_sock(&self) -> Option { + let mut guard = self.sock.write(); + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + let observers = self.invalidate_ssl_identity_locked(); + let sock = guard.take(); + drop(guard); + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + SocketIdentity::notify_closed(observers); + sock + } + + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + fn invalidate_failed_ssl_init(&self) { + let guard = self.sock.write(); + let observers = self.invalidate_ssl_identity_locked(); + drop(guard); + SocketIdentity::notify_closed(observers); + } + /// Snapshot the inner socket without holding `self.sock` across a syscall. /// /// Blocking accept/recv/send must not keep the RwLock: `close()` needs @@ -1171,7 +1319,18 @@ mod _socket { self.kind.store(masked_kind, Ordering::Relaxed); self.proto.store(proto, Ordering::Relaxed); let mut s = self.sock.write(); - let sock = s.insert(sock); + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + let observers = self.invalidate_ssl_identity_locked(); + let old_socket = s.replace(sock); + // Closing a lingering socket may block. Unlock and wake old SSL + // operations first, including if initialization unwinds. + let mut s = scopeguard::guard(s, move |s| { + drop(s); + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + SocketIdentity::notify_closed(observers); + drop(old_socket); + }); + let sock = s.as_mut().unwrap(); // If SOCK_NONBLOCK is set, use timeout 0 (non-blocking) #[cfg(any( target_os = "android", @@ -1202,10 +1361,13 @@ mod _socket { )))] let timeout = DEFAULT_TIMEOUT.load(); self.timeout.store(timeout); - if timeout >= 0.0 { - sock.set_nonblocking(true)?; - } - Ok(()) + let result = if timeout >= 0.0 { + sock.set_nonblocking(true) + } else { + Ok(()) + }; + drop(s); + result } /// returns Err(blocking) @@ -1691,7 +1853,14 @@ mod _socket { type Args = SocketInitArgs; fn init(zelf: &Py, args: Self::Args, vm: &VirtualMachine) -> PyResult<()> { - Self::_init(zelf, args, vm).map_err(|e| e.into_pyexception(vm)) + let result = Self::_init(zelf, args, vm); + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + if result.is_err() { + // _init can fail before reaching init_inner, leaving an older + // raw socket in place but invalidating its SSL initialization. + zelf.invalidate_failed_ssl_init(); + } + result.map_err(|e| e.into_pyexception(vm)) } } @@ -2326,7 +2495,7 @@ mod _socket { #[pymethod] fn close(zelf: &Py) -> io::Result<()> { - let sock = zelf.sock.write().take(); + let sock = zelf.take_sock(); if let Some(sock) = sock { close_inner(into_sock_fileno(sock))?; } @@ -2336,7 +2505,7 @@ mod _socket { #[pymethod] #[inline] fn detach(zelf: &Py) -> i64 { - let sock = zelf.sock.write().take(); + let sock = zelf.take_sock(); sock.map_or(INVALID_SOCKET as i64, |s| into_sock_fileno(s) as i64) } @@ -3005,7 +3174,10 @@ mod _socket { } /// returns Ok(true) on timeout - #[cfg(feature = "ssl")] + #[cfg(any( + feature = "__ssl-rustls", + all(feature = "ssl-openssl", not(all(unix, not(target_os = "aix")))) + ))] pub(crate) fn sock_wait( sock: &Socket, wait_kind: SockWaitKind, @@ -3650,4 +3822,112 @@ mod _socket { host_socket::checked_cmsg_space(length) .ok_or_else(|| vm.new_overflow_error("CMSG_SPACE() argument out of range")) } + + #[cfg(test)] + #[cfg(all(feature = "ssl-openssl", unix, not(target_os = "aix")))] + mod ssl_identity_tests { + use super::{PySocket, SocketCloseObserver, SocketIdentity}; + use alloc::sync::{Arc, Weak}; + use core::sync::atomic::{AtomicUsize, Ordering}; + + struct Observer { + identity: Weak, + notifications: AtomicUsize, + } + + impl Observer { + fn new(identity: &Arc) -> Arc { + Arc::new(Self { + identity: Arc::downgrade(identity), + notifications: AtomicUsize::new(0), + }) + } + } + + impl SocketCloseObserver for Observer { + fn socket_closed(&self) { + if let Some(identity) = self.identity.upgrade() { + assert!(identity.is_closed()); + assert!(identity.observers.try_lock().is_ok()); + } + self.notifications.fetch_add(1, Ordering::Relaxed); + } + } + + #[test] + fn subscribe_before_close_notifies_after_invalidation() { + let identity = Arc::new(SocketIdentity::default()); + let observer = Observer::new(&identity); + assert!(identity.subscribe(observer.clone())); + assert!(identity.subscribe(observer.clone())); + let notifications = identity.invalidate(); + assert!(identity.is_closed()); + assert_eq!(observer.notifications.load(Ordering::Relaxed), 0); + assert_eq!(notifications.len(), 1); + SocketIdentity::notify_closed(notifications); + assert_eq!(observer.notifications.load(Ordering::Relaxed), 1); + assert!(identity.invalidate().is_empty()); + } + + #[test] + fn close_before_subscribe_rejects_registration() { + let identity = Arc::new(SocketIdentity::default()); + SocketIdentity::notify_closed(identity.invalidate()); + let observer = Observer::new(&identity); + assert!(!identity.subscribe(observer.clone())); + assert_eq!(observer.notifications.load(Ordering::Relaxed), 0); + } + + #[test] + fn registrations_do_not_keep_observers_alive() { + let identity = Arc::new(SocketIdentity::default()); + let observer = Observer::new(&identity); + let weak = Arc::downgrade(&observer); + assert!(identity.subscribe(observer.clone())); + drop(observer); + assert!(weak.upgrade().is_none()); + let next = Observer::new(&identity); + assert!(identity.subscribe(next.clone())); + let notifications = identity.invalidate(); + assert_eq!(notifications.len(), 1); + SocketIdentity::notify_closed(notifications); + assert_eq!(next.notifications.load(Ordering::Relaxed), 1); + } + + #[test] + fn uninitialized_socket_keeps_identity_unallocated() { + let socket = PySocket::default(); + assert!(socket.ssl_identity.lock().unwrap().is_none()); + assert!(socket.ssl_identity().is_err()); + assert!(socket.ssl_identity.lock().unwrap().is_none()); + } + + #[test] + fn failed_initialization_and_rust_drop_invalidate_identity() { + // Exercise lifecycle bookkeeping without creating an OS socket. + let socket = PySocket::default(); + let old = Arc::new(SocketIdentity::default()); + let old_observer = Observer::new(&old); + assert!(old.subscribe(old_observer.clone())); + { + let _socket = socket.sock.write(); + *socket.ssl_identity.lock().unwrap() = Some(old.clone()); + } + socket.invalidate_failed_ssl_init(); + assert!(old.is_closed()); + assert_eq!(old_observer.notifications.load(Ordering::Relaxed), 1); + + let new = Arc::new(SocketIdentity::default()); + let new_observer = Observer::new(&new); + assert!(new.subscribe(new_observer.clone())); + { + let _socket = socket.sock.write(); + *socket.ssl_identity.lock().unwrap() = Some(new.clone()); + } + assert!(!new.is_closed()); + drop(socket); + assert!(new.is_closed()); + assert_eq!(new_observer.notifications.load(Ordering::Relaxed), 1); + } + } } diff --git a/extra_tests/snippets/stdlib_ssl.py b/extra_tests/snippets/stdlib_ssl.py index cad505eeb..7ad7427f1 100644 --- a/extra_tests/snippets/stdlib_ssl.py +++ b/extra_tests/snippets/stdlib_ssl.py @@ -1,5 +1,10 @@ import pathlib +import socket import ssl +import threading +import time + +from test.test_ssl import HOST, ThreadedEchoServer, testing_context CERT = ( pathlib.Path(__file__).resolve().parent.parent.parent @@ -23,6 +28,18 @@ 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) +# A write that starts the handshake must release its caller's buffer on WANT. +# Save the valid retry before resizing and changing the original exporter. +exporter = bytearray(b"saved application bytes before WANT_READ") +saved_write = bytes(exporter) +try: + client.write(exporter) +except ssl.SSLWantReadError: + exporter[:] = b"changed after WANT" + exporter.extend(b"; no outstanding buffer export") +else: + raise AssertionError("write before the peer handshake did not need a read") + for _ in range(5): try: client.do_handshake() @@ -37,6 +54,12 @@ for _ in range(5): client.do_handshake() server.do_handshake() +# OpenSSL retries require the original content and length. Never retry with +# the mutated exporter: behavior for an invalid retry is not a parity claim. +assert client.write(saved_write) == len(saved_write) +server_in.write(client_out.read()) +assert server.read(len(saved_write)) == saved_write + try: server.unwrap() except ssl.SSLWantReadError: @@ -44,3 +67,285 @@ except ssl.SSLWantReadError: client_in.write(server_out.read() + TAIL) client.unwrap() assert client_in.read() == TAIL + + +TIMEOUT = 10.0 + + +class BoundedEchoServer(ThreadedEchoServer): + # Keep the canonical accept, handshake and echo implementation. Its normal + # context manager and handler joins need bounds for a deadlock regression. + class ConnectionHandler(ThreadedEchoServer.ConnectionHandler): + def __init__(self, server, connsock, addr): + super().__init__(server, connsock, addr) + self.sock.settimeout(TIMEOUT) + server.handlers.append(self) + + def run(self): + try: + try: + super().run() + finally: + self.close() + except BaseException as exc: + self.server.errors.append(exc) + + def join(self, timeout=TIMEOUT * 10): + super().join(timeout) + if self.is_alive(): + self.server.errors.append( + AssertionError("echo connection handler did not stop") + ) + self.server.stop() + + def __init__(self, context): + self.errors = [] + self.handlers = [] + super().__init__(context=context, chatty=False) + + def __enter__(self): + assert not self._in_context + self._in_context = True + self.start(threading.Event()) + if not self.flag.wait(TIMEOUT): + self.__exit__() + raise AssertionError("echo server did not start") + return self + + def run(self): + try: + try: + super().run() + finally: + self.close() + except BaseException as exc: + self.errors.append(exc) + + def __exit__(self, *args): + self.stop() + self.join(TIMEOUT * 2) + self._in_context = False + assert not self.is_alive(), "echo server did not stop" + for handler in self.handlers: + handler.join(TIMEOUT) + assert not handler.is_alive(), "echo connection handler did not stop" + if self.errors: + raise self.errors[0] + assert not self.conn_errors, self.conn_errors + + +def test_write_return_during_reader_callback(): + client_context, server_context, hostname = testing_context() + client_context.maximum_version = ssl.TLSVersion.TLSv1_2 + server_context.maximum_version = ssl.TLSVersion.TLSv1_2 + writer_thread = threading.current_thread() + reader_thread = None + round_active = False + + def message_callback(conn, direction, version, content_type, msg_type, data): + if ( + not round_active + or conn is not connection + or content_type != ssl._TLSContentType.HEADER + or msg_type != ssl._TLSContentType.APPLICATION_DATA + ): + return + current_thread = threading.current_thread() + if ( + direction == "write" + and current_thread is writer_thread + and not write_started.is_set() + ): + write_started.set() + assert reader_started.wait(TIMEOUT), "reader did not start" + # Encourage the reader to queue behind the native writer. Python + # cannot force its acquisition into the native cleanup window. + time.sleep(0.01) + elif ( + direction == "read" + and current_thread is reader_thread + and not read_callback_seen.is_set() + ): + read_callback_seen.set() + assert write_returned.wait(TIMEOUT), ( + "write did not return while another thread's read callback ran" + ) + + client_context._msg_callback = message_callback + with BoundedEchoServer(server_context) as echo_server: + with client_context.wrap_socket( + socket.socket(), server_hostname=hostname + ) as connection: + connection.settimeout(TIMEOUT) + connection.connect((HOST, echo_server.port)) + try: + # This is a cleanup/callback race, not another generic duplex + # smoke test. The canonical duplex test remains authoritative. + for _ in range(20): + payload = b"WRITE RETURN DURING READER CALLBACK\n" + write_started = threading.Event() + reader_started = threading.Event() + write_returned = threading.Event() + read_callback_seen = threading.Event() + errors = [] + + def read_echo(): + try: + assert write_started.wait(TIMEOUT), ( + "writer did not enter its application-record callback" + ) + reader_started.set() + received = b"" + while len(received) < len(payload): + chunk = connection.read(len(payload) - len(received)) + assert chunk, ( + "unexpected EOF in echoed application data" + ) + received += chunk + assert received == payload.lower(), received + except BaseException as exc: + errors.append(exc) + + reader_thread = threading.Thread(target=read_echo, daemon=True) + round_active = True + reader_thread.start() + try: + assert connection.write(payload) == len(payload) + finally: + # Set only after the Python write call has returned or + # raised. The reader callback never waits on itself. + write_returned.set() + reader_thread.join(TIMEOUT * 2) + if reader_thread.is_alive(): + # Interrupt transport I/O without starting SSL I/O. + try: + socket.socket.shutdown(connection, socket.SHUT_RDWR) + except OSError: + pass + reader_thread.join(TIMEOUT * 2) + round_active = False + assert not reader_thread.is_alive(), "reader did not stop" + if errors: + raise errors[0] + assert write_started.is_set(), "write callback was not exercised" + assert read_callback_seen.is_set(), ( + "read callback was not exercised" + ) + connection.write(b"over\n") + finally: + round_active = False + client_context._msg_callback = None + + +test_write_return_during_reader_callback() + + +# Limit initial nonblocking writes to 4 MiB. Small socket buffers plus a peer +# paused after its handshake must actually produce WANT_WRITE on this platform. +WRITE_SIZE = 64 * 1024 +MAX_INITIAL_WRITES = 64 + + +class PausedSinkServer(BoundedEchoServer): + class ConnectionHandler(BoundedEchoServer.ConnectionHandler): + def __init__(self, server, connsock, addr): + super().__init__(server, connsock, addr) + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_RCVBUF, 4096) + + def read(self): + assert self.server.read_enabled.wait(TIMEOUT), "sink was not released" + data = super().read() + if not data: + self.server.reached_eof.set() + else: + assert data == b"x" * len(data), "pending write data changed" + self.server.received += len(data) + assert self.server.received <= (MAX_INITIAL_WRITES + 1) * WRITE_SIZE + return data + + def write(self, data): + # Data was checked before canonical echo processing normalizes it. + # Send nothing back, avoiding two-peer application backpressure. + pass + + def run(self): + try: + super().run() + finally: + self.server.finished.set() + + def __init__(self, context): + self.read_enabled = threading.Event() + self.finished = threading.Event() + self.reached_eof = threading.Event() + self.received = 0 + super().__init__(context) + + +def test_exporter_release_after_application_want_write(): + client_context, server_context, hostname = testing_context() + client_context.maximum_version = ssl.TLSVersion.TLSv1_2 + server_context.maximum_version = ssl.TLSVersion.TLSv1_2 + with PausedSinkServer(server_context) as sink: + try: + with client_context.wrap_socket( + socket.socket(), server_hostname=hostname + ) as connection: + connection.settimeout(TIMEOUT) + connection.setsockopt(socket.SOL_SOCKET, socket.SO_SNDBUF, 4096) + connection.connect((HOST, sink.port)) + connection.setblocking(False) + exporter = bytearray(b"x" * WRITE_SIZE) + saved_write = bytes(exporter) + acknowledged = 0 + for _ in range(MAX_INITIAL_WRITES): + try: + acknowledged += connection.write(exporter) + except ssl.SSLWantWriteError: + break + else: + raise AssertionError( + "bounded application writes never reached WANT_WRITE" + ) + + # Returning WANT must release the export while native retries + # retain stable original bytes. Resize and mutate only the old + # exporter; the retry still uses identical saved bytes/length. + exporter[:] = b"changed after application WANT_WRITE" + exporter.extend(b"; exporter is no longer held") + sink.read_enabled.set() + # A successful write may accept a prefix. The first retry + # must still use identical bytes/length; advance only after a + # positive result, and keep the same pending slice on WANT. + retry_deadline = time.monotonic() + TIMEOUT + accepted = 0 + pending = saved_write + for _ in range(64): + remaining = retry_deadline - time.monotonic() + assert remaining > 0, "saved-byte retry deadline expired" + connection.settimeout(remaining) + try: + count = connection.write(pending) + except (ssl.SSLWantReadError, ssl.SSLWantWriteError): + continue + assert 0 < count <= len(pending), count + accepted += count + if accepted == len(saved_write): + break + pending = saved_write[accepted:] + else: + raise AssertionError("saved-byte retry exceeded 64 attempts") + assert accepted == len(saved_write) + + # TCP EOF follows every sent TLS byte. Let the sink consume to + # EOF before checking the total, so extra replay is detected. + socket.socket.shutdown(connection, socket.SHUT_WR) + assert sink.finished.wait(TIMEOUT), "sink did not consume through EOF" + assert sink.reached_eof.is_set(), "sink stopped before observing EOF" + assert sink.received == acknowledged + len(saved_write), sink.received + finally: + # Setup or oracle failure must not strand the paused peer. + sink.read_enabled.set() + + +test_exporter_release_after_application_want_write()