diff --git a/s7commplus/connection.py b/s7commplus/connection.py index e1f6bd83..f415e317 100644 --- a/s7commplus/connection.py +++ b/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)") diff --git a/s7commplus/server.py b/s7commplus/server.py index e850fc6c..08a060f8 100644 --- a/s7commplus/server.py +++ b/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 diff --git a/tests/test_s7_server.py b/tests/test_s7_server.py index 6b114562..690e3a93 100644 --- a/tests/test_s7_server.py +++ b/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()