fix(s7commplus): restore V1 challenge wire layout

CHANGES.md

@@ -12,6 +12,9 @@ Major release: new `s7commplus` package with S7CommPlus protocol support.
 * S7CommPlus PLC start/stop via INVOKE
 * S7CommPlus object browsing via EXPLORE
 * S7CommPlus live symbol browsing (`client.browse()`) and datablock listing (experimental)
+* Fix V1 SessionKey challenge requests being rejected by S7-1200 FW 4.2 PLCs,
+  consume non-fatal SystemEvents while waiting for the matching response, and
+  strip per-fragment V3 HMACs from browse responses (#710)
 * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental)
 * S7CommPlus CPU state reading and block transfer (upload/download)
 * **Symbolic (LID-based) access for optimized DBs** (experimental):

s7commplus/async_client.py

@@ -36,9 +36,11 @@
     parse_server_session_version,
 )
 from .connection import (
+    _MAX_SYSTEM_EVENTS_PER_RESPONSE,
     _S7_CIPHERS,
     _build_get_var_substreamed_payload,
     _build_set_variable_payload,
+    _check_system_event,
     _check_set_variable_response,
     _parse_get_var_substreamed_response,
     _parse_protection_level_response,
@@ -764,16 +766,22 @@ async def _send_request(
                 else:
                     self._integrity_id_write = (self._integrity_id_write + 1) & 0xFFFFFFFF
 
+            response_data = await self._recv_response_frame()
+
             # Large responses (e.g. Explore) are split across several S7CommPlus PDUs.
             if reassemble:
-                data = await self._recv_reassembled_payload()
+                data = await self._recv_reassembled_payload(response_data)
                 if len(data) < 10:
                     raise RuntimeError("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 RuntimeError(
+                        f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}"
+                    )
                 return bytes(data[10:])
 
-            response_data = await self._recv_cotp_dt()
-
-            version, data_length, consumed = decode_header(response_data)
+            _, data_length, consumed = decode_header(response_data)
             response = response_data[consumed : consumed + data_length]
 
             if len(response) < 10:
@@ -782,9 +790,25 @@ async def _send_request(
             # 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 RuntimeError(
+                    f"Response sequence mismatch: expected seq={seq_num}, got seq={resp_seq} for function=0x{resp_func:04X}"
+                )
             return response[10:]
 
-    async def _recv_reassembled_payload(self) -> bytes:
+    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):
+            response_data = await self._recv_cotp_dt()
+            version, data_length, consumed = decode_header(response_data)
+            if version != ProtocolVersion.SYSTEM_EVENT:
+                return response_data
+            _check_system_event(bytes(response_data[consumed : consumed + data_length]))
+        raise RuntimeError("Too many S7CommPlus SystemEvents while waiting for a response")
+
+    async def _recv_reassembled_payload(self, initial_data: bytes = b"") -> bytes:
         """Receive a possibly-fragmented S7CommPlus response, returning its data section.
 
         A large response is split into several S7CommPlus PDUs. Each fragment is
@@ -793,7 +817,7 @@ async def _recv_reassembled_payload(self) -> bytes:
         of every fragment until the trailer is seen. Works for single-PDU responses
         too (one fragment immediately followed by the trailer).
         """
-        buf = bytearray()
+        buf = bytearray(initial_data)
 
         async def ensure(n: int) -> None:
             while len(buf) < n:

s7commplus/connection.py

@@ -81,6 +81,87 @@
 # but unavailable on older OpenSSL/CPython; fall back to prime256v1.
 _S7_PREFERRED_GROUPS = ("X25519", "prime256v1")
 
+_MAX_SYSTEM_EVENTS_PER_RESPONSE = 16
+_SYSTEM_EVENT_RETURN_VALUE_ID = 40305
+
+
+def _system_event_return_value(payload: bytes) -> Optional[int]:
+    """Return a fixed-width SystemEvent error value, if the event contains one."""
+    from snap7.error import S7ProtocolError
+
+    if len(payload) < 16:
+        raise S7ProtocolError(f"Malformed S7CommPlus SystemEvent: {payload.hex()}")
+    if len(payload) == 16:
+        return None
+
+    # A non-Struct suffix is an informational message (for example LOGOUT).
+    if len(payload) < 20 or int.from_bytes(payload[16:20], "big") != DataType.STRUCT:
+        return None
+    if len(payload) < 24:
+        raise S7ProtocolError(f"Malformed S7CommPlus SystemEvent Struct: {payload.hex()}")
+
+    offset = 24  # fixed-width PValue header + Struct id
+    scalar_sizes = {
+        DataType.BOOL: 1,
+        DataType.USINT: 1,
+        DataType.UINT: 2,
+        DataType.UDINT: 4,
+        DataType.ULINT: 8,
+        DataType.SINT: 1,
+        DataType.INT: 2,
+        DataType.DINT: 4,
+        DataType.LINT: 8,
+        DataType.BYTE: 1,
+        DataType.WORD: 2,
+        DataType.DWORD: 4,
+        DataType.LWORD: 8,
+        DataType.REAL: 4,
+        DataType.LREAL: 8,
+        DataType.TIMESTAMP: 8,
+        DataType.TIMESPAN: 8,
+        DataType.RID: 4,
+        DataType.AID: 4,
+    }
+
+    while offset + 4 <= len(payload):
+        member_id = int.from_bytes(payload[offset : offset + 4], "big")
+        offset += 4
+        if member_id == 0:
+            break
+        if offset + 4 > len(payload):
+            raise S7ProtocolError(f"Malformed S7CommPlus SystemEvent member: {payload.hex()}")
+
+        flags = payload[offset + 1]
+        datatype = payload[offset + 3]
+        offset += 4
+        if flags != 0:
+            raise S7ProtocolError(f"Unsupported S7CommPlus SystemEvent member flags 0x{flags:02X}: {payload.hex()}")
+
+        size = scalar_sizes.get(datatype)
+        if size is None or offset + size > len(payload):
+            raise S7ProtocolError(f"Unsupported or truncated S7CommPlus SystemEvent datatype 0x{datatype:02X}: {payload.hex()}")
+        if member_id == _SYSTEM_EVENT_RETURN_VALUE_ID:
+            if datatype != DataType.LINT:
+                raise S7ProtocolError(f"Malformed S7CommPlus SystemEvent ReturnValue: {payload.hex()}")
+            return int.from_bytes(payload[offset : offset + size], "big", signed=True)
+        offset += size
+
+    # The reference driver treats a data Struct without ReturnValue as fatal.
+    raise S7ProtocolError(f"S7CommPlus SystemEvent Struct has no ReturnValue: {payload.hex()}")
+
+
+def _check_system_event(payload: bytes) -> None:
+    """Raise for fatal/malformed SystemEvents; ignore confirmations/messages."""
+    from snap7.error import S7ProtocolError
+
+    return_value = _system_event_return_value(payload)
+    if return_value is not None and return_value < 0:
+        raise S7ProtocolError(
+            f"Fatal S7CommPlus SystemEvent return_value={return_value}: {payload.hex()}",
+            error_code=return_value,
+        )
+    logger.debug("Ignoring non-fatal S7CommPlus SystemEvent (%d bytes)", len(payload))
+
 
 def _set_s7_groups(ctx: ssl.SSLContext) -> None:
     for group in _S7_PREFERRED_GROUPS:
@@ -773,18 +854,30 @@ 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()
+
         # Large responses (e.g. Explore) are split across several S7CommPlus PDUs.
         if reassemble:
-            data = self._recv_reassembled_payload()
+            data = self._recv_reassembled_payload(response_frame)
             if len(data) < 10:
                 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}"
+                )
             logger.debug(f"  Reassembled response ({len(data)} bytes), payload {len(data) - 10} bytes")
-            return bytes(data[10:])
+            resp_payload = bytes(data[10:])
+            if self._session_key is not None and len(resp_payload) > 1:
+                _, iid_consumed = decode_uint32_vlq(resp_payload, 0)
+                resp_payload = resp_payload[iid_consumed:]
+            return resp_payload
 
-        # Receive response
-        response_frame = self._recv_s7_data()
         logger.debug(f"=== RECV RESPONSE === raw frame ({len(response_frame)} bytes): {response_frame.hex(' ')}")
 
         # Parse frame header, use data_length to exclude trailer
@@ -800,11 +893,6 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
             response = response[1 + hash_len :]
             logger.debug(f"  V3 HMAC ({hash_len} bytes): {response_hmac.hex()}")
 
-        # V254 frames have no standard header — return raw data
-        if version == ProtocolVersion.SYSTEM_EVENT:
-            logger.debug(f"  V254 frame: returning raw data ({len(response)} bytes)")
-            return bytes(response)
-
         logger.debug(f"  Response data ({len(response)} bytes): {response.hex(' ')}")
 
         if len(response) < 10:
@@ -821,6 +909,12 @@ 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).
@@ -844,12 +938,24 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
 
         return resp_payload
 
+    def _recv_response_frame(self) -> bytes:
+        """Receive the next application response, consuming non-fatal SystemEvents."""
+        from snap7.error import S7ProtocolError
+
+        for _ in range(_MAX_SYSTEM_EVENTS_PER_RESPONSE + 1):
+            response_frame = self._recv_s7_data()
+            version, data_length, consumed = decode_header(response_frame)
+            if version != ProtocolVersion.SYSTEM_EVENT:
+                return response_frame
+            _check_system_event(bytes(response_frame[consumed : consumed + data_length]))
+        raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response")
+
     # Sanity caps for fragment reassembly — generous vs. any real PLC EXPLORE response,
     # but bounded so a malformed/adversarial stream can't drive unbounded allocation.
     _MAX_REASSEMBLED_BYTES = 16 * 1024 * 1024
     _MAX_REASSEMBLED_FRAGMENTS = 4096
 
-    def _recv_reassembled_payload(self) -> bytes:
+    def _recv_reassembled_payload(self, initial_data: bytes = b"") -> bytes:
         """Receive a possibly-fragmented S7CommPlus response, returning its data section.
 
         A large response is split into several S7CommPlus PDUs. Each fragment is
@@ -860,7 +966,7 @@ def _recv_reassembled_payload(self) -> bytes:
         """
         from snap7.error import S7ConnectionError
 
-        buf = bytearray()
+        buf = bytearray(initial_data)
 
         def ensure(n: int) -> None:
             while len(buf) < n:
@@ -875,13 +981,22 @@ def ensure(n: int) -> None:
             ensure(4)
             if buf[0] != 0x72:
                 raise S7ConnectionError("Expected S7CommPlus fragment header (0x72)")
+            version = buf[1]
             frag_len = (buf[2] << 8) | buf[3]
             del buf[:4]
             if frag_len == 0:
                 break  # standalone trailer (defensive)
             ensure(frag_len)
-            data.extend(buf[:frag_len])
+            fragment = bytes(buf[:frag_len])
             del buf[:frag_len]
+            if version == ProtocolVersion.V3 and self._session_key is not None:
+                if not fragment:
+                    raise S7ConnectionError("Missing V3 HMAC prefix")
+                hash_len = fragment[0]
+                if len(fragment) < 1 + hash_len:
+                    raise S7ConnectionError("Truncated V3 HMAC prefix")
+                fragment = fragment[1 + hash_len :]
+            data.extend(fragment)
             fragments += 1
             if fragments > self._MAX_REASSEMBLED_FRAGMENTS or len(data) > self._MAX_REASSEMBLED_BYTES:
                 raise S7ConnectionError(f"Reassembled response exceeds limits ({len(data)} bytes, {fragments} fragments)")
@@ -1264,19 +1379,22 @@ def _setup_session(self) -> bool:
         return False
 
     def _build_get_var_substreamed(self, in_object_id: int, address: int, seq_field: int = 1) -> bytes:
-        """Build a GET_VAR_SUBSTREAMED payload (reused by legitimation).
+        """Build the V1 SessionKey GET_VAR_SUBSTREAMED payload.
 
         The ObjectQualifier KEY_QUALIFIER carries the next sequence number.
-        ``seq_field`` is the two-byte request sequence field; the IntegrityId
-        is spliced before the final four-byte fill by ``send_request``.
+        Unlike the TLS/V2 request shape, the V1 SessionKey exchange encodes
+        ``seq_field`` as a VLQ and has a three-byte fill field. The IntegrityId
+        is spliced immediately before that fill by :meth:`send_request`.
         """
-        return _build_get_var_substreamed_payload(
-            in_object_id,
-            address,
-            key_qualifier=self._sequence_number,
-            sequence_field=seq_field,
-            protocol_version=ProtocolVersion.V1,
-        )
+        oq = encode_object_qualifier(key_qualifier=self._sequence_number, protocol_version=ProtocolVersion.V1)
+        payload = struct.pack(">I", in_object_id)
+        payload += bytes([0x20, DataType.UDINT])
+        payload += encode_uint32_vlq(1)  # field count
+        payload += encode_uint32_vlq(address)
+        payload += oq
+        payload += encode_uint32_vlq(seq_field)
+        payload += bytes(3)  # fill
+        return payload
 
     def _session_activate(self) -> None:
         """Activate the V3 session after the SecurityKey handshake.
@@ -1321,7 +1439,7 @@ def _post_auth_legitimation(self, password: str = "") -> None:
         challenge_resp = self.send_request(
             FunctionCode.GET_VAR_SUBSTREAMED,
             self._build_get_var_substreamed(self._session_id, LegitimationId.SERVER_SESSION_REQUEST),
-            integrity_tail=4,
+            integrity_tail=3,
         )
 
         # Extract the 20-byte challenge from the response.

tests/test_s7_unit.py

@@ -20,7 +20,7 @@
 from s7commplus.codec import encode_object_qualifier, encode_pvalue_blob
 from s7commplus.codec import _pvalue_element_size as _element_size
 from s7commplus.codec import skip_typed_value, parse_server_session_version
-from s7commplus.protocol import DataType, ElementID, ObjectId
+from s7commplus.protocol import DataType, ElementID, ObjectId, ProtocolVersion
 from s7commplus.vlq import (
     encode_uint32_vlq,
     encode_uint64_vlq,
@@ -642,6 +642,17 @@ def test_multiple_fragments_split_across_reads(self) -> None:
         conn = self._conn_yielding([self._frag(b"abc"), self._frag(b"de"), self._TRAILER])
         assert conn._recv_reassembled_payload() == b"abcde"
 
+    def test_v3_session_key_hmac_is_stripped_from_each_fragment(self) -> None:
+        conn = self._conn_yielding([])
+        conn._session_key = bytes(24)
+
+        def v3_frag(data: bytes) -> bytes:
+            protected = b"\x20" + bytes(32) + data
+            return bytes([0x72, ProtocolVersion.V3, 0, len(protected)]) + protected
+
+        initial = v3_frag(b"abc") + v3_frag(b"de") + bytes([0x72, ProtocolVersion.V3, 0, 0])
+        assert conn._recv_reassembled_payload(initial) == b"abcde"
+
     def test_bad_fragment_header_raises(self) -> None:
         from snap7.error import S7ConnectionError