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

@@ -7,7 +7,9 @@
 import logging
 import ssl
 import struct
-from typing import Any, Optional
+from typing import Any, Awaitable, Callable, Optional, TypeVar
+
+from snap7.error import S7ConnectionError, S7ProtocolError
 
 from . import typeinfo
 from .blob_decompressor import find_and_decompress
@@ -36,9 +38,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,
@@ -61,6 +65,8 @@
 
 logger = logging.getLogger(__name__)
 
+_T = TypeVar("_T")
+
 # COTP constants
 _COTP_CR = 0xE0
 _COTP_CC = 0xD0
@@ -81,6 +87,7 @@ def __init__(self) -> None:
         self._protocol_version: int = 0
         self._connected = False
         self._lock = asyncio.Lock()
+        self._connect_params: Optional[dict[str, Any]] = None
 
         # V2+ IntegrityId tracking
         self._integrity_id_read: int = 0
@@ -157,6 +164,16 @@ async def connect(
             tls_key: Path to client private key (PEM)
             tls_ca: Path to CA certificate for PLC verification (PEM)
         """
+        self._connect_params = {
+            "host": host,
+            "port": port,
+            "rack": rack,
+            "slot": slot,
+            "use_tls": use_tls,
+            "tls_cert": tls_cert,
+            "tls_key": tls_key,
+            "tls_ca": tls_ca,
+        }
         self._host = host
 
         # TCP connect
@@ -422,6 +439,24 @@ async def disconnect(self) -> None:
                 pass
             self._writer = None
             self._reader = None
+        self._connect_params = None
+
+    async def _reconnect(self) -> None:
+        """Tear down and re-establish the connection with the same parameters."""
+        if self._connect_params is None:
+            raise RuntimeError("Not connected")
+        params = self._connect_params.copy()
+        await self.disconnect()
+        await self.connect(**params)
+
+    async def _with_reconnect(self, op: Callable[[], Awaitable[_T]]) -> _T:
+        """Run ``op``; if the PLC dropped the socket, reconnect once and retry."""
+        try:
+            return await op()
+        except S7ConnectionError as exc:
+            logger.info("Connection dropped by PLC (%s); reconnecting and retrying", exc)
+            await self._reconnect()
+            return await op()
 
     async def db_read(self, db_number: int, start: int, size: int) -> bytes:
         """Read raw bytes from a data block."""
@@ -637,7 +672,7 @@ async def browse(self) -> list[dict[str, Any]]:
         for db_info in await self.list_datablocks():
             if db_info.get("number", 0) <= 0 or db_info.get("rid", 0) == 0:
                 continue
-            ti_rid = await self._read_typeinfo_rid(db_info["rid"])
+            ti_rid = await self._with_reconnect(lambda: self._read_typeinfo_rid(db_info["rid"]))
             if ti_rid == 0:
                 continue  # load-memory-only DB, skip
             root_nodes.append(
@@ -659,7 +694,7 @@ async def browse(self) -> list[dict[str, Any]]:
             )
 
         # Phase D: explore the OMS type-info container (a large, multi-fragment PDU).
-        type_objects = await self._explore_type_info_container()
+        type_objects = await self._with_reconnect(self._explore_type_info_container)
 
         # Phase E: recombine type-info with the DB/area nodes and flatten.
         typeinfo.build_tree(root_nodes, type_objects)
@@ -686,6 +721,8 @@ async def _read_typeinfo_rid(self, db_rid: int) -> int:
         """Read LID=1 of a DB to get its type-info RID (0 if the DB has no readable value)."""
         try:
             raw = await self.read_symbolic(db_rid, [1], 0)
+        except (S7ConnectionError, S7ProtocolError):
+            raise
         except Exception:
             return 0
         return struct.unpack(">I", raw[:4])[0] if len(raw) >= 4 else 0
@@ -765,27 +802,49 @@ 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")
+                    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}"
+                    )
                 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:
-                raise RuntimeError("Response too short")
+                raise S7ConnectionError("Response too short")
 
             # 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_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 S7ProtocolError("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
@@ -794,21 +853,21 @@ 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:
                 chunk = await self._recv_cotp_dt()
                 if not chunk:
-                    raise RuntimeError("Connection closed during response reassembly")
+                    raise S7ConnectionError("Connection closed during response reassembly")
                 buf.extend(chunk)
 
         data = bytearray()
         fragments = 0
         while True:
             await ensure(4)
             if buf[0] != 0x72:
-                raise RuntimeError("Expected S7CommPlus fragment header (0x72)")
+                raise S7ConnectionError("Expected S7CommPlus fragment header (0x72)")
             frag_len = (buf[2] << 8) | buf[3]
             del buf[:4]
             if frag_len == 0:
@@ -818,7 +877,7 @@ async def ensure(n: int) -> None:
             del buf[:frag_len]
             fragments += 1
             if fragments > self._MAX_REASSEMBLED_FRAGMENTS or len(data) > self._MAX_REASSEMBLED_BYTES:
-                raise RuntimeError(f"Reassembled response exceeds limits ({len(data)} bytes, {fragments} fragments)")
+                raise S7ConnectionError(f"Reassembled response exceeds limits ({len(data)} bytes, {fragments} fragments)")
             # The next 4 bytes are either the trailer (0x72 ver 0x0000) or the next
             # fragment's header (0x72 ver len>0).
             await ensure(4)
@@ -875,8 +934,8 @@ async def _init_ssl(self) -> None:
         version, data_length, consumed = decode_header(response_data)
         response = response_data[consumed : consumed + data_length]
 
-        if len(response) < 14:
-            raise RuntimeError("InitSSL response too short")
+        if len(response) < 10:
+            raise S7ConnectionError("InitSSL response too short")
 
         logger.debug(f"InitSSL response received, version=V{version}")
 
@@ -928,7 +987,7 @@ async def _create_session(self) -> None:
         response = response_data[consumed : consumed + data_length]
 
         if len(response) < 10:
-            raise RuntimeError("CreateObject response too short")
+            raise S7ConnectionError("CreateObject response too short")
 
         # Response header is 10 bytes (opcode+reserved+func+reserved+seq+transport).
         # Responses do NOT carry a SessionId field (unlike requests which are 14 bytes).

s7commplus/client.py

@@ -7,7 +7,7 @@
 import struct
 from typing import Any, Callable, Optional, TypeVar
 
-from snap7.error import S7ConnectionError
+from snap7.error import S7ConnectionError, S7ProtocolError
 
 from . import typeinfo
 from .blob_decompressor import find_and_decompress
@@ -599,8 +599,9 @@ def _read_typeinfo_rid(self, db_rid: int) -> int:
         """Read LID=1 of a DB to get its type-info RID (0 if the DB has no readable value)."""
         try:
             raw = self.read_symbolic(db_rid, [1], 0)
-        except S7ConnectionError:
-            # Socket was RST by the PLC — let the caller reconnect and retry.
+        except (S7ConnectionError, S7ProtocolError):
+            # Connection failures are eligible for reconnect; protocol failures
+            # must reach the caller instead of looking like an unreadable DB.
             raise
         except Exception:
             return 0