From db0747e2652b5fb526f10dfd5656bb2eb02ea777 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Wed, 11 Nov 2026 10:52:51 +0200 Subject: [PATCH] Wait for the peer to confirm the disconnect request A CP 343-1 kept the connection's resource after disconnect() even with the DR addressed to its reference (#979): the capture shows our FIN 7 us after the DR and the CP answering both with a reset, never with a DC. disconnect() now waits up to one second for a DC, or for the peer to close or reset the connection, before closing the socket, in both clients. The DR also carried a zero byte after its header, outside the length indicator. Class 0 allows no user data in a DR, so it is dropped. Refs #972. Co-Authored-By: Claude Opus 5.5 --- CHANGES.md | 4 ++ snap7/async_client.py | 48 +++++++++++++++++++----- snap7/connection.py | 77 +++++++++++++++++++++++++++++++------- tests/test_async_client.py | 37 +++++++++++++++++- tests/test_connection.py | 75 ++++++++++++++++++++++++++++++++++++- 5 files changed, 216 insertions(+), 25 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 2c652f91..601f3645 100644 --- a/CHANGES.md +++ b/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 ----- diff --git a/snap7/async_client.py b/snap7/async_client.py index 1e38b6ea..cf11af22 100644 --- a/snap7/async_client.py +++ b/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: diff --git a/snap7/connection.py b/snap7/connection.py index f6893543..f5f82533 100644 --- a/snap7/connection.py +++ b/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. diff --git a/tests/test_async_client.py b/tests/test_async_client.py index 55d959d7..f30b3f1f 100644 --- a/tests/test_async_client.py +++ b/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"" + 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"" + assert seen["after"] == b"" @pytest.mark.parametrize( diff --git a/tests/test_connection.py b/tests/test_connection.py index 8c98b9e0..5f381bf1 100644 --- a/tests/test_connection.py +++ b/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"" + 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"" + 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__."""