diff --git a/CHANGES.md b/CHANGES.md index 082a6dfd..2a8dac85 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -30,6 +30,9 @@ Unreleased * `parse_list_blocks_response` now requires exactly the seven four-byte block-type entries (28 bytes) that native Snap7 expects, with distinct, known type codes. Empty or partial tables and duplicate entries are rejected instead of being reported as zero block counts (#939). +* `set_plc_datetime` and the session-password calls now accept the null acknowledgement (return code 0x0A, no data) that + real PLCs send for services that return no data, instead of failing with "Object does not exist". The bundled + server acknowledges these services the same way (#940). 3.2.1 ----- diff --git a/snap7/async_client.py b/snap7/async_client.py index a3238906..9adbee18 100644 --- a/snap7/async_client.py +++ b/snap7/async_client.py @@ -1318,7 +1318,9 @@ async def set_plc_datetime(self, dt: datetime) -> int: request = self.protocol.build_set_clock_request(dt) response = await self._send_receive(request) - self.protocol.check_userdata_response(response, S7UserDataGroup.TIME, S7UserDataSubfunction.SET_CLOCK) + self.protocol.check_userdata_response( + response, S7UserDataGroup.TIME, S7UserDataSubfunction.SET_CLOCK, accept_null_ack=True + ) logger.info(f"Set PLC datetime to {dt}") return 0 @@ -1514,7 +1516,9 @@ async def set_session_password(self, password: str) -> int: encoded = self.protocol.encode_password(password) request = self.protocol.build_set_session_password_request(encoded) response = await self._send_receive(request) - self.protocol.check_userdata_response(response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.SET_SESSION_PASSWORD) + self.protocol.check_userdata_response( + response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.SET_SESSION_PASSWORD, accept_null_ack=True + ) logger.info("Session password set successfully") return 0 @@ -1535,7 +1539,9 @@ async def clear_session_password(self) -> int: request = self.protocol.build_clear_session_password_request() response = await self._send_receive(request) - self.protocol.check_userdata_response(response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.CLEAR_SESSION_PASSWORD) + self.protocol.check_userdata_response( + response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.CLEAR_SESSION_PASSWORD, accept_null_ack=True + ) logger.info("Session password cleared successfully") return 0 diff --git a/snap7/client.py b/snap7/client.py index 288ec4bd..7092e763 100644 --- a/snap7/client.py +++ b/snap7/client.py @@ -1892,7 +1892,9 @@ def set_plc_datetime(self, dt: datetime) -> int: # Build and send set clock request request = self.protocol.build_set_clock_request(dt) response = self._send_receive(request) - self.protocol.check_userdata_response(response, S7UserDataGroup.TIME, S7UserDataSubfunction.SET_CLOCK) + self.protocol.check_userdata_response( + response, S7UserDataGroup.TIME, S7UserDataSubfunction.SET_CLOCK, accept_null_ack=True + ) logger.info(f"Set PLC datetime to {dt}") return 0 @@ -2093,7 +2095,9 @@ def build_request() -> bytes: return self.protocol.build_set_session_password_request(encoded) response = self._send_receive_with_reconnect(build_request) - self.protocol.check_userdata_response(response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.SET_SESSION_PASSWORD) + self.protocol.check_userdata_response( + response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.SET_SESSION_PASSWORD, accept_null_ack=True + ) logger.info("Session password set successfully") return 0 @@ -2116,7 +2120,9 @@ def build_request() -> bytes: return self.protocol.build_clear_session_password_request() response = self._send_receive_with_reconnect(build_request) - self.protocol.check_userdata_response(response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.CLEAR_SESSION_PASSWORD) + self.protocol.check_userdata_response( + response, S7UserDataGroup.SECURITY, S7UserDataSubfunction.CLEAR_SESSION_PASSWORD, accept_null_ack=True + ) logger.info("Session password cleared successfully") return 0 diff --git a/snap7/s7protocol.py b/snap7/s7protocol.py index d75fbee8..506ce6cf 100644 --- a/snap7/s7protocol.py +++ b/snap7/s7protocol.py @@ -1488,6 +1488,8 @@ def check_userdata_response( response: Dict[str, Any], expected_group: int | None = None, expected_subfunction: int | None = None, + *, + accept_null_ack: bool = False, ) -> None: """Check a USERDATA response for errors. @@ -1496,6 +1498,11 @@ def check_userdata_response( Args: response: Parsed S7 response from :meth:`parse_response`. + expected_group: Expected function group, if it should be verified. + expected_subfunction: Expected subfunction, if it should be verified. + accept_null_ack: For services that return no data (set clock, session passwords). Real PLCs + acknowledge these with return code 0x0A, transport size 0 and length 0 next to a zero + parameter error code, which is then a success. Raises: ~snap7.error.S7ProtocolError: If the response indicates an error. @@ -1516,7 +1523,13 @@ def check_userdata_response( data_info = response.get("data", {}) if isinstance(data_info, dict): return_code = data_info.get("return_code", 0xFF) - if return_code != 0xFF: + null_ack = ( + accept_null_ack + and return_code == 0x0A + and data_info.get("transport_size", 0) == 0 + and data_info.get("data_length", 0) == 0 + ) + if return_code != 0xFF and not null_ack: desc = get_return_code_description(return_code) raise S7ProtocolError(f"USERDATA request failed: {desc} (0x{return_code:02x})") else: @@ -1653,7 +1666,9 @@ def parse_response(self, pdu: bytes) -> Dict[str, Any]: raise S7ProtocolError("S7 response contains trailing bytes") if pdu_type == S7PDUType.USERDATA: - self.check_userdata_response(response) + # The null acknowledgement is only a success for no-data services; callers that expect data + # run check_userdata_response again without accept_null_ack. + self.check_userdata_response(response, accept_null_ack=True) return response diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index da6fb92c..c8401d1e 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -1945,8 +1945,8 @@ def from_bcd(value: int) -> int: logger.info(f"Set clock from {client_address}: {timestamp}") - # Return success (empty response data) - return self._build_userdata_success_response(request, userdata_params, b"") + # Return success (null acknowledgement, as real PLCs send for set clock) + return self._build_userdata_success_response(request, userdata_params, b"", null_ack=True) def _handle_security( self, request: Dict[str, Any], userdata_params: Dict[str, Any], client_address: Tuple[str, int] @@ -1958,7 +1958,7 @@ def _handle_security( """ logger.debug(f"Security request from {client_address} (returning success)") # Return success - emulator doesn't require password - return self._build_userdata_success_response(request, userdata_params, b"") + return self._build_userdata_success_response(request, userdata_params, b"", null_ack=True) def _handle_list_all_blocks( self, request: Dict[str, Any], userdata_params: Dict[str, Any], client_address: Tuple[str, int] @@ -2203,7 +2203,9 @@ def _build_userdata_error_response(self, request: Dict[str, Any], error_code: in return header + param_data + data_section - def _build_userdata_success_response(self, request: Dict[str, Any], userdata_params: Dict[str, Any], data: bytes) -> bytes: + def _build_userdata_success_response( + self, request: Dict[str, Any], userdata_params: Dict[str, Any], data: bytes, null_ack: bool = False + ) -> bytes: """ Build USER_DATA success response PDU. @@ -2211,6 +2213,8 @@ def _build_userdata_success_response(self, request: Dict[str, Any], userdata_par request: Original request userdata_params: Parsed USER_DATA parameters data: Response data + null_ack: Acknowledge with the no-data form a real PLC uses for services that return nothing + (return code 0x0A, transport size 0, length 0) instead of 0xFF/0x09 Returns: Success response PDU @@ -2237,7 +2241,10 @@ def _build_userdata_success_response(self, request: Dict[str, Any], userdata_par ) # Data section: return code (0xFF = success) + data - data_section = struct.pack(">BBH", 0xFF, 0x09, len(data)) + data + if null_ack: + data_section = struct.pack(">BBH", 0x0A, 0x00, 0) + else: + data_section = struct.pack(">BBH", 0xFF, 0x09, len(data)) + data # Build S7 header for USERDATA (10 bytes, no error_class/error_code in header) header = struct.pack( diff --git a/tests/test_s7protocol.py b/tests/test_s7protocol.py index 7ac519b6..32f6d252 100644 --- a/tests/test_s7protocol.py +++ b/tests/test_s7protocol.py @@ -541,6 +541,34 @@ def test_zero_error_word_is_accepted(self) -> None: assert proto.extract_read_data(response, S7WordLen.BYTE, 4) == [0x11, 0x22, 0x33, 0x44] +class TestUserdataNullAck: + # Captured from a real S7-300 (Wireshark sample s7comm_reading_setting_plc_time.pcap, frame 44) + SET_CLOCK_ACK = bytes.fromhex("320700000c00000c00040001120812870201000000000a000000") + + def test_null_ack_accepted_for_set_clock(self) -> None: + proto = S7Protocol() + response = proto.parse_response(self.SET_CLOCK_ACK) + proto.check_userdata_response(response, S7UserDataGroup.TIME, S7UserDataSubfunction.SET_CLOCK, accept_null_ack=True) + + def test_null_ack_rejected_by_default(self) -> None: + proto = S7Protocol() + response = proto.parse_response(self.SET_CLOCK_ACK) + with pytest.raises(S7ProtocolError, match="0x0a"): + proto.check_userdata_response(response, S7UserDataGroup.TIME, S7UserDataSubfunction.SET_CLOCK) + + def test_null_ack_with_parameter_error_rejected(self) -> None: + pdu = bytearray(self.SET_CLOCK_ACK) + pdu[20:22] = bytes([0x81, 0x04]) # parameter error code + with pytest.raises(S7ProtocolError): + S7Protocol().parse_response(bytes(pdu)) + + def test_other_return_codes_still_rejected_with_null_ack(self) -> None: + pdu = bytearray(self.SET_CLOCK_ACK) + pdu[22] = 0x05 # address out of range + with pytest.raises(S7ProtocolError): + S7Protocol().parse_response(bytes(pdu)) + + class TestParseGetClockResponse: def setup_method(self) -> None: self.proto = S7Protocol()