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