fix(s7commplus): verify authenticated responses

CHANGES.md

@@ -9,6 +9,9 @@ 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.
+* Verify authenticated V3 responses and cumulative fragment digests before
+  parsing, invalidating the connection with a dedicated integrity error on any
+  mismatch or truncated envelope.
 * 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/connection.py

@@ -284,20 +284,37 @@ def _set_s7_groups(ctx: ssl.SSLContext) -> None:
     )
 
 
-def _verify_v3_hmac(protected: bytes, session_key: bytes) -> bytes:
-    """Verify and remove the V3 HMAC prefix from application data."""
-    from snap7.error import S7ConnectionError
+def _verify_v3_hmac(protected: bytes, session_key: bytes, digest_state: hmac.HMAC | None = None) -> bytes:
+    """Verify and remove a V3 HMAC prefix.
+
+    ``digest_state`` implements the legacy fragmented-response behavior: the
+    first digest covers the first fragment and each later digest covers all
+    application bytes accumulated so far. The state advances only after a
+    successful constant-time comparison.
+    """
+    from snap7.error import S7IntegrityError
 
     if not protected:
-        raise S7ConnectionError("Empty V3 frame")
+        raise S7IntegrityError("Empty authenticated S7CommPlus V3 frame; reconnect before retrying")
     digest_length = protected[0]
-    if digest_length != hashlib.sha256().digest_size or len(protected) < 1 + digest_length:
-        raise S7ConnectionError(f"Invalid V3 HMAC length: {digest_length}")
+    expected_length = hashlib.sha256().digest_size
+    if digest_length != expected_length:
+        raise S7IntegrityError(
+            f"Invalid S7CommPlus V3 digest length {digest_length}, expected {expected_length}; reconnect before retrying"
+        )
+    if len(protected) < 1 + digest_length:
+        raise S7IntegrityError(
+            f"Truncated S7CommPlus V3 digest: received {len(protected) - 1} of {digest_length} bytes; reconnect before retrying"
+        )
     received_digest = protected[1 : 1 + digest_length]
     application_data = protected[1 + digest_length :]
-    expected_digest = hmac.new(session_key[:24], application_data, hashlib.sha256).digest()
+    verifier = digest_state.copy() if digest_state is not None else hmac.new(session_key[:24], digestmod=hashlib.sha256)
+    verifier.update(application_data)
+    expected_digest = verifier.digest()
     if not hmac.compare_digest(received_digest, expected_digest):
-        raise S7ConnectionError("Invalid V3 HMAC")
+        raise S7IntegrityError("S7CommPlus V3 response integrity check failed; reconnect before retrying")
+    if digest_state is not None:
+        digest_state.update(application_data)
     return bytes(application_data)
 
 
@@ -845,19 +862,18 @@ def _send_legitimation_legacy(self, response: bytes) -> None:
         _check_set_variable_response(resp_payload)
 
     def collect_explore_frames(self, first_payload: bytes) -> bytes:
-        """Collect multi-fragment EXPLORE continuation frames for V3 PLCs.
+        """Collect unauthenticated multi-fragment EXPLORE continuation frames.
 
         On V3 PLCs (FW >= V4.5) a large EXPLORE response (e.g. RID 0x8A11FFFF)
         spans multiple TPKT frames.  The first frame is the normal response
         (already stripped of its 10-byte header by send_request).  Continuation
-        frames carry **no** response header — they are raw BLOB data protected
-        only by a V3 HMAC prefix.  The caller must concatenate them before
-        parsing.
+        frames carry no response header. Authenticated callers must use
+        ``send_request(..., reassemble=True)`` because this legacy helper no
+        longer has the first frame bytes needed to verify cumulative digests.
 
         Termination: a ``frag_len == 0`` frame is the standard S7CommPlus
-        end-of-stream trailer.  As a fallback, a frame whose body (after HMAC
-        strip) is measurably shorter than the first frame body is treated as the
-        last fragment (5-byte tolerance).
+        end-of-stream trailer. As a fallback, a measurably shorter frame body is
+        treated as the last fragment (5-byte tolerance).
 
         Collection is capped by ``_MAX_REASSEMBLED_FRAGMENTS`` and
         ``_MAX_REASSEMBLED_BYTES`` to prevent unbounded allocation on malformed
@@ -870,6 +886,14 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes:
         Returns:
             All fragment payloads concatenated (first_payload + continuations).
         """
+        if self._session_key is not None:
+            from snap7.error import S7ProtocolError
+
+            raise S7ProtocolError(
+                "Authenticated Explore continuations require send_request(..., reassemble=True) so cumulative digests "
+                "can be verified"
+            )
+
         # The first frame body (already header-stripped) was originally
         # len(first_payload) + 10 bytes on the wire (10-byte response header).
         # Continuation frames of the same "full" size will be that long after
@@ -895,10 +919,6 @@ def collect_explore_frames(self, first_payload: bytes) -> bytes:
                 if frag_len == 0:
                     break  # standard S7CommPlus end-of-stream trailer
                 body = raw[4 : 4 + frag_len]
-                # V3 non-TLS: strip the HMAC prefix ([hash_len][hash_bytes])
-                if self._protocol_version >= ProtocolVersion.V3 and len(body) > 33:
-                    hash_len = body[0]
-                    body = body[1 + hash_len :]
                 if not body:
                     break
                 all_data += body
@@ -943,6 +963,46 @@ def disconnect(self) -> None:
         self._notification_frames.clear()
         self._iso_conn.disconnect()
 
+    def _invalidate_integrity_failure(self) -> None:
+        """Close an untrusted stream without sending protocol data on it."""
+        self._session_ready = False
+        self._session_id = 0
+        self.disconnect()
+
+    def _verify_v3_hmac(self, protected: bytes, digest_state: hmac.HMAC | None = None) -> bytes:
+        """Verify authenticated data and make any failure terminal for this connection."""
+        from snap7.error import S7IntegrityError
+
+        if self._session_key is None:
+            self._invalidate_integrity_failure()
+            raise S7IntegrityError("Authenticated S7CommPlus V3 response arrived without a session key; reconnect")
+        try:
+            return _verify_v3_hmac(protected, self._session_key, digest_state)
+        except S7IntegrityError:
+            self._invalidate_integrity_failure()
+            raise
+
+    def _verify_v3_frame(self, frame: bytes) -> None:
+        """Verify a complete V3 frame before its opcode is inspected or queued."""
+        from snap7.error import S7IntegrityError
+
+        version, data_length, consumed = decode_header(frame)
+        if version != ProtocolVersion.V3:
+            if self._session_key is not None:
+                self._invalidate_integrity_failure()
+                raise S7IntegrityError(
+                    f"Authenticated S7CommPlus response used unauthenticated frame version {version}; reconnect"
+                )
+            return
+        frame_end = consumed + data_length
+        if len(frame) < frame_end:
+            self._invalidate_integrity_failure()
+            raise S7IntegrityError(
+                f"Truncated authenticated S7CommPlus V3 frame: declared {data_length} data bytes, "
+                f"received {max(0, len(frame) - consumed)}; reconnect before retrying"
+            )
+        self._verify_v3_hmac(bytes(frame[consumed:frame_end]))
+
     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:
@@ -1067,11 +1127,7 @@ def _send_request(self, function_code: int, payload: bytes, integrity_tail: int,
 
         # V3 responses have a hash-length byte + HMAC prefix before the payload.
         if version == ProtocolVersion.V3:
-            if self._session_key is None:
-                from snap7.error import S7ConnectionError
-
-                raise S7ConnectionError("V3 response received without a session key")
-            response = _verify_v3_hmac(response, self._session_key)
+            response = self._verify_v3_hmac(response)
             logger.debug("  V3 HMAC verified")
 
         logger.debug(f"  Response data ({len(response)} bytes): {response.hex(' ')}")
@@ -1126,6 +1182,7 @@ def _recv_response_frame(self, expected_sequence: Optional[int] = None) -> bytes
                 if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE:
                     raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response")
                 continue
+            self._verify_v3_frame(response_frame)
             if data_length < 10:
                 return response_frame
             opcode = _incoming_frame_opcode(response_frame)
@@ -1173,6 +1230,7 @@ def receive_notification(self) -> bytes:
 
             raise S7ConnectionError("Not connected")
         frame = self._notification_frames.popleft() if self._notification_frames else self._recv_s7_data()
+        self._verify_v3_frame(frame)
         if not self._is_notification_frame(frame):
             from snap7.error import S7ConnectionError
 
@@ -1206,11 +1264,27 @@ def ensure(n: int) -> None:
 
         data = bytearray()
         fragments = 0
+        expected_version: int | None = None
+        digest_state = hmac.new(self._session_key[:24], digestmod=hashlib.sha256) if self._session_key is not None else None
         while True:
             ensure(4)
             if buf[0] != 0x72:
                 raise S7ConnectionError("Expected S7CommPlus fragment header (0x72)")
             fragment_version = buf[1]
+            if expected_version is None:
+                expected_version = fragment_version
+            elif fragment_version != expected_version:
+                if self._session_key is not None:
+                    from snap7.error import S7IntegrityError
+
+                    self._invalidate_integrity_failure()
+                    raise S7IntegrityError(
+                        f"Authenticated S7CommPlus response changed fragment version from {expected_version} "
+                        f"to {fragment_version}; reconnect"
+                    )
+                raise S7ConnectionError(
+                    f"S7CommPlus response changed fragment version from {expected_version} to {fragment_version}"
+                )
             frag_len = (buf[2] << 8) | buf[3]
             del buf[:4]
             if frag_len == 0:
@@ -1219,17 +1293,15 @@ def ensure(n: int) -> None:
             fragment_data = bytes(buf[:frag_len])
             del buf[:frag_len]
             if fragment_version == ProtocolVersion.V3:
-                if self._session_key is None:
-                    raise S7ConnectionError("V3 response received without a session key")
-                fragment_data = _verify_v3_hmac(fragment_data, self._session_key)
+                fragment_data = self._verify_v3_hmac(fragment_data, digest_state)
             data.extend(fragment_data)
             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)")
             # The next 4 bytes are either the trailer (0x72 ver 0x0000) or the next
             # fragment's header (0x72 ver len>0).
             ensure(4)
-            if buf[0] == 0x72 and buf[2] == 0 and buf[3] == 0:
+            if buf[0] == 0x72 and buf[1] == expected_version and buf[2] == 0 and buf[3] == 0:
                 del buf[:4]  # consume trailer — last fragment
                 break
         return bytes(data)

snap7/error.py

@@ -28,6 +28,12 @@ class S7ProtocolError(S7Error):
     pass
 
 
+class S7IntegrityError(S7ProtocolError):
+    """Raised when authenticated S7CommPlus traffic fails integrity checks."""
+
+    pass
+
+
 class S7TimeoutError(S7Error):
     """Raised when S7 operation times out."""
 

tests/test_error.py

@@ -6,6 +6,7 @@
     S7Error,
     S7ConnectionError,
     S7ProtocolError,
+    S7IntegrityError,
     S7TimeoutError,
     S7AuthenticationError,
     S7StalePacketError,
@@ -33,6 +34,7 @@ def test_s7error_without_code(self) -> None:
     def test_subclass_hierarchy(self) -> None:
         assert issubclass(S7ConnectionError, S7Error)
         assert issubclass(S7ProtocolError, S7Error)
+        assert issubclass(S7IntegrityError, S7ProtocolError)
         assert issubclass(S7TimeoutError, S7Error)
         assert issubclass(S7AuthenticationError, S7Error)
         assert issubclass(S7StalePacketError, S7ProtocolError)
@@ -42,6 +44,7 @@ def test_all_subclasses_instantiate(self) -> None:
         for cls in (
             S7ConnectionError,
             S7ProtocolError,
+            S7IntegrityError,
             S7TimeoutError,
             S7AuthenticationError,
             S7StalePacketError,

tests/test_s7_legacy_request_layout.py

@@ -33,7 +33,9 @@ def test_v1_substreamed_request_matches_accepted_tia_packet() -> None:
     conn._integrity_id_read = 1
     conn._send_s7_data = MagicMock()
     body = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_VAR_SUBSTREAMED, 0, 3, 0x34) + bytes(4)
-    conn._recv_s7_data = MagicMock(return_value=encode_header(ProtocolVersion.V2, len(body)) + body)
+    response_digest = hmac.new(conn._session_key, body, hashlib.sha256).digest()
+    protected_body = bytes([len(response_digest)]) + response_digest + body
+    conn._recv_s7_data = MagicMock(return_value=encode_header(ProtocolVersion.V3, len(protected_body)) + protected_body)
     conn.send_request(FunctionCode.GET_VAR_SUBSTREAMED, payload)
     frame = conn._send_s7_data.call_args.args[0]
     assert frame[37:-4] == expected

tests/test_s7_server.py

@@ -9,7 +9,7 @@
 
 import pytest
 
-from snap7.error import S7ConnectionError
+from snap7.error import S7ConnectionError, S7IntegrityError
 from s7commplus.async_client import S7CommPlusAsyncClient
 from s7commplus.client import S7CommPlusClient
 from s7commplus.connection import _parse_get_var_substreamed_response, _verify_v3_hmac
@@ -489,7 +489,7 @@ def test_v3_hmac_verification(self) -> None:
         tampered = protected[:-1] + bytes([protected[-1] ^ 0x01])
         with pytest.raises(ConnectionError, match="Invalid V3 HMAC"):
             S7CommPlusServer._verify_v3_data(tampered, TEST_SESSION_KEY)
-        with pytest.raises(S7ConnectionError, match="Invalid V3 HMAC"):
+        with pytest.raises(S7IntegrityError, match="integrity check failed"):
             _verify_v3_hmac(tampered, TEST_SESSION_KEY)
 
     def test_create_object_response_contains_fingerprint_and_challenge(self) -> None:

tests/test_s7_subscription.py

@@ -1,12 +1,14 @@
 """Tests for S7CommPlus symbolic data subscriptions."""
 
+import hashlib
+import hmac
 import struct
 from unittest.mock import MagicMock
 
 import pytest
 
 from s7commplus.client import S7CommPlusClient
-from s7commplus.codec import encode_header, encode_pvalue_blob
+from s7commplus.codec import decode_header, encode_header, encode_pvalue_blob
 from s7commplus.connection import S7CommPlusConnection
 from s7commplus.protocol import DataType, FunctionCode, Ids, Opcode, ProtocolVersion
 from s7commplus.subscription import (
@@ -16,6 +18,7 @@
     parse_subscription_notification,
 )
 from s7commplus.vlq import encode_uint32_vlq, encode_uint64_vlq
+from snap7.error import S7IntegrityError
 
 
 def _response_frame(function_code: int, sequence: int, payload: bytes) -> bytes:
@@ -192,3 +195,26 @@ def test_send_request_queues_interleaved_notification(self) -> None:
         assert connection.send_request(FunctionCode.GET_VARIABLE, b"\x00\x00\x00\x00") == b"\x00"
         assert connection.receive_notification() == notification
         assert connection._recv_s7_data.call_count == 2
+
+    def test_authenticated_notification_is_verified_before_queueing(self) -> None:
+        connection = S7CommPlusConnection("127.0.0.1")
+        connection._connected = True
+        connection._protocol_version = ProtocolVersion.V3
+        connection._session_id = 1
+        connection._session_key = bytes(range(24))
+        connection._iso_conn.disconnect = MagicMock()
+
+        notification = _notification_frame(version=ProtocolVersion.V3)
+        _, data_length, consumed = decode_header(notification)
+        data = notification[consumed : consumed + data_length]
+        digest = hmac.new(connection._session_key, data, hashlib.sha256).digest()
+        protected = bytearray(bytes([len(digest)]) + digest + data)
+        protected[1] ^= 1
+        notification = encode_header(ProtocolVersion.V3, len(protected)) + protected
+        connection._send_s7_data = MagicMock()
+        connection._recv_s7_data = MagicMock(return_value=bytes(notification))
+
+        with pytest.raises(S7IntegrityError, match="integrity check failed"):
+            connection.send_request(FunctionCode.GET_VARIABLE, bytes(4))
+        assert not connection.connected
+        assert not connection._notification_frames