Fix: validate byte count in file_message.py decode handlers

pymodbus/pdu/file_message.py

@@ -63,7 +63,8 @@ def encode(self) -> bytes:
     def decode(self, data: bytes) -> None:
         """Decode the incoming request."""
         self.records = []
-        byte_count = int(data[0])
+        if (byte_count := int(data[0])) > len(data) - 1:
+            raise ModbusException(f"Invalid byte count: {byte_count}")
         for count in range(1, byte_count, 7):
             decoded = struct.unpack(">BHHH", data[count : count + 7])
             record = FileRecord(
@@ -169,7 +170,8 @@ def encode(self) -> bytes:
 
     def decode(self, data: bytes) -> None:
         """Decode the incoming request."""
-        byte_count = int(data[0])
+        if (byte_count := int(data[0])) > len(data) - 1:
+            raise ModbusException(f"Invalid byte count: {byte_count}")
         count = 1
         self.records.clear()
         while count < byte_count:
@@ -314,6 +316,8 @@ def decode(self, data: bytes) -> None:
         """Decode a the response."""
         self.values = []
         _, count = struct.unpack(">HH", data[0:4])
+        if 4 + count * 2 > len(data):
+            raise ModbusException(f"Invalid fifo count: {count}")
         for index in range(0, count):
             idx = 4 + index * 2
             self.values.append(struct.unpack(">H", data[idx : idx + 2])[0])

test/pdu/test_file_message.py

@@ -216,6 +216,26 @@ def test_write_file_record_response_decode(self):
         handle.decode(request)
         # assert handle.records[0] == record
 
+    def test_read_file_record_request_decode_invalid_byte_count(self):
+        """Test ReadFileRecordRequest raises ModbusException on oversized byte_count."""
+        handle = ReadFileRecordRequest()
+        with pytest.raises(ModbusException):
+            handle.decode(bytes([0xFF, 0x06, 0x00, 0x01, 0x00, 0x01, 0x00, 0x01]))
+
+    def test_write_file_record_request_decode_invalid_byte_count(self):
+        """Test WriteFileRecordRequest raises ModbusException on oversized byte_count."""
+        handle = WriteFileRecordRequest()
+        with pytest.raises(ModbusException):
+            handle.decode(
+                bytes([0xFF, 0x06, 0x00, 0x01, 0x00, 0x00, 0x00, 0x01, 0x00, 0x00])
+            )
+
+    def test_read_fifo_queue_response_decode_invalid_count(self):
+        """Test ReadFifoQueueResponse raises ModbusException on oversized count."""
+        handle = ReadFifoQueueResponse()
+        with pytest.raises(ModbusException):
+            handle.decode(bytes([0x00, 0x00, 0xFF, 0xFF]))
+
     def test_write_file_record_response_frame_size(self):
         """Test write file record response rtu frame size calculation."""
         request = b"\x00\x00\x0d\x06\x00\x04\x00\x07\x00\x03\x06\xaf\x04\xbe\x10\x0d"