From f69ed34cb595fdb3c324c0ccf6e57da593a7dc77 Mon Sep 17 00:00:00 2001 From: scttbnsn <80784472+scttbnsn@users.noreply.github.com> Date: Thu, 2 Jul 2026 23:22:32 -0400 Subject: [PATCH] Stop reading when the opening handshake fails. Fix #1596. --- docs/project/changelog.rst | 3 ++ src/websockets/sync/connection.py | 21 ++++++++- tests/sync/test_client.py | 74 +++++++++++++++++++++++++++++++ tests/sync/test_connection.py | 27 +++++++++++ 4 files changed, 124 insertions(+), 1 deletion(-) diff --git a/docs/project/changelog.rst b/docs/project/changelog.rst index 0c96d654c..422c87d19 100644 --- a/docs/project/changelog.rst +++ b/docs/project/changelog.rst @@ -48,6 +48,9 @@ Bug fixes * Prevented ``Frame.__str__`` from crashing when a text frame is fragmented in the middle of a UTF-8 sequence. +* Fixed a race condition that could delay closing connections in the + :mod:`threading` implementation. + .. _16.0: 16.0 diff --git a/src/websockets/sync/connection.py b/src/websockets/sync/connection.py index 665f478ac..0ed46bec5 100644 --- a/src/websockets/sync/connection.py +++ b/src/websockets/sync/connection.py @@ -20,7 +20,7 @@ ) from ..frames import DATA_OPCODES, CloseCode, Frame, Opcode from ..http11 import Request, Response -from ..protocol import CLOSED, OPEN, Event, Protocol, State +from ..protocol import CLOSED, CONNECTING, OPEN, Event, Protocol, State from ..typing import BytesLike, Data, DataLike, LoggerLike, Subprotocol from .messages import Assembler from .utils import Deadline @@ -867,6 +867,14 @@ def recv_events(self) -> None: if self.close_deadline is None: self.close_deadline = Deadline(self.close_timeout) + # If the opening handshake failed, the connection never + # reached the OPEN state and there's no closing handshake + # to wait for. Remember this so we can stop reading below. + handshake_failed = ( + self.protocol.state is CONNECTING + and self.protocol.close_expected() + ) + # Unlock conn_mutex before processing events. Else, the # application can't send messages in response to events. @@ -878,6 +886,17 @@ def recv_events(self) -> None: # This isn't expected to raise an exception. self.process_event(event) + # When the opening handshake fails, the connection is unusable + # and close_socket() runs as soon as the handshake failure + # propagates. Stop reading instead of waiting for the peer to + # close the TCP connection, because closing a socket in one + # thread doesn't always interrupt recv() in another thread -- + # e.g. on macOS, when the protocol already half-closed the + # socket, shutdown() fails with ENOTCONN and close() doesn't + # interrupt recv(). + if handshake_failed: + break + # Breaking out of the while True: ... loop means that we believe # that the socket doesn't work anymore. diff --git a/tests/sync/test_client.py b/tests/sync/test_client.py index 415343911..02ce2e82d 100644 --- a/tests/sync/test_client.py +++ b/tests/sync/test_client.py @@ -1,3 +1,4 @@ +import errno import http import logging import os @@ -190,6 +191,79 @@ def close_connection(self, request): "connection closed while reading HTTP status line", ) + def test_connection_closed_promptly_after_invalid_status_response(self): + """Client closes the connection promptly when the handshake fails.""" + + class HalfClosingSocket(socket.socket): + """ + Reproduces the shutdown()/close() semantics observed on macOS: + once the protocol half-closes the socket for writing, a further + shutdown() raises ENOTCONN instead of succeeding, and close() + doesn't interrupt a recv() call already blocked in another + thread. + + """ + + def __init__(self, plain): + super().__init__(fileno=plain.detach()) + self.half_closed = False + + def shutdown(self, how): + if how == socket.SHUT_WR and not self.half_closed: + self.half_closed = True + super().shutdown(how) + else: + raise OSError(errno.ENOTCONN, "Socket is not connected") + + def close(self): + pass # doesn't interrupt a recv() blocked in another thread + + def really_close(self): + super().close() + + response = ( + b"HTTP/1.1 302 Found\r\n" + b"Location: http://localhost/\r\n" + b"Content-Length: 0\r\n" + b"\r\n" + ) + keep_open = threading.Event() + + def keep_connection_open(): + conn, _ = server.accept() + with conn: + request = b"" + while b"\r\n\r\n" not in request: + data = conn.recv(4096) + if data == b"": # pragma: no cover + return + request += data + conn.sendall(response) + # Keep the TCP connection open past the handshake failure. + keep_open.wait() + + with socket.create_server(("localhost", 0)) as server: + host, port = server.getsockname() + + thread = threading.Thread(target=keep_connection_open) + thread.start() + self.addCleanup(thread.join) + self.addCleanup(keep_open.set) + + plain = socket.create_connection((host, port)) + sock = HalfClosingSocket(plain) + self.addCleanup(sock.really_close) + + t0 = time.time() + with self.assertRaises(InvalidStatus): + with connect(f"ws://{host}:{port}/", sock=sock, close_timeout=20 * MS): + self.fail("did not raise") + t1 = time.time() + + # The client doesn't wait anywhere close to close_timeout because it + # doesn't wait for the peer to close the connection. + self.assertLess(t1 - t0, 10 * MS) + def test_http_response(self): """Client reads HTTP response.""" diff --git a/tests/sync/test_connection.py b/tests/sync/test_connection.py index 6c45bc749..2b1d3492f 100644 --- a/tests/sync/test_connection.py +++ b/tests/sync/test_connection.py @@ -655,6 +655,33 @@ def fragments(): ) self.assertIsNone(exc.__cause__) + # Test closing the connection when the opening handshake fails. + + def test_stops_reading_when_the_opening_handshake_fails(self): + """recv_events stops reading when the opening handshake fails.""" + socket_, remote_socket = socket.socketpair() + self.addCleanup(remote_socket.close) + # Simulate the state of the protocol after the opening handshake + # failed: EOF was sent, incoming data is discarded, and the state + # is still CONNECTING. See ClientProtocol.parse and + # ServerProtocol.parse. + protocol = Protocol(self.LOCAL, state=State.CONNECTING) + protocol.send_eof() + protocol.parser = protocol.discard() + next(protocol.parser) # start coroutine + connection = Connection(socket_, protocol, close_timeout=20 * MS) + self.addCleanup(connection.close_socket) + + t0 = time.time() + remote_socket.sendall(b"remaining data to discard") + connection.recv_events_thread.join(1) + t1 = time.time() + + # The connection closes promptly instead of waiting for the peer to + # close the TCP connection. + self.assertFalse(connection.recv_events_thread.is_alive()) + self.assertLess(t1 - t0, 10 * MS) + # Test ping. @patch("random.getrandbits")