diff --git a/CHANGES.md b/CHANGES.md index 43e7e218..450ab349 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -22,6 +22,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) * S7CommPlus active-alarm browsing and alarm subscriptions (experimental) * S7CommPlus symbolic data subscriptions and notification decoding (experimental) * TIA Portal XML import for SymbolTable (`SymbolTable.from_tia_xml()`) (experimental) diff --git a/s7commplus/async_client.py b/s7commplus/async_client.py index eed68675..2e20821d 100644 --- a/s7commplus/async_client.py +++ b/s7commplus/async_client.py @@ -8,7 +8,9 @@ import ssl import struct from collections.abc import Sequence -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 @@ -40,9 +42,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, _log_create_object_return_value, _parse_get_var_substreamed_response, @@ -85,6 +89,8 @@ logger = logging.getLogger(__name__) +_T = TypeVar("_T") + # COTP constants _COTP_CR = 0xE0 _COTP_CC = 0xD0 @@ -108,6 +114,7 @@ def __init__(self) -> None: self._session_ready = False self._connected = False self._lock = asyncio.Lock() + self._connect_params: Optional[dict[str, Any]] = None # V2+ IntegrityId tracking self._integrity_id_read: int = 0 @@ -189,6 +196,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 @@ -497,6 +514,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 S7ConnectionError("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.""" @@ -793,7 +828,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( @@ -815,7 +850,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) @@ -842,6 +877,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 @@ -882,7 +919,7 @@ async def _send_request( """ async with self._lock: if not (self._connected or self._transport_connected) or self._writer is None or self._reader is None: - raise RuntimeError("Not connected") + raise S7ConnectionError("Not connected") seq_num = self._next_sequence_number() @@ -921,27 +958,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 @@ -950,13 +1009,13 @@ 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() @@ -964,7 +1023,7 @@ async def ensure(n: int) -> None: 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: @@ -974,7 +1033,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) @@ -986,7 +1045,7 @@ async def ensure(n: int) -> None: async def _cotp_connect(self, local_tsap: int, remote_tsap: bytes) -> None: """Perform COTP Connection Request / Confirm handshake.""" if self._writer is None or self._reader is None: - raise RuntimeError("Not connected") + raise S7ConnectionError("Not connected") base_pdu = struct.pack(">BBHHB", 6, _COTP_CR, 0x0000, 0x0001, 0x00) calling_tsap = struct.pack(">BBH", 0xC1, 2, local_tsap) @@ -1004,8 +1063,10 @@ async def _cotp_connect(self, local_tsap: int, remote_tsap: bytes) -> None: _, _, length = struct.unpack(">BBH", tpkt_header) payload = await self._reader.readexactly(length - 4) - if len(payload) < 7 or payload[1] != _COTP_CC: - raise RuntimeError(f"Expected COTP CC, got {payload[1]:#04x}") + if len(payload) < 7: + raise S7ConnectionError(f"COTP CC response too short: {len(payload)} bytes") + if payload[1] != _COTP_CC: + raise S7ConnectionError(f"Expected COTP CC, got {payload[1]:#04x}") async def _init_ssl(self) -> None: """Send InitSSL request (required before CreateObject).""" @@ -1031,8 +1092,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}") @@ -1084,7 +1145,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). @@ -1182,7 +1243,7 @@ async def _recv_cotp_dt(self) -> bytes: async def _send_cotp_raw(self, data: bytes) -> None: """Send raw bytes wrapped in COTP DT + TPKT (no TLS).""" if self._writer is None: - raise RuntimeError("Not connected") + raise S7ConnectionError("Not connected") cotp_dt = struct.pack(">BBB", 2, _COTP_DT, 0x80) + data tpkt = struct.pack(">BBH", 3, 0, 4 + len(cotp_dt)) + cotp_dt @@ -1192,14 +1253,16 @@ async def _send_cotp_raw(self, data: bytes) -> None: async def _recv_cotp_raw(self) -> bytes: """Receive one TPKT + COTP DT frame and return the payload (no TLS).""" if self._reader is None: - raise RuntimeError("Not connected") + raise S7ConnectionError("Not connected") tpkt_header = await self._reader.readexactly(4) _, _, length = struct.unpack(">BBH", tpkt_header) payload = await self._reader.readexactly(length - 4) - if len(payload) < 3 or payload[1] != _COTP_DT: - raise RuntimeError(f"Expected COTP DT, got {payload[1]:#04x}") + if len(payload) < 3: + raise S7ConnectionError(f"COTP DT response too short: {len(payload)} bytes") + if payload[1] != _COTP_DT: + raise S7ConnectionError(f"Expected COTP DT, got {payload[1]:#04x}") return payload[3:] diff --git a/s7commplus/client.py b/s7commplus/client.py index b36a9f28..be0a4cb9 100644 --- a/s7commplus/client.py +++ b/s7commplus/client.py @@ -8,7 +8,7 @@ from collections.abc import Callable, Sequence from typing import Any, Optional, TypeAlias, TypeVar -from snap7.error import S7ConnectionError +from snap7.error import S7ConnectionError, S7ProtocolError from . import typeinfo from .alarm import ( @@ -688,8 +688,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 diff --git a/s7commplus/connection.py b/s7commplus/connection.py index 7fc63495..39a73833 100644 --- a/s7commplus/connection.py +++ b/s7commplus/connection.py @@ -121,6 +121,87 @@ def _log_create_object_return_value(return_value: int, tls_active: bool) -> None # set_ecdh_curve call here would silently overwrite it). _S7_PREFERRED_GROUPS = ("X25519",) +_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: @@ -885,13 +966,23 @@ 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") resp_payload = bytes(data[10:]) if self._session_key is not None: @@ -900,11 +991,6 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: resp_payload = resp_payload[iid_consumed:] return resp_payload - # Receive response - response_frame = self._recv_s7_data() - while self._is_notification_frame(response_frame): - self._notification_frames.append(response_frame) - 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 @@ -922,11 +1008,6 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: 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: - 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: @@ -943,6 +1024,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). @@ -966,6 +1053,25 @@ def send_request(self, function_code: int, payload: bytes = b"", integrity_tail: return resp_payload + def _recv_response_frame(self) -> bytes: + """Receive the next response, queueing notifications and consuming non-fatal SystemEvents.""" + from snap7.error import S7ProtocolError + + system_events = 0 + while True: + response_frame = self._recv_s7_data() + version, data_length, consumed = decode_header(response_frame) + if version == ProtocolVersion.SYSTEM_EVENT: + _check_system_event(bytes(response_frame[consumed : consumed + data_length])) + system_events += 1 + if system_events > _MAX_SYSTEM_EVENTS_PER_RESPONSE: + raise S7ProtocolError("Too many S7CommPlus SystemEvents while waiting for a response") + continue + if self._is_notification_frame(response_frame): + self._notification_frames.append(response_frame) + continue + return response_frame + @staticmethod def _is_notification_frame(frame: bytes) -> bool: """Return whether a complete frame contains an unsolicited notification.""" @@ -1004,7 +1110,7 @@ def receive_notification(self) -> bytes: _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 @@ -1015,7 +1121,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: @@ -1428,13 +1534,25 @@ 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 the captured GET_VAR_SUBSTREAMED layout for legacy sessions. + """Build the protocol-specific captured GET_VAR_SUBSTREAMED layout. - TIA's V1-initial requests use the same zero-valued VLQ qualifier and - two-byte request field as the reference driver's V2 requests. Using - a fixed-width qualifier plus that field adds two bytes (#872). - IntegrityId is inserted before the final four-byte fill. + S7-1200 SessionKey captures use the frame sequence as a fixed-width + ObjectQualifier, a VLQ request field, and a three-byte fill. S7-1500 + captures use a zero-valued VLQ qualifier, a two-byte request field, + and a four-byte fill. The IntegrityId is inserted before that fill. """ + from .session_auth.keys import KeyFamily + + if self._session_auth_family == KeyFamily.S7_1200: + 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 return _build_get_var_substreamed_payload( in_object_id, address, @@ -1481,10 +1599,12 @@ def _post_auth_legitimation(self, password: str = "") -> None: """ # Step 1: Read legitimation challenge from session, address 303 logger.debug("Post-auth legitimation: reading challenge from address 303") + from .session_auth.keys import KeyFamily + 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 if self._session_auth_family == KeyFamily.S7_1200 else 4, ) # Never substitute the earlier CreateObject challenge when this read diff --git a/tests/test_s7_unit.py b/tests/test_s7_unit.py index 970edb81..ea64f3e1 100644 --- a/tests/test_s7_unit.py +++ b/tests/test_s7_unit.py @@ -1,5 +1,7 @@ """Unit tests for S7CommPlus client payload builders, connection parsing, and error paths.""" +import hashlib +import hmac import struct from unittest.mock import MagicMock, call @@ -242,7 +244,7 @@ def test_write_payload_encodes_explicit_datatype(self) -> None: @pytest.mark.parametrize(("with_integrity", "integrity_id"), [(False, 0), (True, 7)]) def test_connection_conditionally_inserts_integrity_id(self, with_integrity: bool, integrity_id: int) -> None: payload = _build_read_payload([(1, 0, 4)]) - response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 1, 0x34) + response = struct.pack(">BHHHHB", Opcode.RESPONSE, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 0, 0x34) connection = S7CommPlusConnection("127.0.0.1") connection._connected = True connection._protocol_version = ProtocolVersion.V2 @@ -712,6 +714,18 @@ 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: + digest = hmac.new(conn._session_key, data, hashlib.sha256).digest() + protected = bytes([len(digest)]) + digest + 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 d66cb358..9b70719e 100644 --- a/tests/test_s7_v2.py +++ b/tests/test_s7_v2.py @@ -4,14 +4,17 @@ and V2 connection behavior. """ +import asyncio import hashlib +import hmac import logging import struct -from unittest.mock import AsyncMock, MagicMock +from unittest.mock import AsyncMock, MagicMock, patch import pytest from s7commplus.async_client import S7CommPlusAsyncClient +from s7commplus.client import S7CommPlusClient from s7commplus.codec import encode_header, encode_object_qualifier from s7commplus.connection import ( S7CommPlusConnection, @@ -43,7 +46,102 @@ ) from s7commplus.server import S7CommPlusServer from s7commplus.vlq import decode_uint32_vlq, encode_uint32_vlq -from snap7.error import S7ConnectionError +from snap7.error import S7ConnectionError, S7ProtocolError + + +class TestTypeInfoRidErrorHandling: + def test_sync_returns_rid(self) -> None: + client = S7CommPlusClient() + with patch.object(client, "read_symbolic", return_value=struct.pack(">I", 0x12345678)): + assert client._read_typeinfo_rid(1) == 0x12345678 + + def test_sync_returns_zero_for_short_read(self) -> None: + client = S7CommPlusClient() + with patch.object(client, "read_symbolic", return_value=b"\x01\x02\x03"): + assert client._read_typeinfo_rid(1) == 0 + + def test_sync_returns_zero_for_unreadable_db(self) -> None: + client = S7CommPlusClient() + with patch.object(client, "read_symbolic", side_effect=RuntimeError("Symbolic read failed")): + assert client._read_typeinfo_rid(1) == 0 + + @pytest.mark.parametrize("error", [S7ConnectionError("reset"), S7ProtocolError("fatal event")]) + def test_sync_propagates_transport_and_protocol_errors(self, error: Exception) -> None: + client = S7CommPlusClient() + with patch.object(client, "read_symbolic", side_effect=error), pytest.raises(type(error), match=str(error)): + client._read_typeinfo_rid(1) + + @pytest.mark.asyncio + async def test_async_returns_rid(self) -> None: + client = S7CommPlusAsyncClient() + with patch.object(client, "read_symbolic", new=AsyncMock(return_value=struct.pack(">I", 0x12345678))): + assert await client._read_typeinfo_rid(1) == 0x12345678 + + @pytest.mark.asyncio + async def test_async_returns_zero_for_short_read(self) -> None: + client = S7CommPlusAsyncClient() + with patch.object(client, "read_symbolic", new=AsyncMock(return_value=b"\x01\x02\x03")): + assert await client._read_typeinfo_rid(1) == 0 + + @pytest.mark.asyncio + async def test_async_returns_zero_for_unreadable_db(self) -> None: + client = S7CommPlusAsyncClient() + error = RuntimeError("Symbolic read failed") + with patch.object(client, "read_symbolic", new=AsyncMock(side_effect=error)): + assert await client._read_typeinfo_rid(1) == 0 + + @pytest.mark.asyncio + @pytest.mark.parametrize("error", [S7ConnectionError("reset"), S7ProtocolError("fatal event")]) + async def test_async_propagates_transport_and_protocol_errors(self, error: Exception) -> None: + client = S7CommPlusAsyncClient() + with ( + patch.object(client, "read_symbolic", new=AsyncMock(side_effect=error)), + pytest.raises(type(error), match=str(error)), + ): + await client._read_typeinfo_rid(1) + + @pytest.mark.asyncio + async def test_async_reconnect_wrapper_retries_connection_error_once(self) -> None: + client = S7CommPlusAsyncClient() + operation = AsyncMock(side_effect=[S7ConnectionError("reset"), 42]) + client._reconnect = AsyncMock() # type: ignore[method-assign] + + assert await client._with_reconnect(operation) == 42 + client._reconnect.assert_awaited_once() + assert operation.await_count == 2 + + @pytest.mark.asyncio + async def test_async_reconnect_wrapper_does_not_retry_protocol_error(self) -> None: + client = S7CommPlusAsyncClient() + operation = AsyncMock(side_effect=S7ProtocolError("fatal event")) + client._reconnect = AsyncMock() # type: ignore[method-assign] + + with pytest.raises(S7ProtocolError, match="fatal event"): + await client._with_reconnect(operation) + client._reconnect.assert_not_awaited() + + @pytest.mark.asyncio + async def test_async_request_disconnected_is_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + with pytest.raises(S7ConnectionError, match="Not connected"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"") + + @pytest.mark.asyncio + async def test_async_reconnect_without_parameters_is_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + with pytest.raises(S7ConnectionError, match="Not connected"): + await client._reconnect() + + @pytest.mark.asyncio + async def test_async_non_data_cotp_frame_is_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + client._reader = asyncio.StreamReader() + payload = bytes.fromhex("02e080") + client._reader.feed_data(struct.pack(">BBH", 3, 0, len(payload) + 4) + payload) + client._reader.feed_eof() + + with pytest.raises(S7ConnectionError, match="Expected COTP DT"): + await client._recv_cotp_raw() class TestReadFunctionCodes: @@ -302,7 +400,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) @@ -315,6 +413,141 @@ def test_tls_v2_response_application_payload_is_not_stripped(self) -> None: # GetMultiVariables is in FLAGS_34_FUNCTION_CODES assert conn._send_s7_data.call_args[0][0][17] == 0x34 + 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 + conn._session_key = bytes(range(24)) + + 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 += encode_uint32_vlq(0) + application_payload + digest = hmac.new(conn._session_key, response, hashlib.sha256).digest() + protected_response = bytes([len(digest)]) + digest + response + response_frame = encode_header(ProtocolVersion.V3, len(protected_response)) + protected_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 + + @pytest.mark.asyncio + async def test_too_many_async_system_events_raise_protocol_error_from_typeinfo_read(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 + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=event_frame) + + with pytest.raises(S7ProtocolError, match="Too many S7CommPlus SystemEvents"): + await client._read_typeinfo_rid(0x8A0E0001) + + @pytest.mark.asyncio + @pytest.mark.parametrize("reassemble", [False, True]) + async def test_async_short_response_raises_connection_error(self, reassemble: bool) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = b"short" + 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(return_value=response_frame) + + with pytest.raises(S7ConnectionError, match="Response too short"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + + @pytest.mark.asyncio + @pytest.mark.parametrize("reassemble", [False, True]) + async def test_async_sequence_mismatch_raises_protocol_error(self, reassemble: bool) -> None: + client = S7CommPlusAsyncClient() + client._connected = True + client._reader = MagicMock() + client._writer = MagicMock() + client._protocol_version = ProtocolVersion.V2 + response = struct.pack(">BHHHHB", 0x32, 0, FunctionCode.GET_MULTI_VARIABLES, 0, 99, 0x34) + 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(return_value=response_frame) + + with pytest.raises(S7ProtocolError, match="Response sequence mismatch"): + await client._send_request(FunctionCode.GET_MULTI_VARIABLES, b"", reassemble=reassemble) + + +class TestAsyncReassembledPayloadErrors: + @pytest.mark.asyncio + async def test_closed_connection_raises_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + client._recv_cotp_dt = AsyncMock(return_value=b"") + + with pytest.raises(S7ConnectionError, match="closed during"): + await client._recv_reassembled_payload() + + @pytest.mark.asyncio + async def test_bad_fragment_header_raises_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + + with pytest.raises(S7ConnectionError, match="fragment header"): + await client._recv_reassembled_payload(b"\x99\x02\x00\x01x") + + @pytest.mark.asyncio + async def test_fragment_limit_raises_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + client._MAX_REASSEMBLED_FRAGMENTS = 1 + fragments = b"\x72\x02\x00\x01a\x72\x02\x00\x01b\x72\x02\x00\x00" + + with pytest.raises(S7ConnectionError, match="exceeds limits"): + await client._recv_reassembled_payload(fragments) + class TestServerResponseIntegrityId: """Test V2 response IntegrityId selection and encoding.""" @@ -439,6 +672,37 @@ 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 + conn._session_auth_family = 1 + + 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._session_auth_family = 1 + challenge = bytes(range(20)) + challenge_response = bytes([0x00, 0x00, 0x10, DataType.USINT, len(challenge)]) + challenge + bytes([0x00]) + conn.send_request = MagicMock(side_effect=[challenge_response, 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]) @@ -623,7 +887,7 @@ async def test_async_request_shape(self) -> None: client._send_cotp_dt = AsyncMock() client._recv_cotp_dt = AsyncMock(return_value=bytes.fromhex("72010000")) - with pytest.raises(RuntimeError, match="CreateObject response too short"): + with pytest.raises(S7ConnectionError, match="CreateObject response too short"): await client._create_session() client._send_cotp_dt.assert_awaited_once() @@ -645,6 +909,26 @@ async def test_async_request_shape(self) -> None: assert frame[-4:] == struct.pack(">BBH", 0x72, ProtocolVersion.V1, 0x0000) +class TestInitSSLResponse: + @pytest.mark.asyncio + async def test_async_short_response_raises_connection_error(self) -> None: + client = S7CommPlusAsyncClient() + client._send_cotp_dt = AsyncMock() + client._recv_cotp_dt = AsyncMock(return_value=bytes.fromhex("72010000")) + + with pytest.raises(S7ConnectionError, match="InitSSL response too short"): + await client._init_ssl() + + @pytest.mark.asyncio + async def test_async_ten_byte_response_is_accepted(self) -> None: + client = S7CommPlusAsyncClient() + client._send_cotp_dt = AsyncMock() + response = bytes(10) + client._recv_cotp_dt = AsyncMock(return_value=encode_header(ProtocolVersion.V1, len(response)) + response) + + await client._init_ssl() + + class TestDeleteSessionRequest: """The DeleteObject request that closes an S7CommPlus session."""