diff --git a/pymodbus/pdu/other_message.py b/pymodbus/pdu/other_message.py index 02323f590..82fed8b21 100644 --- a/pymodbus/pdu/other_message.py +++ b/pymodbus/pdu/other_message.py @@ -6,6 +6,7 @@ from ..constants import ModbusStatus from ..datastore import ModbusServerContext +from ..exceptions import ModbusIOException from .decoders import DecodePDU from .device import DeviceInformationFactory, ModbusControlBlock from .pdu import ModbusPDU @@ -243,8 +244,14 @@ def decode(self, data: bytes) -> None: raw value that a user can decode to whatever it should be. """ self.byte_count = int(data[0]) - self.identifier = data[1 : self.byte_count + 1] - status = int(data[-1]) + if not 1 <= self.byte_count <= len(data) - 1: + raise ModbusIOException( + f"byte_count {self.byte_count} outside 1..{len(data) - 1} " + f"for packet of length {len(data)}", + function_code=self.function_code, + ) + self.identifier = data[1 : self.byte_count] + status = int(data[self.byte_count]) self.status = status == ID_ON diff --git a/test/pdu/test_other_messages.py b/test/pdu/test_other_messages.py index 6e14ed1c0..266f77758 100644 --- a/test/pdu/test_other_messages.py +++ b/test/pdu/test_other_messages.py @@ -3,7 +3,10 @@ from typing import cast from unittest import mock +import pytest + import pymodbus.pdu.other_message as pymodbus_message +from pymodbus.exceptions import ModbusIOException class TestOtherMessage: @@ -155,9 +158,28 @@ async def test_report_device_id(self, mock_server_context): ) assert response.encode() == b"\tPymodbus\xff" - response.decode(b"\x03\x12\x00") + response.decode(b"\tPymodbus\xff") + assert response.status + assert response.identifier == b"Pymodbus" + response.decode(b"\x03\x12\x00\x00") assert not response.status assert response.identifier == b"\x12\x00" response.status = False assert response.encode() == b"\x03\x12\x00\x00" + + def test_report_device_id_response_roundtrip(self): + """Test decode(encode()) keeps identifier and status unchanged.""" + for identifier, status in ((b"Pymodbus", True), (b"\x12", False)): + sent = pymodbus_message.ReportDeviceIdResponse(identifier, status) + received = pymodbus_message.ReportDeviceIdResponse() + received.decode(sent.encode()) + assert received.identifier == identifier + assert received.status == status + + def test_report_device_id_response_invalid_byte_count(self): + """Test byte count pointing outside the received data raises.""" + response = pymodbus_message.ReportDeviceIdResponse() + for frame in (b"\x00\x12\xff", b"\x04\x12\xff"): + with pytest.raises(ModbusIOException): + response.decode(frame)