feat(server): add V3 HMAC framing

s7commplus/connection.py

@@ -92,6 +92,23 @@ def _set_s7_groups(ctx: ssl.SSLContext) -> None:
     logger.warning("Could not restrict TLS groups — PLC may reject unsupported groups in ClientHello")
 
 
+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
+
+    if not protected:
+        raise S7ConnectionError("Empty V3 frame")
+    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}")
+    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()
+    if not hmac.compare_digest(received_digest, expected_digest):
+        raise S7ConnectionError("Invalid V3 HMAC")
+    return bytes(application_data)
+
+
 def _build_get_var_substreamed_payload(
     in_object_id: int,
     address: int,
@@ -716,7 +733,12 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
 
                 raise S7ConnectionError("Response too short")
             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:
+                resp_iid, iid_consumed = decode_uint32_vlq(resp_payload, 0)
+                logger.debug(f"  Response IntegrityId: {resp_iid} ({iid_consumed} bytes)")
+                resp_payload = resp_payload[iid_consumed:]
+            return resp_payload
 
         # Receive response
         response_frame = self._recv_s7_data()
@@ -728,12 +750,14 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail:
 
         response = response_frame[consumed : consumed + data_length]
 
-        # V3 responses have a hash-length byte + HMAC prefix before the payload
-        if version == ProtocolVersion.V3 and len(response) > 33:
-            hash_len = response[0]
-            response_hmac = response[1 : 1 + hash_len]
-            response = response[1 + hash_len :]
-            logger.debug(f"  V3 HMAC ({hash_len} bytes): {response_hmac.hex()}")
+        # 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)
+            logger.debug("  V3 HMAC verified")
 
         # V254 frames have no standard header — return raw data
         if version == ProtocolVersion.SYSTEM_EVENT:
@@ -810,13 +834,19 @@ def ensure(n: int) -> None:
             ensure(4)
             if buf[0] != 0x72:
                 raise S7ConnectionError("Expected S7CommPlus fragment header (0x72)")
+            fragment_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_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)
+            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)")

s7commplus/server.py

@@ -23,6 +23,8 @@
     server.start(port=11020, use_tls=True, tls_cert="cert.pem", tls_key="key.pem")
 """
 
+import hashlib
+import hmac
 import logging
 import socket
 import ssl
@@ -190,8 +192,8 @@ class S7CommPlusServer:
 
     Emulates an S7-1200/1500 PLC with:
     - Internal data block storage with named variables
-    - S7CommPlus protocol handling (V1 and V2)
-    - V2 TLS support with IntegrityId tracking
+    - S7CommPlus protocol handling (V1, V2, and V3)
+    - V2 TLS and V3 HMAC support with IntegrityId tracking
     - Multi-client support (threaded)
     - CPU state management
     """
@@ -201,6 +203,7 @@ def __init__(
         protocol_version: int = ProtocolVersion.V1,
         public_key_fingerprint: Optional[str] = None,
         session_challenge: Optional[bytes] = None,
+        session_key: Optional[bytes] = None,
         rst_after_symbolic_read: bool = False,
     ) -> None:
         self._data_blocks: dict[int, DataBlock] = {}
@@ -225,6 +228,12 @@ def __init__(
         # the post-auth legitimation flow (GetVarSubStreamed 303 / SetVarSubStreamed 1846).
         self._public_key_fingerprint = public_key_fingerprint
         self._session_challenge = session_challenge
+        if session_key is not None and len(session_key) < 24:
+            raise ValueError("session_key must contain at least 24 bytes")
+        # An emulator cannot recover the client's random key from a blob
+        # encrypted for a real Siemens PLC. Tests may therefore provide the
+        # negotiated key explicitly to exercise V3 framing end to end.
+        self._session_key = session_key
 
         # When True, the server closes the TCP connection after responding
         # to a GetMultiVariables request (emulating firmware like S7-1200
@@ -409,9 +418,16 @@ def recv_app_frame() -> Optional[bytes]:
                             return None
                         tls["in"].write(more)
 
-            def send_app_frame(data: bytes) -> None:
-                frame = encode_header(self._protocol_version, len(data)) + data
-                frame += struct.pack(">BBH", 0x72, self._protocol_version, 0x0000)
+            def send_app_frame(data: bytes, frame_version: int) -> None:
+                if frame_version == ProtocolVersion.V3:
+                    if self._session_key is None:
+                        raise ConnectionError("V3 response requested without a session key")
+                    digest = hmac.new(self._session_key[:24], data, hashlib.sha256).digest()
+                    frame_data = bytes([len(digest)]) + digest + data
+                else:
+                    frame_data = data
+                frame = encode_header(frame_version, len(frame_data)) + frame_data
+                frame += struct.pack(">BBH", 0x72, frame_version, 0x0000)
                 if tls["obj"] is None:
                     self._send_cotp_dt_raw(client_sock, frame)
                 else:
@@ -426,6 +442,15 @@ def send_app_frame(data: bytes) -> None:
                     if data is None:
                         break
 
+                    request_version, data_length, hdr_consumed = decode_header(data)
+                    if request_version == ProtocolVersion.V3:
+                        if self._session_key is None:
+                            raise ConnectionError("V3 request received without a session key")
+                        protected = data[hdr_consumed : hdr_consumed + data_length]
+                        request_data = self._verify_v3_data(protected, self._session_key)
+                        # _process_request consumes a normal S7CommPlus frame.
+                        data = encode_header(request_version, len(request_data)) + request_data
+
                     # Decode the request function code once (used for TLS + IntegrityId).
                     func_code = None
                     try:
@@ -440,7 +465,12 @@ def send_app_frame(data: bytes) -> None:
                     if response is not None:
                         if session_id == 0 and len(response) >= 14:
                             session_id = struct.unpack_from(">I", response, 9)[0]
-                        send_app_frame(response)
+                        if request_version == ProtocolVersion.V3 and func_code is not None:
+                            response_integrity_id = integrity_id_read if func_code in READ_FUNCTION_CODES else integrity_id_write
+                            response = response[:10] + encode_uint32_vlq(response_integrity_id) + response[10:]
+                        send_app_frame(
+                            response, request_version if request_version == ProtocolVersion.V3 else self._protocol_version
+                        )
 
                     if rst:
                         logger.debug(f"RST emulation: closing connection to {address}")
@@ -457,7 +487,11 @@ def send_app_frame(data: bytes) -> None:
                         logger.debug(f"TLS activated (COTP-tunneled) for client {address}")
 
                     # Update IntegrityId counters based on function code (V2+).
-                    if self._protocol_version >= ProtocolVersion.V2 and session_id != 0 and func_code is not None:
+                    if (
+                        (self._protocol_version >= ProtocolVersion.V2 or request_version == ProtocolVersion.V3)
+                        and session_id != 0
+                        and func_code is not None
+                    ):
                         if func_code in READ_FUNCTION_CODES:
                             integrity_id_read = (integrity_id_read + 1) & 0xFFFFFFFF
                         elif func_code not in (FunctionCode.INIT_SSL, FunctionCode.CREATE_OBJECT):
@@ -477,6 +511,21 @@ def send_app_frame(data: bytes) -> None:
                 pass
             logger.info(f"Client disconnected: {address}")
 
+    @staticmethod
+    def _verify_v3_data(protected: bytes, session_key: bytes) -> bytes:
+        """Verify and remove a V3 HMAC prefix from application data."""
+        if not protected:
+            raise ConnectionError("Empty V3 frame")
+        digest_length = protected[0]
+        if digest_length != hashlib.sha256().digest_size or len(protected) < 1 + digest_length:
+            raise ConnectionError(f"Invalid V3 HMAC length: {digest_length}")
+        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()
+        if not hmac.compare_digest(received_digest, expected_digest):
+            raise ConnectionError("Invalid V3 HMAC")
+        return bytes(application_data)
+
     def _server_tls_handshake(self, sock: socket.socket) -> tuple[Any, Any, Any]:
         """Perform the server-side TLS handshake, tunneling records through COTP DT frames."""
         assert self._ssl_context is not None

tests/test_s7_server.py

@@ -1,15 +1,18 @@
 """Integration tests for S7CommPlus server, client, and async client."""
 
 import asyncio
+import hashlib
+import hmac
 import struct
 import time
 from collections.abc import Generator
 
 import pytest
 
+from snap7.error import S7ConnectionError
 from s7commplus.async_client import S7CommPlusAsyncClient
 from s7commplus.client import S7CommPlusClient
-from s7commplus.connection import _parse_get_var_substreamed_response
+from s7commplus.connection import S7CommPlusConnection, _parse_get_var_substreamed_response, _verify_v3_hmac
 from s7commplus.protocol import DataType, ElementID, Ids, LegitimationId, ObjectId, ProtocolVersion
 from s7commplus.server import CPUState, DataBlock, S7CommPlusServer
 from s7commplus.vlq import encode_uint32_vlq
@@ -25,6 +28,7 @@
 
 # Fixed 20-byte challenge for deterministic tests
 TEST_CHALLENGE = bytes(range(20))
+TEST_SESSION_KEY = bytes(range(24))
 
 
 @pytest.fixture()
@@ -419,9 +423,28 @@ def test_session_key_server_construction(self) -> None:
         srv = S7CommPlusServer(
             public_key_fingerprint=TEST_FINGERPRINT,
             session_challenge=TEST_CHALLENGE,
+            session_key=TEST_SESSION_KEY,
         )
         assert srv._public_key_fingerprint == TEST_FINGERPRINT
         assert srv._session_challenge == TEST_CHALLENGE
+        assert srv._session_key == TEST_SESSION_KEY
+
+    def test_short_session_key_rejected(self) -> None:
+        with pytest.raises(ValueError, match="at least 24 bytes"):
+            S7CommPlusServer(session_key=b"too short")
+
+    def test_v3_hmac_verification(self) -> None:
+        application_data = b"S7CommPlus request"
+        digest = hmac.new(TEST_SESSION_KEY, application_data, hashlib.sha256).digest()
+        protected = bytes([len(digest)]) + digest + application_data
+        assert S7CommPlusServer._verify_v3_data(protected, TEST_SESSION_KEY) == application_data
+        assert _verify_v3_hmac(protected, TEST_SESSION_KEY) == application_data
+
+        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"):
+            _verify_v3_hmac(tampered, TEST_SESSION_KEY)
 
     def test_create_object_response_contains_fingerprint_and_challenge(self) -> None:
         """Verify the CreateObject response includes attributes 233 and 303."""
@@ -498,10 +521,23 @@ class TestSessionKeyIntegration:
     """
 
     @pytest.fixture()
-    def session_key_server(self) -> Generator[S7CommPlusServer, None, None]:
+    def session_key_server(self, monkeypatch: pytest.MonkeyPatch) -> Generator[S7CommPlusServer, None, None]:
+        # The emulator does not own Siemens' private key, so make the client-side
+        # key exchange deterministic and configure the matching negotiated key.
+        from s7commplus.session_auth import legitimate
+        from s7commplus.session_auth.keys import KeyFamily
+
+        def authenticate(_connection: S7CommPlusConnection) -> tuple[bytes, bytes]:
+            _connection._session_auth_public_key = bytes(40)
+            _connection._session_auth_family = KeyFamily.S7_1200
+            return bytes(180), TEST_SESSION_KEY
+
+        monkeypatch.setattr(S7CommPlusConnection, "_try_session_key_auth", authenticate)
+        monkeypatch.setattr(legitimate, "solve_legitimate_challenge_real_plc", lambda *args: bytes(248))
         srv = S7CommPlusServer(
             public_key_fingerprint=TEST_FINGERPRINT,
             session_challenge=TEST_CHALLENGE,
+            session_key=TEST_SESSION_KEY,
         )
         srv.register_db(1, {"temperature": ("Real", 0)})
         db1 = srv.get_db(1)
@@ -519,6 +555,8 @@ def test_client_session_setup_with_struct_version(self, session_key_server: S7Co
         assert client.connected
         assert client.session_id != 0
         assert client.session_setup_ok
+        assert client._connection is not None
+        assert client._connection._session_key == TEST_SESSION_KEY
         client.disconnect()
 
     def test_read_write_after_session_setup(self, session_key_server: S7CommPlusServer) -> None:
@@ -536,3 +574,12 @@ def test_read_write_after_session_setup(self, session_key_server: S7CommPlusServ
             assert abs(value - 42.0) < 0.001
         finally:
             client.disconnect()
+
+    def test_explore_after_session_setup(self, session_key_server: S7CommPlusServer) -> None:
+        """Explore reassembly verifies V3 HMAC and removes the response IntegrityId."""
+        client = S7CommPlusClient()
+        client.connect("127.0.0.1", port=SESSION_KEY_PORT)
+        try:
+            assert client.list_datablocks() == [{"name": "DB1", "number": 1, "rid": 0x8A0E0001}]
+        finally:
+            client.disconnect()