Fix SSL handshake over-reading in STARTTLS (#7417)

During STARTTLS handshake, sock_recv(16KB) could consume
application data that arrived alongside handshake records.
The consumed data ended up in rustls's internal buffer where
select() could not detect it, causing asyncore-based servers
to miss readable events and the peer to time out.

Use MSG_PEEK to find the TLS record boundary, then recv()
only one complete record. Remaining data stays in the kernel
TCP buffer, visible to select(). This matches OpenSSL's
default no-read-ahead behaviour.

Fixes flaky test_poplib (TestPOP3_TLSClass) failures.

crates/stdlib/src/openssl.rs

@@ -2859,10 +2859,11 @@ mod _ssl {
                     // Wait briefly for peer's close_notify before retrying
                     match socket_stream.select(SslNeeds::Read, &deadline) {
                         SelectRet::TimedOut => {
-                            return Err(vm.new_exception_msg(
-                                vm.ctx.exceptions.timeout_error.to_owned(),
-                                "The read operation timed out".to_owned(),
-                            ));
+                            return Err(socket::timeout_error_msg(
+                                vm,
+                                "The read operation timed out".to_string(),
+                            )
+                            .upcast());
                         }
                         SelectRet::Closed => {
                             return Err(socket_closed_error(vm));
@@ -2901,10 +2902,7 @@ mod _ssl {
                         } else {
                             "The write operation timed out"
                         };
-                        return Err(vm.new_exception_msg(
-                            vm.ctx.exceptions.timeout_error.to_owned(),
-                            msg.to_owned(),
-                        ));
+                        return Err(socket::timeout_error_msg(vm, msg.to_string()).upcast());
                     }
                     SelectRet::Closed => {
                         return Err(socket_closed_error(vm));

crates/stdlib/src/ssl.rs

@@ -4188,10 +4188,11 @@ mod _ssl {
                                     let now = std::time::Instant::now();
                                     if now >= dl {
                                         // Timeout reached - raise TimeoutError
-                                        return Err(vm.new_exception_msg(
-                                            vm.ctx.exceptions.timeout_error.to_owned(),
-                                            "The read operation timed out".into(),
-                                        ));
+                                        return Err(timeout_error_msg(
+                                            vm,
+                                            "The read operation timed out".to_string(),
+                                        )
+                                        .upcast());
                                     }
                                     Some(dl - now)
                                 } else {
@@ -4207,11 +4208,11 @@ mod _ssl {
 
                                 if timed_out {
                                     // Timeout waiting for peer's close_notify
-                                    // Raise TimeoutError
-                                    return Err(vm.new_exception_msg(
-                                        vm.ctx.exceptions.timeout_error.to_owned(),
-                                        "The read operation timed out".into(),
-                                    ));
+                                    return Err(timeout_error_msg(
+                                        vm,
+                                        "The read operation timed out".to_string(),
+                                    )
+                                    .upcast());
                                 }
 
                                 // Try to read data from socket

crates/stdlib/src/ssl/compat.rs

@@ -1161,6 +1161,83 @@ fn handshake_write_loop(
 ///
 /// Waits for and reads TLS records from the peer, handling SNI callback setup.
 /// Returns (made_progress, is_first_sni_read).
+/// TLS record header size (content_type + version + length).
+const TLS_RECORD_HEADER_SIZE: usize = 5;
+
+/// Determine how many bytes to read from the socket during a TLS handshake.
+///
+/// OpenSSL reads one TLS record at a time (no read-ahead by default).
+/// Rustls, however, consumes all available TCP data when fed via read_tls().
+/// If application data arrives simultaneously with the final handshake record,
+/// the eager read drains the TCP buffer, leaving the app data in rustls's
+/// internal buffer where select() cannot see it.  This causes asyncore-based
+/// servers (which rely on select() for readability) to miss the data and the
+/// peer times out.
+///
+/// Fix: peek at the TCP buffer to find the first complete TLS record boundary
+/// and recv() only that many bytes.  Any remaining data (including application
+/// data that piggybacked on the same TCP segment) stays in the kernel buffer
+/// and remains visible to select().
+fn handshake_recv_one_record(socket: &PySSLSocket, vm: &VirtualMachine) -> SslResult<PyObjectRef> {
+    // Peek at what is available without consuming it.
+    let peeked_obj = match socket.sock_peek(SSL3_RT_MAX_PLAIN_LENGTH, vm) {
+        Ok(d) => d,
+        Err(e) => {
+            if is_blocking_io_error(&e, vm) {
+                return Err(SslError::WantRead);
+            }
+            return Err(SslError::Py(e));
+        }
+    };
+
+    let peeked = ArgBytesLike::try_from_object(vm, peeked_obj)
+        .map_err(|_| SslError::Syscall("Expected bytes-like object from peek".to_string()))?;
+    let peeked_bytes = peeked.borrow_buf();
+
+    if peeked_bytes.is_empty() {
+        return Err(SslError::WantRead);
+    }
+
+    if peeked_bytes.len() < TLS_RECORD_HEADER_SIZE {
+        // Not enough data for a TLS record header yet.
+        // Read all available bytes so rustls can buffer the partial header;
+        // this avoids busy-waiting because the kernel buffer is now empty
+        // and select() will only wake us when new data arrives.
+        return socket.sock_recv(peeked_bytes.len(), vm).map_err(|e| {
+            if is_blocking_io_error(&e, vm) {
+                SslError::WantRead
+            } else {
+                SslError::Py(e)
+            }
+        });
+    }
+
+    // Parse the TLS record length from the header.
+    let record_body_len = u16::from_be_bytes([peeked_bytes[3], peeked_bytes[4]]) as usize;
+    let total_record_size = TLS_RECORD_HEADER_SIZE + record_body_len;
+
+    let recv_size = if peeked_bytes.len() >= total_record_size {
+        // Complete record available — consume exactly one record.
+        total_record_size
+    } else {
+        // Incomplete record — consume everything so the kernel buffer is
+        // drained and select() will block until more data arrives.
+        peeked_bytes.len()
+    };
+
+    // Must drop the borrow before calling sock_recv (which re-enters Python).
+    drop(peeked_bytes);
+    drop(peeked);
+
+    socket.sock_recv(recv_size, vm).map_err(|e| {
+        if is_blocking_io_error(&e, vm) {
+            SslError::WantRead
+        } else {
+            SslError::Py(e)
+        }
+    })
+}
+
 fn handshake_read_data(
     conn: &mut TlsConnection,
     socket: &PySSLSocket,
@@ -1189,23 +1266,25 @@ fn handshake_read_data(
         }
     }
 
-    let data_obj = match socket.sock_recv(SSL3_RT_MAX_PLAIN_LENGTH, vm) {
-        Ok(d) => d,
-        Err(e) => {
-            if is_blocking_io_error(&e, vm) {
-                return Err(SslError::WantRead);
-            }
-            // In socket mode, if recv times out and we're only waiting for read
-            // (no wants_write), we might be waiting for optional NewSessionTicket in TLS 1.3
-            // Consider the handshake complete
-            if !is_bio && !conn.wants_write() {
-                // Check if it's a timeout exception
-                if e.fast_isinstance(vm.ctx.exceptions.timeout_error) {
-                    // Timeout waiting for optional data - handshake is complete
+    let data_obj = if !is_bio {
+        // In socket mode, read one TLS record at a time to avoid consuming
+        // application data that may arrive alongside the final handshake
+        // record.  This matches OpenSSL's default (no read-ahead) behaviour
+        // and keeps remaining data in the kernel buffer where select() can
+        // detect it.
+        handshake_recv_one_record(socket, vm)?
+    } else {
+        match socket.sock_recv(SSL3_RT_MAX_PLAIN_LENGTH, vm) {
+            Ok(d) => d,
+            Err(e) => {
+                if is_blocking_io_error(&e, vm) {
+                    return Err(SslError::WantRead);
+                }
+                if !conn.wants_write() && e.fast_isinstance(vm.ctx.exceptions.timeout_error) {
                     return Ok((false, false));
                 }
+                return Err(SslError::Py(e));
             }
-            return Err(SslError::Py(e));
         }
     };
 

crates/vm/src/stdlib/_winapi.rs

@@ -1614,7 +1614,13 @@ mod _winapi {
 
             if let Some(err) = err {
                 if err == windows_sys::Win32::Foundation::WAIT_TIMEOUT {
-                    return Err(vm.new_exception_empty(vm.ctx.exceptions.timeout_error.to_owned()));
+                    return Err(vm
+                        .new_os_subtype_error(
+                            vm.ctx.exceptions.timeout_error.to_owned(),
+                            None,
+                            "timed out",
+                        )
+                        .upcast());
                 }
                 if err == windows_sys::Win32::Foundation::ERROR_CONTROL_C_EXIT {
                     return Err(vm
@@ -1783,7 +1789,13 @@ mod _winapi {
             // Return result
             if let Some(e) = thread_err {
                 if e == windows_sys::Win32::Foundation::WAIT_TIMEOUT {
-                    return Err(vm.new_exception_empty(vm.ctx.exceptions.timeout_error.to_owned()));
+                    return Err(vm
+                        .new_os_subtype_error(
+                            vm.ctx.exceptions.timeout_error.to_owned(),
+                            None,
+                            "timed out",
+                        )
+                        .upcast());
                 }
                 if e == windows_sys::Win32::Foundation::ERROR_CONTROL_C_EXIT {
                     return Err(vm

crates/vm/src/vm/vm_new.rs

@@ -110,8 +110,8 @@ impl VirtualMachine {
         debug_assert_eq!(
             exc_type.slots.basicsize,
             core::mem::size_of::<PyBaseException>(),
-            "vm.new_exception() is only for exception types without additional payload. The given type '{}' is not allowed.",
-            exc_type.class().name()
+            "vm.new_exception() is only for exception types without additional payload. The given type '{}' is not allowed. Use vm.new_os_subtype_error() for OSError subtypes.",
+            exc_type.name()
         );
 
         PyRef::new_ref(