fix: accept the null acknowledgement real PLCs send for set clock and session passwords

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
 -----

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
 

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
 

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
 

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,14 +2203,18 @@ 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.
 
         Args:
             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(

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()