Wait for the peer to confirm the disconnect request

CHANGES.md

@@ -17,6 +17,10 @@ Unreleased
   The optimizer now also puts at most 20 items in one request; `optimizer.packetize` takes a `max_items` argument
   for this. The ctypes form of `read_multi_vars` and `write_multi_vars` keep the limit.
 * `read_tags` now raises `NotImplementedError` for a symbolic tag, as `read_tag` does, in both clients.
+* `disconnect()` now waits up to one second for the PLC to confirm the COTP disconnect request, or close the
+  connection, before closing the socket, in both clients. Previously the FIN followed the request within
+  microseconds, and a CP 343-1 answered with a reset and kept the connection's resource until the PLC restarted.
+  The request also no longer carries a stray byte after its header, which class 0 doesn't allow (#972).
 
 3.4.0
 -----

snap7/async_client.py

@@ -55,8 +55,12 @@ class AsyncISOTCPConnection:
     COTP_CR = 0xE0  # Connection Request
     COTP_CC = 0xD0  # Connection Confirm
     COTP_DR = 0x80  # Disconnect Request
+    COTP_DC = 0xC0  # Disconnect Confirm
     COTP_DT = 0xF0  # Data Transfer
 
+    # See ISOTCPConnection.DISCONNECT_CONFIRM_TIMEOUT (#972).
+    DISCONNECT_CONFIRM_TIMEOUT = 1.0
+
     # COTP parameter codes (ISO 8073)
     COTP_PARAM_PDU_SIZE = 0xC0
     COTP_PARAM_CALLING_TSAP = 0xC1
@@ -128,17 +132,15 @@ async def disconnect(self) -> None:
         if self._writer:
             try:
                 if self.connected:
-                    dr_pdu = struct.pack(
-                        ">BBHHBB",
-                        6,
-                        self.COTP_DR,
-                        self.dst_ref,
-                        self.src_ref,
-                        0x00,
-                        0x00,
-                    )
+                    # Class 0: no user data after the reason byte (#972).
+                    dr_pdu = struct.pack(">BBHHB", 6, self.COTP_DR, self.dst_ref, self.src_ref, 0x00)
                     self._writer.write(self._build_tpkt(dr_pdu))
                     await self._writer.drain()
+                    # Give the peer the chance to confirm the DR before our FIN arrives.
+                    await self._await_disconnect_confirm()
+            except Exception:
+                pass
+            try:
                 self._writer.close()
                 await self._writer.wait_closed()
             except Exception:
@@ -149,6 +151,34 @@ async def disconnect(self) -> None:
                 self.connected = False
                 logger.info(f"Disconnected from {self.host}:{self.port}")
 
+    async def _await_disconnect_confirm(self) -> None:
+        """Wait, briefly, for the peer to confirm the DR (a COTP DC) or close the connection."""
+        if self._reader is None:
+            return
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + min(self.timeout, self.DISCONNECT_CONFIRM_TIMEOUT)
+        buffer = b""
+        while True:
+            remaining = deadline - loop.time()
+            if remaining <= 0:
+                return
+            try:
+                chunk = await asyncio.wait_for(self._reader.read(1024), timeout=remaining)
+            except (asyncio.TimeoutError, OSError):
+                return
+            if not chunk:
+                return  # the peer closed the connection
+            buffer += chunk
+            while len(buffer) >= 6:
+                length = struct.unpack(">H", buffer[2:4])[0]
+                if buffer[0] != 3 or length < 6:
+                    return
+                if len(buffer) < length:
+                    break
+                if buffer[5] == self.COTP_DC:
+                    return
+                buffer = buffer[length:]
+
     async def send_data(self, data: bytes) -> None:
         """Send data over ISO connection."""
         if not self.connected or self._writer is None:

snap7/connection.py

@@ -9,6 +9,7 @@
 import socket
 import struct
 import logging
+import time
 from enum import IntEnum
 from typing import Optional, Type, Union
 from types import TracebackType
@@ -66,6 +67,11 @@ class ISOTCPConnection:
     COTP_PARAM_SUBNET_ID = 0xC6
     COTP_PARAM_ROUTING_TSAP = 0xC7
 
+    # How long disconnect() waits for the peer to answer the DR (a DC, or closing the
+    # connection) before it closes the socket itself. A CP 343-1 that received our FIN
+    # right behind the DR reset the connection and never freed its resource (#972).
+    DISCONNECT_CONFIRM_TIMEOUT = 1.0
+
     def __init__(
         self,
         host: str,
@@ -155,12 +161,17 @@ def disconnect(self) -> None:
         if self.socket:
             try:
                 if self.connected:
-                    # Send COTP disconnect request
+                    # Send the COTP disconnect request, and give the peer the chance to
+                    # confirm it before our FIN arrives.
                     self._send_cotp_disconnect()
-                self.socket.close()
+                    self._await_disconnect_confirm()
             except Exception:
                 pass  # Ignore errors during disconnect
             finally:
+                try:
+                    self.socket.close()
+                except Exception:
+                    pass
                 self.socket = None
                 self.connected = False
                 logger.info(f"Disconnected from {self.host}:{self.port}")
@@ -436,22 +447,62 @@ def _send_cotp_disconnect(self) -> None:
         if self.socket is None:
             return  # Nothing to disconnect
 
-        dr_pdu = struct.pack(
-            ">BBHHBB",
-            6,  # PDU length
-            self.COTP_DR,  # PDU type
-            self.dst_ref,  # Destination reference
-            self.src_ref,  # Source reference
-            0x00,  # Reason (normal disconnect)
-            0x00,  # Additional info
-        )
-
-        tpkt_frame = self._build_tpkt(dr_pdu)
+        tpkt_frame = self._build_tpkt(self._build_cotp_dr())
         try:
             self.socket.sendall(tpkt_frame)
         except socket.error:
             pass  # Ignore errors during disconnect
 
+    def _build_cotp_dr(self) -> bytes:
+        """Build a class 0 COTP Disconnect Request.
+
+        The length indicator covers the PDU type, both references and the reason, and
+        nothing follows: class 0 allows no user data in a DR. Up to 3.4.0 a trailing
+        zero byte was sent after the header (#972).
+        """
+        return struct.pack(
+            ">BBHHB",
+            6,  # Length indicator
+            self.COTP_DR,  # PDU type
+            self.dst_ref,  # Destination reference: the peer's, from its CC
+            self.src_ref,  # Source reference: ours
+            0x00,  # Reason: normal disconnect
+        )
+
+    def _await_disconnect_confirm(self) -> None:
+        """Wait, briefly, for the peer to confirm the DR or close the connection.
+
+        Returns on a COTP DC, on the peer closing or resetting the connection, or when
+        :attr:`DISCONNECT_CONFIRM_TIMEOUT` (at most the connection timeout) has passed.
+        Any other data still in flight is discarded.
+        """
+        if self.socket is None:
+            return
+        deadline = time.monotonic() + min(self.timeout, self.DISCONNECT_CONFIRM_TIMEOUT)
+        buffer = b""
+        while True:
+            remaining = deadline - time.monotonic()
+            if remaining <= 0:
+                return
+            self.socket.settimeout(remaining)
+            try:
+                chunk = self.socket.recv(1024)
+            except (socket.timeout, OSError):
+                return
+            if not chunk:
+                return  # the peer closed the connection
+            buffer += chunk
+            # Scan complete TPKT frames for a COTP DC.
+            while len(buffer) >= 6:
+                length = struct.unpack(">H", buffer[2:4])[0]
+                if buffer[0] != 3 or length < 6:
+                    return  # not a TPKT stream we can follow; stop waiting
+                if len(buffer) < length:
+                    break
+                if buffer[5] == self.COTP_DC:
+                    return
+                buffer = buffer[length:]
+
     def _recv_exact(self, size: int) -> bytes:
         """
         Receive exactly the specified number of bytes.

tests/test_async_client.py

@@ -355,7 +355,42 @@ async def test_async_disconnect_request_addresses_the_peer_reference() -> None:
 
     await connection.disconnect()
 
-    writer.write.assert_called_once_with(bytes.fromhex("03 00 00 0c 06 80 4431 0001 00 00"))
+    writer.write.assert_called_once_with(bytes.fromhex("03 00 00 0b 06 80 4431 0001 00"))
+
+
+@pytest.mark.asyncio
+async def test_async_disconnect_waits_for_the_dc_before_closing() -> None:
+    """The FIN goes out only after the peer has answered the DR (#972)."""
+    seen: dict[str, bytes] = {}
+    done = asyncio.Event()
+
+    async def peer(reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
+        seen["dr"] = await reader.read(64)
+        try:
+            seen["early"] = await asyncio.wait_for(reader.read(64), timeout=0.2)
+        except asyncio.TimeoutError:
+            seen["early"] = b"<none>"
+        writer.write(bytes.fromhex("03 00 00 09 05 c0 0001 4431"))
+        await writer.drain()
+        seen["after"] = await reader.read(64)
+        writer.close()
+        done.set()
+
+    server = await asyncio.start_server(peer, "127.0.0.1", 0)
+    port = server.sockets[0].getsockname()[1]
+    connection = AsyncISOTCPConnection("127.0.0.1", port)
+    connection._reader, connection._writer = await asyncio.open_connection("127.0.0.1", port)
+    connection.connected = True
+    connection.dst_ref = 0x4431
+
+    await connection.disconnect()
+    await asyncio.wait_for(done.wait(), timeout=5.0)
+    server.close()
+    await server.wait_closed()
+
+    assert seen["dr"] == bytes.fromhex("03 00 00 0b 06 80 4431 0001 00")
+    assert seen["early"] == b"<none>"
+    assert seen["after"] == b""
 
 
 @pytest.mark.parametrize(

tests/test_connection.py

@@ -2,6 +2,8 @@
 
 import socket
 import struct
+import threading
+import time
 import pytest
 from unittest.mock import patch, MagicMock
 
@@ -89,8 +91,9 @@ def test_valid_cc(self) -> None:
     # COTP CC from a CP 343-1 in front of an S7-313C-2 DP, captured for #972 (after the
     # TPKT header): dst-ref 0x0001 echoes our reference, src-ref 0x4431 is the CP's.
     CAPTURED_CC = bytes.fromhex("11 d0 0001 4431 00 c0 01 0a c1 02 01 00 c2 02 01 02")
-    # The DR must address the CP's reference; python-snap7 3.3.1 sent 06 80 0001 0001 00 00.
-    EXPECTED_DR = bytes.fromhex("03 00 00 0c 06 80 4431 0001 00 00")
+    # The DR must address the CP's reference; python-snap7 3.3.1 sent 06 80 0001 0001 00 00,
+    # and up to 3.4.0 a stray user-data byte followed the reason.
+    EXPECTED_DR = bytes.fromhex("03 00 00 0b 06 80 4431 0001 00")
 
     def test_captured_cc_gives_the_peer_reference(self) -> None:
         conn = ISOTCPConnection("1.2.3.4")
@@ -387,6 +390,74 @@ def test_disconnect_ignores_errors(self) -> None:
         assert conn.socket is None
 
 
+class TestDisconnectWaitsForThePeer:
+    """disconnect() lets the peer answer the DR before the FIN goes out (#972)."""
+
+    DC = bytes.fromhex("03 00 00 09 05 c0 0001 4431")
+
+    @staticmethod
+    def _connected_pair() -> tuple[ISOTCPConnection, socket.socket]:
+        ours, peer = socket.socketpair()
+        conn = ISOTCPConnection("1.2.3.4")
+        conn.socket = ours
+        conn.connected = True
+        conn.dst_ref = 0x4431
+        return conn, peer
+
+    def test_no_fin_before_the_dc(self) -> None:
+        conn, peer = self._connected_pair()
+        seen: dict[str, bytes] = {}
+
+        def answer() -> None:
+            peer.settimeout(2.0)
+            seen["dr"] = peer.recv(64)
+            # Nothing else may arrive while the DR is unanswered: recv() would return b"" on a FIN.
+            peer.settimeout(0.2)
+            try:
+                seen["early"] = peer.recv(64)
+            except socket.timeout:
+                seen["early"] = b"<none>"
+            peer.sendall(self.DC)
+            peer.settimeout(2.0)
+            seen["after"] = peer.recv(64)
+
+        thread = threading.Thread(target=answer)
+        thread.start()
+        conn.disconnect()
+        thread.join(5.0)
+        peer.close()
+
+        assert seen["dr"] == bytes.fromhex("03 00 00 0b 06 80 4431 0001 00")
+        assert seen["early"] == b"<none>"
+        assert seen["after"] == b""  # our FIN, after the DC
+        assert conn.socket is None and conn.connected is False
+
+    def test_returns_when_the_peer_closes(self) -> None:
+        conn, peer = self._connected_pair()
+        peer.close()
+        start = time.monotonic()
+        conn.disconnect()
+        assert time.monotonic() - start < 0.5
+
+    def test_silent_peer_is_bounded(self, monkeypatch: pytest.MonkeyPatch) -> None:
+        monkeypatch.setattr(ISOTCPConnection, "DISCONNECT_CONFIRM_TIMEOUT", 0.2)
+        conn, peer = self._connected_pair()
+        start = time.monotonic()
+        conn.disconnect()
+        elapsed = time.monotonic() - start
+        peer.close()
+        assert 0.15 <= elapsed < 1.0
+
+    def test_skips_other_frames_until_the_dc(self) -> None:
+        conn, peer = self._connected_pair()
+        late_response = bytes.fromhex("03 00 00 07 02 f0 80")
+        peer.sendall(late_response + self.DC)
+        start = time.monotonic()
+        conn.disconnect()
+        peer.close()
+        assert time.monotonic() - start < 0.5
+
+
 class TestContextManager:
     """Test __enter__ / __exit__."""