From 932e1f0a7ac2c325dd136773bdb93b64eff7449d Mon Sep 17 00:00:00 2001 From: jan iversen Date: Thu, 13 Aug 2026 09:33:45 +0200 Subject: [PATCH] Control count in bit/register PDUs. --- pymodbus/pdu/bit_message.py | 12 +++++++++--- pymodbus/pdu/register_message.py | 12 +++++++++--- 2 files changed, 18 insertions(+), 6 deletions(-) diff --git a/pymodbus/pdu/bit_message.py b/pymodbus/pdu/bit_message.py index 5c263179b..4597d54fc 100644 --- a/pymodbus/pdu/bit_message.py +++ b/pymodbus/pdu/bit_message.py @@ -16,16 +16,18 @@ class ReadCoilsRequest(ModbusPDU): rtu_frame_size = 8 function_code = 1 + MAX_COUNT = 2000 def encode(self) -> bytes: """Encode a request pdu.""" - self.verifyCount(2000) + self.verifyCount(self.MAX_COUNT) self.verifyAddress() return struct.pack(">HH", self.address, self.count) def decode(self, data: bytes) -> None: """Decode a request pdu.""" self.address, self.count = struct.unpack(">HH", data[:4]) + self.verifyCount(self.MAX_COUNT) def get_response_pdu_size(self) -> int: """Get response pdu size. @@ -139,6 +141,7 @@ class WriteMultipleCoilsRequest(ModbusPDU): count: int byte_count: int | None data_byte_count: int + MAX_COUNT = 2000 def __init__( self, @@ -162,7 +165,7 @@ def encode(self) -> bytes: """Encode write coils request.""" self.count = len(self.bits) self.verifyAddress() - self.verifyCount(2000) + self.verifyCount(self.MAX_COUNT) byte_count = (self.count + 7) // 8 return struct.pack( ">HHB", self.address, self.count, byte_count @@ -171,8 +174,11 @@ def encode(self) -> bytes: def decode(self, data: bytes) -> None: """Decode a write coils request.""" self.address, self.count, self.byte_count = struct.unpack(">HHB", data[0:5]) + self.verifyCount(self.MAX_COUNT) self.data_byte_count = len(data) - 5 - self.bits = unpack_bitstring(data[5 : 5 + self.byte_count])[: self.count] + self.bits = unpack_bitstring(data[5 : 5 + cast(int, self.byte_count)])[ + : self.count + ] async def datastore_update( self, context: ModbusServerContext, device_id: int diff --git a/pymodbus/pdu/register_message.py b/pymodbus/pdu/register_message.py index 3bd13e3f2..3832728dc 100644 --- a/pymodbus/pdu/register_message.py +++ b/pymodbus/pdu/register_message.py @@ -19,16 +19,18 @@ class ReadHoldingRegistersRequest(ModbusPDU): function_code = 3 rtu_frame_size = 8 + MAX_COUNT = 125 def encode(self) -> bytes: """Encode the request packet.""" self.verifyAddress() - self.verifyCount(125) + self.verifyCount(self.MAX_COUNT) return struct.pack(">HH", self.address, self.count) def decode(self, data: bytes) -> None: """Decode a register request packet.""" self.address, self.count = struct.unpack(">HH", data[:4]) + self.verifyCount(self.MAX_COUNT) def get_response_pdu_size(self) -> int: """Get response pdu size. @@ -100,6 +102,8 @@ class ReadWriteMultipleRegistersRequest(ModbusPDU): function_code = 23 rtu_byte_count_pos = 10 + MAX_READ_COUNT = 125 + MAX_WRITE_COUNT = 121 def __init__( self, @@ -126,8 +130,8 @@ def encode(self) -> bytes: """Encode the request packet.""" self.verifyAddress(address=self.read_address) self.verifyAddress(address=self.write_address) - self.verifyCount(125, count=self.read_count) - self.verifyCount(121, count=self.write_count) + self.verifyCount(self.MAX_READ_COUNT, count=self.read_count) + self.verifyCount(self.MAX_READ_COUNT, count=self.write_count) result = struct.pack( ">HHHHB", self.read_address, @@ -149,6 +153,8 @@ def decode(self, data: bytes) -> None: self.write_count, self.write_byte_count, ) = struct.unpack(">HHHHB", data[:9]) + self.verifyCount(self.MAX_READ_COUNT, count=self.read_count) + self.verifyCount(self.MAX_READ_COUNT, count=self.write_count) self._payload_byte_count = len(data) - 9 self.write_registers = [ struct.unpack(">H", data[i : i + 2])[0]