diff --git a/CHANGES.md b/CHANGES.md index 2a8dac85..73346d4f 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -33,6 +33,9 @@ Unreleased * `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). +* Compress now uses the PI service `_GARB` and copy RAM to ROM uses file system `EP` with `_MODU`, matching real + engineering tools and native Snap7, instead of the invented `_MSZL`. The bundled server recognises the new + requests (#941). 3.2.1 ----- diff --git a/snap7/client.py b/snap7/client.py index 7092e763..dbfe8472 100644 --- a/snap7/client.py +++ b/snap7/client.py @@ -1917,7 +1917,7 @@ def compress(self, timeout: int) -> int: """ Compress PLC memory. - Sends real S7 PLC_CONTROL protocol with PI service "_MSZL". + Sends real S7 PLC_CONTROL protocol with PI service "_GARB". Args: timeout: Timeout in milliseconds (used for receive timeout) @@ -1940,7 +1940,7 @@ def copy_ram_to_rom(self, timeout: int = 0) -> int: """ Copy RAM to ROM. - Sends real S7 PLC_CONTROL protocol with PI service "_MSZL" and file ID "P". + Sends real S7 PLC_CONTROL protocol with PI service "_MODU" and file system "EP". Args: timeout: Timeout in milliseconds (used for receive timeout) diff --git a/snap7/s7protocol.py b/snap7/s7protocol.py index 506ce6cf..8791d65a 100644 --- a/snap7/s7protocol.py +++ b/snap7/s7protocol.py @@ -114,6 +114,10 @@ def get_return_code_description(return_code: int) -> str: _HOT_START_PARAMS = bytes.fromhex("28000000000000fd000009") + b"P_PROGRAM" # Cold start: same as hot start but with block "C " before PI service _COLD_START_PARAMS = bytes.fromhex("28000000000000fd0002432009") + b"P_PROGRAM" +# Compress: function(0x28) + 7 reserved + block count(0) + PI length(5) + "_GARB" (as native Snap7 TReqFunCompress) +_COMPRESS_PARAMS = bytes.fromhex("28000000000000fd000005") + b"_GARB" +# Copy RAM to ROM: like compress but with file system "EP" before PI service "_MODU" (native TReqFunCopyRamToRom) +_COPY_RAM_TO_ROM_PARAMS = bytes.fromhex("28000000000000fd0002455005") + b"_MODU" class S7Protocol: @@ -402,7 +406,7 @@ def build_plc_control_request(self, operation: str) -> bytes: Build PLC control request using the S7 PI service PDU format. Args: - operation: Control operation ('stop', 'hot_start', 'cold_start') + operation: Control operation ('stop', 'hot_start', 'cold_start', 'compress', 'copy_ram_to_rom') Returns: Complete S7 PDU for PLC control @@ -411,6 +415,8 @@ def build_plc_control_request(self, operation: str) -> bytes: "stop": _STOP_PARAMS, "hot_start": _HOT_START_PARAMS, "cold_start": _COLD_START_PARAMS, + "compress": _COMPRESS_PARAMS, + "copy_ram_to_rom": _COPY_RAM_TO_ROM_PARAMS, } if operation not in params: @@ -449,87 +455,23 @@ def build_compress_request(self) -> bytes: """ Build PLC control request for memory compression. - Uses PI service "_MSZL" (compress memory). + Uses PI service "_GARB" (compress memory), as native Snap7 does. Returns: Complete S7 PDU for compress request """ - # PI service command for compress - pi_service = b"_MSZL" - - # Parameter section: function code + PI service - # Format: func(1) + unknown(7) + pi_len(1) + pi_service - param_data = ( - struct.pack( - ">BBBBBBBBB", - S7Function.PLC_CONTROL, # 0x28 - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - len(pi_service), # PI service length - ) - + pi_service - ) - - header = struct.pack( - ">BBHHHH", - 0x32, # Protocol ID - S7PDUType.REQUEST, # PDU type - 0x0000, # Reserved - self._next_sequence(), # Sequence - len(param_data), # Parameter length - 0x0000, # Data length - ) - - return header + param_data + return self.build_plc_control_request("compress") def build_copy_ram_to_rom_request(self) -> bytes: """ Build PLC control request for copying RAM to ROM. - Uses PI service "_MSZL" with file system parameters. + Uses PI service "_MODU" with the "EP" file system, as native Snap7 does. Returns: Complete S7 PDU for copy RAM to ROM request """ - # PI service command for copy RAM to ROM - # Uses EP parameter for target file system - pi_service = b"_MSZL" - file_id = b"P" # P = passive file system (ROM) - - # Parameter section with file system identifier - param_data = ( - struct.pack( - ">BBBBBBBBB", - S7Function.PLC_CONTROL, # 0x28 - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - 0x00, # Reserved - len(file_id), # File ID length - len(pi_service), # PI service length - ) - + file_id - + pi_service - ) - - header = struct.pack( - ">BBHHHH", - 0x32, # Protocol ID - S7PDUType.REQUEST, # PDU type - 0x0000, # Reserved - self._next_sequence(), # Sequence - len(param_data), # Parameter length - 0x0000, # Data length - ) - - return header + param_data + return self.build_plc_control_request("copy_ram_to_rom") # ======================================================================== # Block Transfer PDU Builders (Upload/Download) diff --git a/snap7/server/__init__.py b/snap7/server/__init__.py index c8401d1e..fa82e0ac 100644 --- a/snap7/server/__init__.py +++ b/snap7/server/__init__.py @@ -1129,14 +1129,10 @@ def _handle_plc_control(self, request: Dict[str, Any], client_address: Tuple[str pi_service = params.get("pi_service", b"") # Check for PI service operations - if pi_service == b"_MSZL": - file_id = params.get("file_id", b"") - if file_id == b"P": - # Copy RAM to ROM - logger.info(f"Copy RAM to ROM requested from {client_address}") - else: - # Compress memory - logger.info(f"Compress memory requested from {client_address}") + if pi_service == b"_MODU": + logger.info(f"Copy RAM to ROM requested from {client_address}") + elif pi_service == b"_GARB": + logger.info(f"Compress memory requested from {client_address}") elif len(params) >= 2: # Has restart type parameter - start operation restart_type = params.get("restart_type", 1) @@ -1411,9 +1407,10 @@ def _parse_request_parameters(self, param_data: bytes) -> Dict[str, Any]: if b"P_PROGRAM" in param_data: restart_type = 2 if b"C " in param_data else 1 return {"function_code": function_code, "restart_type": restart_type, "pi_service": b"P_PROGRAM"} - elif b"_MSZL" in param_data: - file_id = b"P" if b"P_MSZL" in param_data else b"" - return {"function_code": function_code, "pi_service": b"_MSZL", "file_id": file_id} + elif b"_MODU" in param_data: + return {"function_code": function_code, "pi_service": b"_MODU", "file_id": b"EP"} + elif b"_GARB" in param_data: + return {"function_code": function_code, "pi_service": b"_GARB"} elif b"_DELE" in param_data: return {"function_code": function_code, "pi_service": b"_DELE"} if len(param_data) >= 2 and param_data[1] in (1, 2): diff --git a/tests/test_s7protocol.py b/tests/test_s7protocol.py index 32f6d252..049748d4 100644 --- a/tests/test_s7protocol.py +++ b/tests/test_s7protocol.py @@ -274,6 +274,16 @@ def test_set_clock_year_range(self) -> None: with pytest.raises(ValueError, match="1990-2089"): self.proto.build_set_clock_request(datetime(year, 1, 1)) + def test_compress_request_matches_real_job(self) -> None: + # Parameter block sent by an engineering tool (CISA icsnpp-s7comm snap7.pcap): _GARB + request = self.proto.build_compress_request() + assert request[10:] == bytes.fromhex("28000000000000fd0000055f47415242") + + def test_copy_ram_to_rom_request_matches_real_job(self) -> None: + # Parameter block sent by an engineering tool (CISA icsnpp-s7comm snap7.pcap): file system EP, _MODU + request = self.proto.build_copy_ram_to_rom_request() + assert request[10:] == bytes.fromhex("28000000000000fd00024550055f4d4f4455") + def test_request_db_download(self) -> None: assert self.proto.build_download_request(0x41, 1, bytes(64)).hex() == ( "320100000001002000001a00010000000000095f30413030303031500d31303030303634303030303238"