Fix race conditions and improve robustness during socket I/O

.flake8

@@ -133,7 +133,7 @@ per-file-ignores =
   cheroot/test/conftest.py: DAR101, DAR201, DAR301, I001, I003, I005, WPS100, WPS130, WPS325, WPS354, WPS420, WPS422, WPS430, WPS457
   cheroot/test/helper.py: DAR101, DAR201, DAR401, I001, I003, I004, N802, WPS110, WPS111, WPS121, WPS201, WPS220, WPS231, WPS301, WPS414, WPS421, WPS422, WPS505
   cheroot/test/test_cli.py: DAR101, DAR201, I001, I005, N802, S101, S108, WPS110, WPS421, WPS431, WPS473
-  cheroot/test/test_makefile.py: DAR101, DAR201, I004, RST304, S101, WPS110, WPS122
+  cheroot/test/test_makefile.py: DAR101, DAR201, I004, RST304, S101, WPS110, WPS122, WPS202
   cheroot/test/test_wsgi.py: DAR101, DAR301, I001, I004, S101, WPS110, WPS111, WPS117, WPS118, WPS121, WPS210, WPS421, WPS430, WPS432, WPS441, WPS509
   cheroot/test/test_core.py: C815, DAR101, DAR201, DAR401, I003, I004, N805, N806, S101, WPS110, WPS111, WPS114, WPS121, WPS202, WPS204, WPS226, WPS229, WPS324, WPS421, WPS422, WPS432, WPS602
   cheroot/test/test_dispatch.py: DAR101, DAR201, S101, WPS111, WPS121, WPS422, WPS430

cheroot/errors.py

@@ -3,6 +3,8 @@
 import errno
 import sys
 
+from . import _compat
+
 
 class MaxSizeExceeded(Exception):
     """Exception raised when a client sends more data then allowed under limit.
@@ -66,19 +68,23 @@ def plat_specific_errors(*errnames):
 
 
 acceptable_sock_shutdown_error_codes = {
+    errno.EBADF,
     errno.ENOTCONN,
     errno.EPIPE,
     errno.ESHUTDOWN,  # corresponds to BrokenPipeError in Python 3
     errno.ECONNRESET,  # corresponds to ConnectionResetError in Python 3
+    *((errno.WSAENOTSOCK,) if _compat.IS_WINDOWS else ()),
 }
 """Errors that may happen during the connection close sequence.
 
+* EBADF - raised when operating on a closed socket
 * ENOTCONN — client is no longer connected
 * EPIPE — write on a pipe while the other end has been closed
 * ESHUTDOWN — write on a socket which has been shutdown for writing
 * ECONNRESET — connection is reset by the peer, we received a TCP RST packet
 
 Refs:
+
 * https://github.com/cherrypy/cheroot/issues/341#issuecomment-735884889
 * https://bugs.python.org/issue30319
 * https://bugs.python.org/issue30329
@@ -87,4 +93,8 @@ def plat_specific_errors(*errnames):
 * https://docs.microsoft.com/windows/win32/api/winsock/nf-winsock-shutdown
 """
 
-acceptable_sock_shutdown_exceptions = (BrokenPipeError, ConnectionResetError)
+
+acceptable_sock_shutdown_exceptions = (
+    BrokenPipeError,  # Covers EPIPE and ESHUTDOWN
+    ConnectionResetError,  # Covers ECONNRESET
+)

cheroot/errors.pyi

@@ -10,4 +10,7 @@ socket_error_eintr: List[int]
 socket_errors_to_ignore: List[int]
 socket_errors_nonblocking: List[int]
 acceptable_sock_shutdown_error_codes: Set[int]
-acceptable_sock_shutdown_exceptions: Tuple[Type[Exception], ...]
+acceptable_sock_shutdown_exceptions: Tuple[
+    Type[BrokenPipeError],
+    Type[ConnectionResetError],
+]

cheroot/makefile.py

@@ -4,6 +4,8 @@
 import _pyio as io
 import socket
 
+from . import errors as _errors
+
 
 # Write only 16K at a time to sockets
 SOCK_WRITE_BLOCKSIZE = 16384
@@ -32,8 +34,44 @@ def _flush_unlocked(self):
                 n = self.raw.write(bytes(self._write_buf))
             except io.BlockingIOError as e:
                 n = e.characters_written
+
+            if n == 0:
+                # If nothing was written we need to break
+                # to avoid infinte loops
+                break
+
             del self._write_buf[:n]
 
+    def close(self):
+        """
+        Close the stream and its underlying file object.
+
+        This method is designed to be idempotent (it can be called multiple
+        times without side effects). It gracefully handles a race condition
+        where the underlying socket may have already been closed by the remote
+        client or another thread.
+
+        A :exc:`ConnectionError` or :exc:`OSError` with
+        :data:`~errno.EBADF` or :data:`~errno.ENOTCONN` is caught
+        and ignored, as these indicate a normal, expected connection teardown.
+        Other exceptions are re-raised.
+        """
+        # pylint incorrectly flags inherited self.closed property as constant
+        if self.closed:  # pylint: disable=using-constant-test
+            return
+
+        try:
+            super().close()
+        except _errors.acceptable_sock_shutdown_exceptions:
+            return
+        except ConnectionError:
+            return
+        except OSError as err:
+            # Handle EBADF and other acceptable socket shutdown errors
+            if err.errno in _errors.acceptable_sock_shutdown_error_codes:
+                return
+            raise
+
 
 class StreamReader(io.BufferedReader):
     """Socket stream reader."""

cheroot/server.py

@@ -1192,9 +1192,11 @@ def write(self, chunk):
         if self.chunked_write and chunk:
             chunk_size_hex = hex(len(chunk))[2:].encode('ascii')
             buf = [chunk_size_hex, CRLF, chunk, CRLF]
-            self.conn.wfile.write(EMPTY.join(buf))
+            data = EMPTY.join(buf)
         else:
-            self.conn.wfile.write(chunk)
+            data = chunk
+
+        self.conn.wfile.write(data)
 
     def send_headers(self):  # noqa: C901  # FIXME
         """Assert, process, and send the HTTP response message-headers.

cheroot/ssl/pyopenssl.py

@@ -50,6 +50,8 @@
    pyopenssl
 """
 
+import errno
+import os
 import socket
 import sys
 import threading
@@ -77,6 +79,45 @@
 from . import Adapter
 
 
+@contextlib.contextmanager
+def _morph_syscall_to_connection_error(method_name, /):
+    """
+    Handle :exc:`OpenSSL.SSL.SysCallError` in a wrapped method.
+
+    This context manager catches and re-raises SSL system call errors
+    with appropriate exception types.
+
+    Yields:
+        None: Execution continues within the context block.
+    """  # noqa: DAR301
+    try:
+        yield
+    except SSL.SysCallError as ssl_syscall_err:
+        connection_error_map = {
+            errno.EBADF: ConnectionError,  # socket is gone?
+            errno.ECONNABORTED: ConnectionAbortedError,
+            errno.ECONNREFUSED: ConnectionRefusedError,
+            errno.ECONNRESET: ConnectionResetError,
+            errno.ENOTCONN: ConnectionError,
+            errno.EPIPE: BrokenPipeError,
+            errno.ESHUTDOWN: BrokenPipeError,
+        }
+        error_code = ssl_syscall_err.args[0] if ssl_syscall_err.args else None
+        error_msg = (
+            os.strerror(error_code)
+            if error_code is not None
+            else repr(ssl_syscall_err)
+        )
+        conn_err_cls = connection_error_map.get(
+            error_code,
+            ConnectionError,
+        )
+        raise conn_err_cls(
+            error_code,
+            f'Faied to {method_name!s} the PyOpenSSL connection: {error_msg!s}',
+        ) from ssl_syscall_err
+
+
 class SSLFileobjectMixin:
     """Base mixin for a TLS socket stream."""
 
@@ -269,6 +310,18 @@ def __init__(self, *args):
         self._ssl_conn = SSL.Connection(*args)
         self._lock = threading.RLock()
 
+    @_morph_syscall_to_connection_error('close')
+    def close(self):
+        """Close the connection, translating OpenSSL errors for shutdown."""
+        with self._lock:
+            return self._ssl_conn.close()
+
+    @_morph_syscall_to_connection_error('shutdown')
+    def shutdown(self):
+        """Shutdown the connection, translating OpenSSL errors."""
+        with self._lock:
+            return self._ssl_conn.shutdown()
+
 
 class pyOpenSSLAdapter(Adapter):
     """A wrapper for integrating :doc:`pyOpenSSL <pyopenssl:index>`."""

cheroot/test/test_makefile.py

@@ -1,10 +1,16 @@
 """Tests for :py:mod:`cheroot.makefile`."""
 
+import errno
+import io
+import math
+
+import pytest
+
 from cheroot import makefile
 
 
 class MockSocket:
-    """A mock socket."""
+    """A mock socket for emulating buffered I/O."""
 
     def __init__(self):
         """Initialize :py:class:`MockSocket`."""
@@ -51,3 +57,116 @@ def test_bytes_written():
     wfile = makefile.MakeFile(sock, 'w')
     wfile.write(b'bar')
     assert wfile.bytes_written == 3
+
+
+def test_close_is_idempotent():
+    """Test that double ``close()`` does not error out."""
+    raw_buffer = io.BytesIO()
+    buffered_writer = makefile.BufferedWriter(raw_buffer)
+
+    # Should not raise any exceptions
+    buffered_writer.close()
+    assert buffered_writer.closed
+
+    buffered_writer.close()  # Second call should be safe
+    assert buffered_writer.closed
+
+
+def test_close_handles_already_closed_buffer():
+    """Test that ``close()`` handles already closed underlying buffer."""
+    raw_buffer = io.BytesIO()
+    buffered_writer = makefile.BufferedWriter(raw_buffer)
+
+    # Close the underlying buffer first
+    raw_buffer.close()
+
+    # This should not raise an exception
+    assert raw_buffer.closed
+    assert buffered_writer.closed
+
+
+def test_flush_unlocked_handles_blocking_io_error(mock_buffer_writer, mocker):
+    """
+    Test that a BlockingIOError is handled correctly.
+
+    We extracting characters_written,
+    and execution continues without raising the error.
+    """
+    # 1. Create a mock object to replace the real 'write' method
+    mock_write_method = mocker.Mock()
+
+    # 2. Set the side effect on the mock object
+    err = io.BlockingIOError(errno.EAGAIN, 'Resource temporarily unavailable')
+    err.characters_written = 5
+    mock_write_method.side_effect = err
+
+    # 3. Use mocker.patch.object to replace the 'write' method
+    # with mock_write_method
+    mocker.patch.object(mock_buffer_writer.raw, 'write', new=mock_write_method)
+
+    # Check the initial state of the buffer
+    initial_len = len(mock_buffer_writer._write_buf)
+
+    # 4. Execute the code
+    try:
+        mock_buffer_writer._flush_unlocked()
+    except Exception as exc:
+        pytest.fail(f'Unexpected exception raised: {type(exc).__name__}')
+
+    # 5. Verify the side-effect (buffer should be empty)
+    assert len(mock_buffer_writer._write_buf) == 0
+
+    # 6 Check mock calls (Logic/Mechanism)
+    # The number of calls should be
+    # initial_len / bytes_written_per_call
+    expected_calls = math.ceil(initial_len / 5)
+    assert mock_write_method.call_count == expected_calls
+
+
+class MockRawSocket:
+    """
+    A mock raw socket for emulating low level unbuffered I/O.
+
+    We use this mock with ``io.BufferedWriter``, which accesses it via
+    the ``.raw`` attribute.
+    """
+
+    def __init__(self, *args, **kwargs):
+        """Initialize :py:class:`MockRawSocket`."""
+        # 1. Call the parent's init to set up self.messages
+        super().__init__(*args, **kwargs)
+
+        # 2. Rquired by the io.BufferedWriter base class
+        self._is_closed = False
+
+    def write(self, message):
+        """Emulate ``io.RawIOBase write``."""
+        # Use the underlying send method implemented in MockSocket
+        return self.send(message)
+
+    def writable(self):
+        """Indicate that the raw stream supports writing."""
+        return True
+
+    def send(self, message):
+        """Emulate a send."""
+        return len(message)
+
+    def close(self):
+        """Emulate close."""
+        self._is_closed = True
+
+    @property
+    def closed(self):
+        """Emulate the required ``closed`` property."""
+        return self._is_closed
+
+
+@pytest.fixture
+def mock_buffer_writer():
+    """Fixture to create a BufferedWriter instance with a mock raw socket."""
+    # Create a BufferedWriter instance with a buffer that has content
+    # to ensure _flush_unlocked attempts to write.
+    writer = makefile.BufferedWriter(MockRawSocket())
+    writer._write_buf = bytearray(b'data to flush')
+    return writer

docs/changelog-fragments.d/779.bugfix.rst

@@ -0,0 +1,4 @@
+Socket I/O is now resilient to race conditions happening during connection teardown
+due to sockets dying independently or being closed externally.
+
+-- by :user:`julianz-`

docs/spelling_wordlist.txt

@@ -65,6 +65,7 @@ subpackages
 symlinked
 syscall
 systemd
+teardown
 threadpool
 Tidelift
 TLS