fix(s7commplus): correlate request responses

CHANGES.md

@@ -6,6 +6,9 @@ CHANGES
 
 Major release: new `s7commplus` package with S7CommPlus protocol support.
 
+* 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.

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
 
@@ -43,16 +44,21 @@
     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,
     _set_s7_groups,
+    _validate_response_header,
 )
 from .alarm import (
     Alarm,
@@ -115,6 +121,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 +513,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 +745,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]:
@@ -959,47 +970,64 @@ 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:
                 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):
+    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:
+                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")
+            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:
         """Receive a possibly-fragmented S7CommPlus response, returning its data section.

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,71 @@
 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 _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
+
+    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:
@@ -122,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
 
 
@@ -478,6 +545,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 +944,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
@@ -966,7 +1039,7 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
             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:
@@ -975,14 +1048,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 +1076,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 +1087,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
@@ -1053,38 +1109,56 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
 
         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 S7ProtocolError
+        from snap7.error import S7ConnectionError, S7ProtocolError
 
         system_events = 0
+        stale_responses = 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]))
                 system_events += 1
                 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")
+            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
     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.

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)