Fix write multiple coils byte count validation

pymodbus/pdu/bit_message.py

@@ -136,6 +136,9 @@ class WriteMultipleCoilsRequest(ModbusPDU):
 
     function_code = 15
     rtu_byte_count_pos = 6
+    count: int
+    byte_count: int | None = None
+    data_byte_count = 0
 
     def encode(self) -> bytes:
         """Encode write coils request."""
@@ -149,14 +152,23 @@ def encode(self) -> bytes:
 
     def decode(self, data: bytes) -> None:
         """Decode a write coils request."""
-        self.address, count, _byte_count = struct.unpack(">HHB", data[0:5])
-        self.bits = unpack_bitstring(data[5:])[:count]
+        self.address, self.count, self.byte_count = struct.unpack(">HHB", data[0:5])
+        self.data_byte_count = len(data) - 5
+        self.bits = unpack_bitstring(data[5 : 5 + self.byte_count])[: self.count]
 
     async def datastore_update(
         self, context: ModbusServerContext, device_id: int
     ) -> ModbusPDU:
         """Run a request against a datastore."""
         count = len(self.bits)
+        if self.byte_count is not None:
+            expected_byte_count = (self.count + 7) // 8
+            if (
+                self.byte_count != expected_byte_count
+                or self.data_byte_count != self.byte_count
+            ):
+                return ExceptionResponse(self.function_code, ExcCodes.ILLEGAL_VALUE)
+            count = self.count
         rc = await context.async_setValues(
             device_id, self.function_code, self.address, self.bits
         )

test/pdu/test_bit.py

@@ -130,6 +130,45 @@ def test_invalid_write_multiple_coils_request(self):
         request = bit_msg.WriteMultipleCoilsRequest(address=1, bits=None)
         assert not request.bits
 
+    @pytest.mark.parametrize(
+        "frame",
+        [
+            b"\x00\x01\x00\x10\x01\xff",
+            b"\x00\x01\x00\x08\x02\xff\x00",
+            b"\x00\x01\x00\x08\x01",
+            b"\x00\x01\x00\x08\x01\xff\x00",
+        ],
+    )
+    async def test_write_multiple_coils_rejects_invalid_byte_count(
+        self, frame, mock_server_context
+    ):
+        """Test write multiple coils rejects inconsistent byte counts."""
+        request = bit_msg.WriteMultipleCoilsRequest()
+        request.decode(frame)
+        context = mock_server_context()
+        context.async_setValues = mock.AsyncMock()
+
+        result = await request.datastore_update(context, 0)
+
+        assert result.exception_code == ExcCodes.ILLEGAL_VALUE
+        context.async_setValues.assert_not_awaited()
+
+    async def test_write_multiple_coils_accepts_valid_byte_count(
+        self, mock_server_context
+    ):
+        """Test write multiple coils accepts a consistent byte count."""
+        request = bit_msg.WriteMultipleCoilsRequest()
+        request.decode(b"\x00\x01\x00\x09\x02\xff\x01")
+        context = mock_server_context()
+        context.async_setValues = mock.AsyncMock(return_value=0)
+
+        result = await request.datastore_update(context, 0)
+
+        assert result.count == 9
+        context.async_setValues.assert_awaited_once_with(
+            0, request.function_code, 1, [True] * 9
+        )
+
     def test_write_single_coil_request_encode(self):
         """Test write single coil."""
         request = bit_msg.WriteSingleCoilRequest(address=1, bits=[False])