Stop wire close from SMBRawIO finalizer

src/smbclient/_io.py

@@ -3,6 +3,7 @@
 
 import io
 import logging
+import warnings
 
 from smbclient._pool import ClientConfig, dfs_request, get_smb_tree
 from smbprotocol import MAX_PAYLOAD_SIZE
@@ -424,6 +425,17 @@ def __init__(
 
         super().__init__()
 
+    def __del__(self):
+        # io.IOBase.__del__ would call self.close(), which self-deadlocks if GC
+        # runs it on the SMB worker thread.
+        try:
+            closed = self.closed
+        except AttributeError:
+            return
+        if closed:
+            return
+        warnings.warn(f"unclosed SMB handle {self._name!r}", ResourceWarning, source=self)
+
     def __enter__(self):
         self.open()
         return self

src/smbprotocol/connection.py

@@ -1033,6 +1033,12 @@ def receive(self, request, wait=True, timeout=None, resolve_symlinks=True):
         :param resolve_symlinks: Set to automatically resolve symlinks in the path when opening a file or directory.
         :return: SMB2HeaderResponse of the received message
         """
+        # Only the worker thread delivers responses, so calling receive() from it would wait
+        # on an event only it can set.
+        worker = self._t_worker
+        if worker is not None and threading.get_ident() == worker.ident:
+            raise SMBException("Connection.receive() called from the SMB worker thread, would self-deadlock")
+
         # Make sure the receiver is still active, if not this raises an exception.
         self._check_worker_running()
 

tests/test_connection.py

@@ -1115,6 +1115,16 @@ def test_message_worker_timeout_connection_dead(self, mocker):
         with pytest.raises(SMBConnectionClosed):
             connection.receive(None)
 
+    def test_receive_from_worker_thread_raises(self, mocker):
+        # receive() on the worker thread would wait for an event only the worker can set.
+        connection = Connection(uuid.uuid4(), "server", 445, True)
+        fake_worker = mocker.MagicMock()
+        fake_worker.ident = threading.get_ident()
+        connection._t_worker = fake_worker
+
+        with pytest.raises(SMBException, match="self-deadlock"):
+            connection.receive(None)
+
     def test_verify_fail_no_session(self, smb_real):
         connection = Connection(uuid.uuid4(), smb_real[2], smb_real[3], True)
         connection.connect()

tests/test_smbclient_io.py

@@ -0,0 +1,41 @@
+# Copyright: (c) 2026, Acconeer AB <henrik.nilsson@acconeer.com>
+# MIT License (see LICENSE or https://opensource.org/licenses/MIT)
+
+import warnings
+
+import pytest
+
+import smbclient._io as io
+from smbclient._io import SMBRawIO
+
+
+@pytest.fixture
+def raw(mocker):
+    """Construct an SMBRawIO without opening any socket.
+
+    SMBRawIO.__init__ normally calls get_smb_tree which establishes a connection.
+    Patching that and Open lets the finalizer be exercised deterministically.
+    """
+    mock_tree = mocker.MagicMock()
+    mocker.patch.object(io, "get_smb_tree", return_value=(mock_tree, "file"))
+    mocker.patch.object(io, "Open")
+
+    return SMBRawIO(r"\\server\share\file.txt", mode="r", share_access="r")
+
+
+def test_del_does_not_close_a_connected_handle(raw):
+    # The override must skip self.close() to avoid deadlocking the worker.
+    raw.fd.connected = True
+
+    with warnings.catch_warnings():
+        warnings.simplefilter("ignore", ResourceWarning)
+        raw.__del__()
+
+    raw.fd.close.assert_not_called()
+
+
+def test_del_emits_resource_warning_for_leaked_handle(raw):
+    raw.fd.connected = True
+
+    with pytest.warns(ResourceWarning, match=r"unclosed SMB handle"):
+        raw.__del__()