From 68eb519936ac3490f575a298eea7b11800697449 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Fri, 11 Sep 2026 09:39:49 +0200 Subject: [PATCH 1/2] fix(s7commplus): correlate request responses --- CHANGES.md | 2 + s7commplus/async_client.py | 52 +++++----- s7commplus/connection.py | 92 ++++++++++++------ tests/test_s7_alarm.py | 3 + tests/test_s7_v2.py | 188 +++++++++++++++++++++++++++++++++++++ 5 files changed, 287 insertions(+), 50 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index f664baeb..81d83f71 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -6,6 +6,8 @@ CHANGES Major release: new `s7commplus` package with S7CommPlus protocol support. +* Correlate S7CommPlus responses by opcode, function, and sequence; preserve + interleaved notifications and serialize synchronous wire requests. * Decode corroborating CPU execution attributes so S7CommPlus `get_cpu_state()` distinguishes RUN from STOP on S7-1500 and returns UNKNOWN for absent or inconsistent state attributes, including S7-1200 responses that omit them. diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 1a54774f..d33afe56 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -7,6 +7,7 @@ import logging import ssl import struct +from collections import deque from collections.abc import Sequence from typing import Any, Awaitable, Callable, Optional, TypeVar @@ -49,10 +50,12 @@ _build_set_variable_payload, _check_system_event, _check_set_variable_response, + _incoming_frame_opcode, _log_create_object_return_value, _parse_get_var_substreamed_response, _parse_protection_level_response, _set_s7_groups, + _validate_response_header, ) from .alarm import ( Alarm, @@ -115,6 +118,7 @@ def __init__(self) -> None: self._session_ready = False self._connected = False self._lock = asyncio.Lock() + self._notification_frames: deque[bytes] = deque() self._connect_params: Optional[dict[str, Any]] = None # V2+ IntegrityId tracking @@ -506,6 +510,7 @@ async def disconnect(self) -> None: self._server_session_version = None self._session_setup_ok = False self._protection_level = None + self._notification_frames.clear() if self._writer: try: @@ -737,8 +742,11 @@ async def receive_alarm_notification( async with self._lock: if not self._connected: raise RuntimeError("Not connected") - receive = self._recv_cotp_dt() - frame = await asyncio.wait_for(receive, timeout) if timeout is not None else await receive + if self._notification_frames: + frame = self._notification_frames.popleft() + else: + receive = self._recv_cotp_dt() + frame = await asyncio.wait_for(receive, timeout) if timeout is not None else await receive return parse_alarm_notification(frame, language_ids) async def read_alarms(self, language_ids: Optional[list[LanguageId | int]] = None) -> list[Alarm]: @@ -966,40 +974,42 @@ async def _send_request( data = await self._recv_reassembled_payload(response_data) if len(data) < 10: raise S7ConnectionError("Response too short") - resp_func = struct.unpack_from(">H", data, 3)[0] - resp_seq = struct.unpack_from(">H", data, 7)[0] - if resp_seq != seq_num: - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) + _validate_response_header(data, function_code, seq_num) return bytes(data[10:]) _, data_length, consumed = decode_header(response_data) response = response_data[consumed : consumed + data_length] - if len(response) < 10: - raise S7ConnectionError("Response too short") + _validate_response_header(response, function_code, seq_num) # RESPONSE header is 10 bytes (opcode+res+func+res+seqnr+transport) — responses # carry no SessionId field (requests do, hence their 14-byte header). For V2+ the # IntegrityId travels at the END of the payload and is ignored by the parsers. - resp_func = struct.unpack_from(">H", response, 3)[0] - resp_seq = struct.unpack_from(">H", response, 7)[0] - if resp_seq != seq_num: - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) return response[10:] async def _recv_response_frame(self) -> bytes: - """Receive the next application response, consuming non-fatal SystemEvents.""" - for _ in range(_MAX_SYSTEM_EVENTS_PER_RESPONSE + 1): + """Receive the next response, queueing unsolicited application frames.""" + system_events = 0 + while True: response_data = await self._recv_cotp_dt() + if not response_data: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus response") version, data_length, consumed = decode_header(response_data) - if version != ProtocolVersion.SYSTEM_EVENT: + if version == ProtocolVersion.SYSTEM_EVENT: + _check_system_event(bytes(response_data[consumed : consumed + data_length])) + system_events += 1 + if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE: + raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") + continue + if data_length < 10: return response_data - _check_system_event(bytes(response_data[consumed : consumed + data_length])) - raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") + opcode = _incoming_frame_opcode(response_data) + if opcode == Opcode.NOTIFICATION: + self._notification_frames.append(response_data) + continue + if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): + raise S7ProtocolError(f"Unexpected S7CommPlus opcode 0x{opcode:02X} while waiting for a response") + return response_data async def _recv_reassembled_payload(self, initial_data: bytes = b"") -> bytes: """Receive a possibly-fragmented S7CommPlus response, returning its data section. diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 39a73833..be951194 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -45,6 +45,7 @@ import ssl import struct import tempfile +import threading from collections import deque from types import TracebackType from typing import Any, Optional, Type @@ -80,6 +81,47 @@ logger = logging.getLogger(__name__) +def _incoming_frame_opcode(frame: bytes) -> int: + """Return the application opcode from a complete non-SystemEvent frame.""" + from snap7.error import S7ConnectionError, S7ProtocolError + + if not frame: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus frame") + version, data_length, consumed = decode_header(frame) + data = bytes(frame[consumed : consumed + data_length]) + if version == ProtocolVersion.V3 and data: + hash_length = data[0] + if len(data) <= 1 + hash_length: + raise S7ProtocolError("Truncated S7CommPlus V3 integrity envelope") + data = data[1 + hash_length :] + if not data: + raise S7ProtocolError("S7CommPlus frame has no application opcode") + return data[0] + + +def _validate_response_header(response: bytes, expected_function: int, expected_sequence: int) -> None: + """Validate that application data is the response to one outstanding request.""" + from snap7.error import S7ConnectionError, S7ProtocolError + + if len(response) < 10: + raise S7ConnectionError("Response too short") + opcode = response[0] + function = struct.unpack_from(">H", response, 3)[0] + sequence = struct.unpack_from(">H", response, 7)[0] + if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): + raise S7ProtocolError(f"Unexpected response opcode 0x{opcode:02X}") + # A PLC may answer any failed request with the protocol's generic ERROR + # function while retaining the request sequence number. + if function not in (expected_function, FunctionCode.ERROR): + raise S7ProtocolError( + f"Response function mismatch: expected function=0x{expected_function:04X}, got function=0x{function:04X}" + ) + if sequence != expected_sequence: + raise S7ProtocolError( + f"Response sequence mismatch: expected seq={expected_sequence}, got seq={sequence} for function=0x{function:04X}" + ) + + def _log_create_object_return_value(return_value: int, tls_active: bool) -> None: """Log a non-zero CreateObject status without guessing at TLS requirements.""" if return_value == 0: @@ -478,6 +520,7 @@ def __init__( # Password for post-auth legitimation (V1-initial PLCs) self._connect_password: str = "" self._notification_frames: deque[bytes] = deque() + self._request_lock = threading.Lock() # Effective protection level, read once the session is up self._protection_level: Optional[int] = None @@ -876,6 +919,11 @@ def disconnect(self) -> None: self._iso_conn.disconnect() def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: int = 4, reassemble: bool = False) -> bytes: + """Serialize one request/response exchange on the connection.""" + with self._request_lock: + return self._send_request(function_code, payload, integrity_tail, reassemble) + + def _send_request(self, function_code: int, payload: bytes, integrity_tail: int, reassemble: bool) -> bytes: """Send an S7CommPlus request and receive the response. For V2+ with IntegrityId tracking enabled, the IntegrityId is spliced into @@ -975,14 +1023,7 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: from snap7.error import S7ConnectionError raise S7ConnectionError("Response too short") - resp_func = struct.unpack_from(">H", data, 3)[0] - resp_seq = struct.unpack_from(">H", data, 7)[0] - if resp_seq != seq_num: - from snap7.error import S7ProtocolError - - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) + _validate_response_header(data, function_code, seq_num) logger.debug(f" Reassembled response ({len(data)} bytes), payload {len(data) - 10} bytes") resp_payload = bytes(data[10:]) if self._session_key is not None: @@ -1010,10 +1051,7 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: logger.debug(f" Response data ({len(response)} bytes): {response.hex(' ')}") - if len(response) < 10: - from snap7.error import S7ConnectionError - - raise S7ConnectionError("Response too short") + _validate_response_header(response, function_code, seq_num) # Parse the 10-byte response header for debug (responses carry no SessionId) resp_opcode = response[0] @@ -1024,13 +1062,6 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: f" Response header: opcode=0x{resp_opcode:02X} function=0x{resp_func:04X} " f"seq={resp_seq} transport=0x{resp_transport:02X}" ) - if resp_seq != seq_num: - from snap7.error import S7ProtocolError - - raise S7ProtocolError( - f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}" - ) - # RESPONSE header is 10 bytes (opcode+res+func+res+seqnr+transport) — responses have # NO SessionId field (requests do, making their header 14 bytes). resp_offset = 10 @@ -1055,11 +1086,13 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: def _recv_response_frame(self) -> bytes: """Receive the next response, queueing notifications and consuming non-fatal SystemEvents.""" - from snap7.error import S7ProtocolError + from snap7.error import S7ConnectionError, S7ProtocolError system_events = 0 while True: response_frame = self._recv_s7_data() + if not response_frame: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus response") version, data_length, consumed = decode_header(response_frame) if version == ProtocolVersion.SYSTEM_EVENT: _check_system_event(bytes(response_frame[consumed : consumed + data_length])) @@ -1067,24 +1100,25 @@ def _recv_response_frame(self) -> bytes: if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE: raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") continue - if self._is_notification_frame(response_frame): + if data_length < 10: + return response_frame + opcode = _incoming_frame_opcode(response_frame) + if opcode == Opcode.NOTIFICATION: self._notification_frames.append(response_frame) continue + if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): + raise S7ProtocolError(f"Unexpected S7CommPlus opcode 0x{opcode:02X} while waiting for a response") return response_frame @staticmethod def _is_notification_frame(frame: bytes) -> bool: """Return whether a complete frame contains an unsolicited notification.""" + from snap7.error import S7ConnectionError, S7ProtocolError + try: - version, data_length, consumed = decode_header(frame) - except (IndexError, ValueError): + return _incoming_frame_opcode(frame) == Opcode.NOTIFICATION + except (IndexError, ValueError, S7ConnectionError, S7ProtocolError): return False - data = frame[consumed : consumed + data_length] - if version == ProtocolVersion.V3 and data: - hash_length = data[0] - if hash_length and len(data) > 1 + hash_length: - data = data[1 + hash_length :] - return bool(data) and data[0] == Opcode.NOTIFICATION def receive_notification(self) -> bytes: """Receive one unsolicited S7CommPlus notification frame. diff --git a/tests/test_s7_alarm.py b/tests/test_s7_alarm.py index d5b2f157..ade11a6a 100644 --- a/tests/test_s7_alarm.py +++ b/tests/test_s7_alarm.py @@ -203,6 +203,9 @@ async def test_async_alarm_client_apis() -> None: assert await client.create_alarm_subscription([1031]) == 0x55667788 assert (await client.read_alarms([1031]))[0].texts[1031].alarm_text == "Alarm 4 =F6+S2-G1" assert (await client.receive_alarm_notification(timeout=1)).credit_tick == 5 + client._notification_frames.append(_notification_frame()) + assert (await client.receive_alarm_notification(timeout=1)).sequence_number == 12 + client._recv_cotp_dt.assert_awaited_once() await client.delete_alarm_subscription(0x55667788) diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index 9b70719e..b1d4ed18 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -9,6 +9,7 @@ import hmac import logging import struct +import threading from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -522,6 +523,193 @@ async def test_async_sequence_mismatch_raises_protocol_error(self, reassemble: b with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + @pytest.mark.parametrize("reassemble", [False, True]) + def test_sync_function_mismatch_raises_protocol_error(self, reassemble: bool) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.SET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Response function mismatch"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + + @pytest.mark.asyncio + @pytest.mark.parametrize("reassemble", [False, True]) + async def test_async_function_mismatch_raises_protocol_error(self, reassemble: bool) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.SET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Response function mismatch"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + + @pytest.mark.parametrize("opcode", [Opcode.REQUEST, 0x7F]) + def test_sync_unexpected_opcode_raises_protocol_error(self, opcode: int) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", opcode, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Unexpected S7CommPlus opcode"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + + @pytest.mark.asyncio + async def test_async_unexpected_opcode_raises_protocol_error(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.REQUEST, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Unexpected S7CommPlus opcode"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + + def test_sync_notification_before_response_is_queued(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + notification = struct.pack(">BHHHHB", Opcode.NOTIFICATION, 0, 0, 0, 12, 0x34) + notification_frame = encode_header(ProtocolVersion.V2, len(notification)) + notification + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(side_effect=[notification_frame, response_frame]) + + assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES) == b"" + assert conn.receive_notification() == notification_frame + assert conn._recv_s7_data.call_count == 2 + + @pytest.mark.asyncio + async def test_async_notification_before_response_is_queued(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + notification = struct.pack(">BHHHHB", Opcode.NOTIFICATION, 0, 0, 0, 12, 0x34) + notification_frame = encode_header(ProtocolVersion.V2, len(notification)) + notification + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(side_effect=[notification_frame, response_frame]) + + assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" + assert list(client._notification_frames) == [notification_frame] + + def test_duplicate_sync_response_cannot_satisfy_next_request(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=response_frame) + + assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES) == b"" + with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + + @pytest.mark.asyncio + async def test_duplicate_async_response_cannot_satisfy_next_request(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=response_frame) + + assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" + with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + + def test_sync_requests_are_serialized(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + first_entered = threading.Event() + release_first = threading.Event() + second_started = threading.Event() + second_entered = threading.Event() + call_count = 0 + count_lock = threading.Lock() + + def exchange(*_args: object) -> bytes: + nonlocal call_count + with count_lock: + call_count += 1 + current = call_count + if current == 1: + first_entered.set() + assert release_first.wait(1) + else: + second_entered.set() + return b"" + + conn._send_request = MagicMock(side_effect=exchange) + first = threading.Thread(target=conn.send_request, args=(FunctionCode.GET_MULTI_VARIABLES,)) + + def run_second() -> None: + second_started.set() + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + + second = threading.Thread(target=run_second) + first.start() + assert first_entered.wait(1) + second.start() + assert second_started.wait(1) + assert not second_entered.wait(0.05) + release_first.set() + first.join(1) + second.join(1) + assert not first.is_alive() + assert not second.is_alive() + assert second_entered.is_set() + + @pytest.mark.parametrize("client_kind", ["sync", "async"]) + @pytest.mark.asyncio + async def test_connection_close_while_waiting_is_connection_error(self, client_kind: str) -> None: + if client_kind == "sync": + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V2 + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=b"") + with pytest.raises(S7ConnectionError, match="Connection closed"): + conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + return + + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=b"") + with pytest.raises(S7ConnectionError, match="Connection closed"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + class TestAsyncReassembledPayloadErrors: @pytest.mark.asyncio From 652a62c9167b50b97b8a8e9bc08590a5aa69ef37 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Fri, 11 Sep 2026 15:57:40 +0200 Subject: [PATCH 2/2] fix(s7commplus): skip stale response frames --- CHANGES.md | 5 +++-- s7commplus/async_client.py | 24 ++++++++++++++++++--- s7commplus/connection.py | 44 ++++++++++++++++++++++++++++++++++++-- tests/test_s7_v2.py | 22 ++++++++++++------- 4 files changed, 80 insertions(+), 15 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 81d83f71..5c2b789f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -6,8 +6,9 @@ CHANGES Major release: new `s7commplus` package with S7CommPlus protocol support. -* Correlate S7CommPlus responses by opcode, function, and sequence; preserve - interleaved notifications and serialize synchronous wire requests. +* Correlate S7CommPlus responses by opcode, function, and sequence; discard + bounded stale replies from earlier requests, preserve interleaved + notifications, and serialize synchronous wire requests. * Decode corroborating CPU execution attributes so S7CommPlus `get_cpu_state()` distinguishes RUN from STOP on S7-1500 and returns UNKNOWN for absent or inconsistent state attributes, including S7-1200 responses that omit them. diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index d33afe56..22a47a02 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -44,13 +44,16 @@ parse_server_session_version, ) from .connection import ( + _MAX_STALE_RESPONSES_PER_REQUEST, _MAX_SYSTEM_EVENTS_PER_RESPONSE, _S7_CIPHERS, _build_get_var_substreamed_payload, _build_set_variable_payload, - _check_system_event, _check_set_variable_response, + _check_system_event, _incoming_frame_opcode, + _incoming_response_sequence, + _is_stale_response_sequence, _log_create_object_return_value, _parse_get_var_substreamed_response, _parse_protection_level_response, @@ -967,7 +970,7 @@ async def _send_request( else: self._integrity_id_write = (self._integrity_id_write + 1) & 0xFFFFFFFF - response_data = await self._recv_response_frame() + response_data = await self._recv_response_frame(seq_num) # Large responses (e.g. Explore) are split across several S7CommPlus PDUs. if reassemble: @@ -987,9 +990,10 @@ async def _send_request( # IntegrityId travels at the END of the payload and is ignored by the parsers. return response[10:] - async def _recv_response_frame(self) -> bytes: + async def _recv_response_frame(self, expected_sequence: Optional[int] = None) -> bytes: """Receive the next response, queueing unsolicited application frames.""" system_events = 0 + stale_responses = 0 while True: response_data = await self._recv_cotp_dt() if not response_data: @@ -1009,6 +1013,20 @@ async def _recv_response_frame(self) -> bytes: continue if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): raise S7ProtocolError(f"Unexpected S7CommPlus opcode 0x{opcode:02X} while waiting for a response") + if expected_sequence is not None: + sequence = _incoming_response_sequence(response_data) + if _is_stale_response_sequence(sequence, expected_sequence): + stale_responses += 1 + logger.warning( + "Ignoring stale S7CommPlus response sequence %d while waiting for sequence %d", + sequence, + expected_sequence, + ) + if stale_responses > _MAX_STALE_RESPONSES_PER_REQUEST: + raise S7ProtocolError( + f"Too many stale S7CommPlus responses while waiting for sequence {expected_sequence}" + ) + continue return response_data async def _recv_reassembled_payload(self, initial_data: bytes = b"") -> bytes: diff --git a/s7commplus/connection.py b/s7commplus/connection.py index be951194..bb80879d 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -99,6 +99,30 @@ def _incoming_frame_opcode(frame: bytes) -> int: return data[0] +def _incoming_response_sequence(frame: bytes) -> int: + """Return the sequence number from a complete response frame.""" + from snap7.error import S7ConnectionError, S7ProtocolError + + if not frame: + raise S7ConnectionError("Connection closed while waiting for an S7CommPlus frame") + version, data_length, consumed = decode_header(frame) + data = bytes(frame[consumed : consumed + data_length]) + if version == ProtocolVersion.V3 and data: + hash_length = data[0] + if len(data) <= 1 + hash_length: + raise S7ProtocolError("Truncated S7CommPlus V3 integrity envelope") + data = data[1 + hash_length :] + if len(data) < 10: + raise S7ConnectionError("Response too short") + return struct.unpack_from(">H", data, 7)[0] + + +def _is_stale_response_sequence(sequence: int, expected_sequence: int) -> bool: + """Return whether a 16-bit response sequence precedes the expected one.""" + distance = (expected_sequence - sequence) & 0xFFFF + return 0 < distance < 0x8000 + + def _validate_response_header(response: bytes, expected_function: int, expected_sequence: int) -> None: """Validate that application data is the response to one outstanding request.""" from snap7.error import S7ConnectionError, S7ProtocolError @@ -164,6 +188,7 @@ def _log_create_object_return_value(return_value: int, tls_active: bool) -> None _S7_PREFERRED_GROUPS = ("X25519",) _MAX_SYSTEM_EVENTS_PER_RESPONSE = 16 +_MAX_STALE_RESPONSES_PER_REQUEST = 16 _SYSTEM_EVENT_RETURN_VALUE_ID = 40305 @@ -1014,7 +1039,7 @@ def _send_request(self, function_code: int, payload: bytes, integrity_tail: int, else: self._integrity_id_write = (self._integrity_id_write + 1) & 0xFFFFFFFF - response_frame = self._recv_response_frame() + response_frame = self._recv_response_frame(seq_num) # Large responses (e.g. Explore) are split across several S7CommPlus PDUs. if reassemble: @@ -1084,11 +1109,12 @@ def _send_request(self, function_code: int, payload: bytes, integrity_tail: int, return resp_payload - def _recv_response_frame(self) -> bytes: + def _recv_response_frame(self, expected_sequence: Optional[int] = None) -> bytes: """Receive the next response, queueing notifications and consuming non-fatal SystemEvents.""" from snap7.error import S7ConnectionError, S7ProtocolError system_events = 0 + stale_responses = 0 while True: response_frame = self._recv_s7_data() if not response_frame: @@ -1108,6 +1134,20 @@ def _recv_response_frame(self) -> bytes: continue if opcode not in (Opcode.RESPONSE, Opcode.RESPONSE2): raise S7ProtocolError(f"Unexpected S7CommPlus opcode 0x{opcode:02X} while waiting for a response") + if expected_sequence is not None: + sequence = _incoming_response_sequence(response_frame) + if _is_stale_response_sequence(sequence, expected_sequence): + stale_responses += 1 + logger.warning( + "Ignoring stale S7CommPlus response sequence %d while waiting for sequence %d", + sequence, + expected_sequence, + ) + if stale_responses > _MAX_STALE_RESPONSES_PER_REQUEST: + raise S7ProtocolError( + f"Too many stale S7CommPlus responses while waiting for sequence {expected_sequence}" + ) + continue return response_frame @staticmethod diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index b1d4ed18..bb654789 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -616,7 +616,7 @@ async def test_async_notification_before_response_is_queued(self) -> None: assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" assert list(client._notification_frames) == [notification_frame] - def test_duplicate_sync_response_cannot_satisfy_next_request(self) -> None: + def test_stale_sync_response_is_skipped_before_next_response(self) -> None: conn = S7CommPlusConnection("127.0.0.1") conn._connected = True conn._protocol_version = ProtocolVersion.V2 @@ -624,14 +624,17 @@ def test_duplicate_sync_response_cannot_satisfy_next_request(self) -> None: response_frame = encode_header(ProtocolVersion.V2, len(response)) + response response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) conn._send_s7_data = MagicMock() - conn._recv_s7_data = MagicMock(return_value=response_frame) + next_response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 1, 0x34) + next_response_frame = encode_header(ProtocolVersion.V2, len(next_response)) + next_response + next_response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + conn._recv_s7_data = MagicMock(side_effect=[response_frame, response_frame, next_response_frame]) assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES) == b"" - with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): - conn.send_request(FunctionCode.GET_MULTI_VARIABLES) + assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES) == b"" + assert conn._recv_s7_data.call_count == 3 @pytest.mark.asyncio - async def test_duplicate_async_response_cannot_satisfy_next_request(self) -> None: + async def test_stale_async_response_is_skipped_before_next_response(self) -> None: client = S7CommPlusAsyncClient() client._connected = True client._reader = MagicMock() @@ -640,11 +643,14 @@ async def test_duplicate_async_response_cannot_satisfy_next_request(self) -> Non response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) response_frame = encode_header(ProtocolVersion.V2, len(response)) + response client._send_cotp_dt = AsyncMock() - client._recv_cotp_dt = AsyncMock(return_value=response_frame) + next_response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 1, 0x34) + next_response_frame = encode_header(ProtocolVersion.V2, len(next_response)) + next_response + next_response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + client._recv_cotp_dt = AsyncMock(side_effect=[response_frame, response_frame, next_response_frame]) assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" - with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): - await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") == b"" + assert client._recv_cotp_dt.await_count == 3 def test_sync_requests_are_serialized(self) -> None: conn = S7CommPlusConnection("127.0.0.1")