diff --git a/CHANGES.md b/CHANGES.md index 4474ffe9..409f6a31 100644 --- a/CHANGES.md +++ b/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): diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index 93f4b4aa..453501a3 100644 --- a/s7commplus/async_client.py +++ b/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: diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 756fd5e3..0820d93d 100644 --- a/s7commplus/connection.py +++ b/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. diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index 7466b967..63d38572 100644 --- a/tests/test_s7_unit.py +++ b/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 diff --git a/tests/test_s7_v2.py b/tests/test_s7_v2.py index a93efa6f..ff443c4e 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -6,7 +6,7 @@ import hashlib import struct -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest @@ -36,7 +36,7 @@ ProtocolVersion, ) from s7commplus.vlq import decode_uint32_vlq, encode_uint32_vlq -from snap7.error import S7ConnectionError +from snap7.error import S7ConnectionError, S7ProtocolError class TestReadFunctionCodes: @@ -214,7 +214,7 @@ def test_tls_v2_response_application_payload_is_not_stripped(self) -> None: conn._with_integrity_id = True application_payload = bytes.fromhex("000100100201") - response = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 1, 0x34) + response = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) response += application_payload frame = encode_header(ProtocolVersion.V2, len(response)) + response frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) @@ -224,6 +224,62 @@ def test_tls_v2_response_application_payload_is_not_stripped(self) -> None: assert conn.send_request(FunctionCode.GET_MULTI_VARIABLES, bytes(4)) == application_payload + def test_nonfatal_system_event_is_consumed_before_sync_response(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V3 + conn._session_id = 0x0000039B + + confirmation = bytes.fromhex("00000000000002f60000000000000000") + event_frame = encode_header(ProtocolVersion.SYSTEM_EVENT, len(confirmation)) + confirmation + application_payload = b"\x00\x01" + response = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.SET_VAR_SUBSTREAMED, 0, 0, 0x34) + response += application_payload + response_frame = encode_header(ProtocolVersion.V3, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V3, 0) + + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(side_effect=[event_frame, response_frame]) + + assert conn.send_request(FunctionCode.SET_VAR_SUBSTREAMED) == application_payload + assert conn._recv_s7_data.call_count == 2 + + def test_fatal_system_event_raises_protocol_error(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._connected = True + conn._protocol_version = ProtocolVersion.V3 + conn._session_id = 0x0000039B + + fatal = bytes(16) + fatal += bytes.fromhex("0000001700009d6c") + fatal += struct.pack(">I", 40305) + bytes.fromhex("00000009") + (-1).to_bytes(8, "big", signed=True) + event_frame = encode_header(ProtocolVersion.SYSTEM_EVENT, len(fatal)) + fatal + conn._send_s7_data = MagicMock() + conn._recv_s7_data = MagicMock(return_value=event_frame) + + with pytest.raises(S7ProtocolError, match="Fatal S7CommPlus SystemEvent"): + conn.send_request(FunctionCode.SET_VAR_SUBSTREAMED) + + @pytest.mark.asyncio + async def test_nonfatal_system_event_is_consumed_before_async_response(self) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + + confirmation = bytes.fromhex("00000000000002f60000000000000000") + event_frame = encode_header(ProtocolVersion.SYSTEM_EVENT, len(confirmation)) + confirmation + response = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) + response += b"\x00\x01" + response_frame = encode_header(ProtocolVersion.V2, len(response)) + response + response_frame += struct.pack(">BBH", 0x72, ProtocolVersion.V2, 0) + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(side_effect=[event_frame, response_frame]) + + assert await client._send_request(FunctionCode.GET_MULTI_VARIABLES, bytes(4)) == b"\x00\x01" + assert client._recv_cotp_dt.await_count == 2 + class TestIntegrityIdVlqEncoding: """Test VLQ encoding used for IntegrityId values.""" @@ -264,6 +320,33 @@ def test_build_get_var_substreamed_payload(self) -> None: expected += struct.pack(">I", 0) assert payload == expected + def test_build_v1_session_key_challenge_payload(self) -> None: + """Match the challenge request accepted by the S7-1200 in GH-710.""" + conn = S7CommPlusConnection("127.0.0.1") + conn._sequence_number = 4 + + payload = conn._build_get_var_substreamed(0x0000039B, LegitimationId.SERVER_SESSION_REQUEST) + + assert payload == bytes.fromhex("0000039b200401822f000004e88969001200000000896a001300896b00040000000401000000") + + def test_v1_session_key_challenge_splices_integrity_before_three_byte_fill(self) -> None: + conn = S7CommPlusConnection("127.0.0.1") + conn._session_id = 0x0000039B + conn._session_challenge = bytes(range(20)) + conn._session_key = bytes(range(24)) + conn._session_auth_public_key = bytes(range(24)) + conn.send_request = MagicMock(side_effect=[b"", b"\x00"]) + + with patch("s7commplus.session_auth.legitimate.solve_legitimate_challenge_real_plc", return_value=bytes(248)): + conn._post_auth_legitimation() + + first_call = conn.send_request.call_args_list[0] + assert first_call.args == ( + FunctionCode.GET_VAR_SUBSTREAMED, + conn._build_get_var_substreamed(0x0000039B, LegitimationId.SERVER_SESSION_REQUEST), + ) + assert first_call.kwargs == {"integrity_tail": 3} + def test_parse_get_var_substreamed_usint_array(self) -> None: challenge = bytes(range(20)) response = bytes([0x00, 0x00, 0x10, 0x02])