diff --git a/CHANGES.md b/CHANGES.md index d5b9d88e..ddd4a9f3 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -22,6 +22,9 @@ Unreleased Previously it used the client's own reference. That is the likely reason a CP 343-1 kept each connection allocated until it was restarted (#972, not yet confirmed on the CP). The passive `Partner` now echoes the active partner's reference in its connection confirm (#979). +* All `snap7.util` setters now raise `ValueError` when the value doesn't fit in the buffer. Most of them silently grew + a short `bytearray`, which changed the length written to the PLC or shifted later fields, and `set_lreal`, + `set_string`, `set_bool` and the other single-byte setters raised `struct.error` or `IndexError` (#975). 3.3.1 ----- diff --git a/snap7/util/setters.py b/snap7/util/setters.py index 0c168db0..471277b8 100644 --- a/snap7/util/setters.py +++ b/snap7/util/setters.py @@ -10,6 +10,16 @@ Buffer = Union[bytearray, memoryview] +def _check_space(buffer: Buffer, byte_index: int, size: int) -> None: + """Raise ValueError if ``size`` bytes don't fit at ``byte_index``. + + A slice assignment past the end would otherwise grow a bytearray, which changes + the length of what gets written back to the PLC or shifts the fields after it. + """ + if byte_index + size > len(buffer): + raise ValueError(f"{size} bytes at index {byte_index} do not fit in a buffer of {len(buffer)} bytes") + + def set_bool(bytearray_: Buffer, byte_index: int, bool_index: int, value: bool) -> Buffer: """Set boolean value on location in bytearray. @@ -25,6 +35,7 @@ def set_bool(bytearray_: Buffer, byte_index: int, bool_index: int, value: bool) >>> buffer bytearray(b"\\x01") """ + _check_space(bytearray_, byte_index, 1) if value not in {0, 1, True, False}: raise TypeError(f"Value value:{value} is not a boolean expression.") @@ -60,6 +71,7 @@ def set_byte(bytearray_: Buffer, byte_index: int, _int: int) -> Buffer: >>> set_byte(buffer, 0, 255) bytearray(b"\\xFF") """ + _check_space(bytearray_, byte_index, 1) _int = int(_int) bytearray_[byte_index : byte_index + 1] = struct.pack("B", _int) return bytearray_ @@ -79,6 +91,7 @@ def set_word(bytearray_: Buffer, byte_index: int, _int: int) -> Buffer: Return: buffer with the written value """ + _check_space(bytearray_, byte_index, 2) _int = int(_int) bytearray_[byte_index : byte_index + 2] = struct.pack(">H", _int) return bytearray_ @@ -103,6 +116,7 @@ def set_int(bytearray_: Buffer, byte_index: int, _int: int) -> Buffer: >>> set_int(data, 0, 255) bytearray(b'\\x00\\xff') """ + _check_space(bytearray_, byte_index, 2) # make sure were dealing with an int _int = int(_int) bytearray_[byte_index : byte_index + 2] = struct.pack(">h", _int) @@ -129,6 +143,7 @@ def set_uint(bytearray_: Buffer, byte_index: int, _int: int) -> Buffer: >>> set_uint(data, 0, 65535) bytearray(b'\\xff\\xff') """ + _check_space(bytearray_, byte_index, 2) # make sure were dealing with an int _int = int(_int) bytearray_[byte_index : byte_index + 2] = struct.pack(">H", _int) @@ -155,6 +170,7 @@ def set_real(bytearray_: Buffer, byte_index: int, real: Union[bool, str, float, >>> set_real(data, 0, 123.321) bytearray(b'B\\xf6\\xa4Z') """ + _check_space(bytearray_, byte_index, 4) bytearray_[byte_index : byte_index + 4] = struct.pack(">f", float(real)) return bytearray_ @@ -180,6 +196,7 @@ def set_fstring(bytearray_: Buffer, byte_index: int, value: str, max_length: int >>> data bytearray(b'hello world \x00\x00\x00\x00\x00') """ + _check_space(bytearray_, byte_index, max_length) encoded = value.encode(encoding) size = len(encoded) if size > max_length: @@ -216,6 +233,7 @@ def set_string(bytearray_: Buffer, byte_index: int, value: str, max_size: int = >>> data bytearray(b'\\xff\\x0bhello world\\x00\\x00\\x00\\x00\\x00\\x00\\x00') """ + _check_space(bytearray_, byte_index, 2 + max_size) if not isinstance(value, str): raise TypeError(f"Value value:{value} is not from Type string") @@ -260,6 +278,7 @@ def set_dword(bytearray_: Buffer, byte_index: int, dword: int) -> Buffer: >>> data bytearray(b'\\xff\\xff\\xff\\xff') """ + _check_space(bytearray_, byte_index, 4) dword = int(dword) bytearray_[byte_index : byte_index + 4] = struct.pack(">I", dword) return bytearray_ @@ -284,6 +303,7 @@ def set_dint(bytearray_: Buffer, byte_index: int, dint: int) -> Buffer: >>> data bytearray(b'\\x7f\\xff\\xff\\xff') """ + _check_space(bytearray_, byte_index, 4) dint = int(dint) bytearray_[byte_index : byte_index + 4] = struct.pack(">i", dint) return bytearray_ @@ -308,6 +328,7 @@ def set_udint(bytearray_: Buffer, byte_index: int, udint: int) -> Buffer: >>> data bytearray(b'\\xff\\xff\\xff\\xff') """ + _check_space(bytearray_, byte_index, 4) udint = int(udint) bytearray_[byte_index : byte_index + 4] = struct.pack(">I", udint) return bytearray_ @@ -334,6 +355,7 @@ def set_time(bytearray_: Buffer, byte_index: int, time_string: str) -> Buffer: >>> data bytearray(b'\x8d\xda\xaf\x00') """ + _check_space(bytearray_, byte_index, 4) sign = 1 if re.fullmatch( r"(-?(2[0-3]|1?\d):(2[0-3]|1?\d|\d):([1-5]?\d):[1-5]?\d.\d{1,3})|" @@ -381,6 +403,7 @@ def set_usint(bytearray_: Buffer, byte_index: int, _int: int) -> Buffer: >>> set_usint(data, 0, 255) bytearray(b'\\xff') """ + _check_space(bytearray_, byte_index, 1) _int = int(_int) bytearray_[byte_index] = struct.pack(">B", _int)[0] return bytearray_ @@ -407,6 +430,7 @@ def set_sint(bytearray_: Buffer, byte_index: int, _int: int) -> Buffer: >>> set_sint(data, 0, 127) bytearray(b'\\x7f') """ + _check_space(bytearray_, byte_index, 1) _int = int(_int) bytearray_[byte_index] = struct.pack(">b", _int)[0] return bytearray_ @@ -436,6 +460,7 @@ def set_lreal(bytearray_: Buffer, byte_index: int, lreal: float) -> Buffer: >>> Client().db_write(db_number=1, start=10, data=data) """ + _check_space(bytearray_, byte_index, 8) lreal = float(lreal) struct.pack_into(">d", bytearray_, byte_index, lreal) return bytearray_ @@ -463,6 +488,7 @@ def set_lword(bytearray_: Buffer, byte_index: int, lword: int) -> Buffer: >>> data bytearray(b'\\x00\\x00\\x00\\x00\\x00\\x00\\xab\\xcd') """ + _check_space(bytearray_, byte_index, 8) lword = int(lword) bytearray_[byte_index : byte_index + 8] = struct.pack(">Q", lword) return bytearray_ @@ -489,6 +515,7 @@ def set_char(bytearray_: Buffer, byte_index: int, chr_: str) -> Buffer: >>> data bytearray('0x43') """ + _check_space(bytearray_, byte_index, 1) if not isinstance(chr_, str): raise TypeError(f"Value value:{chr_} is not from Type string") @@ -519,6 +546,7 @@ def set_date(bytearray_: Buffer, byte_index: int, date_: date) -> Buffer: >>> data bytearray(b'\x30\xd8') """ + _check_space(bytearray_, byte_index, 2) if date_ < date(1990, 1, 1): raise ValueError("date is lower than specification allows.") elif date_ > date(2168, 12, 31): @@ -553,6 +581,7 @@ def set_wchar(bytearray_: Buffer, byte_index: int, chr_: str) -> Buffer: >>> data bytearray(b'\\x00C') """ + _check_space(bytearray_, byte_index, 2) if not isinstance(chr_, str): raise TypeError(f"Value value:{chr_} is not from Type string") if len(chr_) != 1: @@ -602,14 +631,16 @@ def set_wstring(bytearray_: Buffer, byte_index: int, value: str, max_size: int = if size > max_size: raise ValueError(f"size {size} > max_size {max_size}") + encoded = value.encode("utf-16-be") + _check_space(bytearray_, byte_index, 4 + len(encoded)) + # set max string size (2 bytes, big-endian) bytearray_[byte_index : byte_index + 2] = struct.pack(">H", max_size) # set current length (2 bytes, big-endian) bytearray_[byte_index + 2 : byte_index + 4] = struct.pack(">H", size) - # encode and write UTF-16-BE characters - encoded = value.encode("utf-16-be") + # write UTF-16-BE characters bytearray_[byte_index + 4 : byte_index + 4 + len(encoded)] = encoded @@ -633,6 +664,7 @@ def set_tod(bytearray_: Buffer, byte_index: int, tod: timedelta) -> Buffer: >>> data = bytearray(4) >>> set_tod(data, 0, timedelta(hours=12, minutes=30, seconds=15, milliseconds=500)) """ + _check_space(bytearray_, byte_index, 4) if tod.days >= 1 or tod < timedelta(0): raise ValueError("TIME_OF_DAY must be between 00:00:00.000 and 23:59:59.999") ms = (tod.days * 86400 + tod.seconds) * 1000 + tod.microseconds // 1000 @@ -667,6 +699,7 @@ def set_dtl(bytearray_: Buffer, byte_index: int, dt_: datetime) -> Buffer: >>> data = bytearray(12) >>> set_dtl(data, 0, datetime(2024, 3, 27, 14, 30, 0)) """ + _check_space(bytearray_, byte_index, 12) if dt_ > datetime(2554, 12, 31, 23, 59, 59): raise ValueError("date_val is higher than specification allows.") @@ -711,6 +744,7 @@ def set_dt(bytearray_: Buffer, byte_index: int, dt_: datetime) -> Buffer: >>> data = bytearray(8) >>> set_dt(data, 0, datetime(2020, 7, 12, 17, 32, 2, 854000)) """ + _check_space(bytearray_, byte_index, 8) def byte_to_bcd(val: int) -> int: return ((val // 10) << 4) | (val % 10) @@ -759,6 +793,7 @@ def set_lint(bytearray_: Buffer, byte_index: int, value: int) -> Buffer: >>> data = bytearray(8) >>> set_lint(data, 0, 12345) """ + _check_space(bytearray_, byte_index, 8) bytearray_[byte_index : byte_index + 8] = struct.pack(">q", value) return bytearray_ @@ -782,6 +817,7 @@ def set_ulint(bytearray_: Buffer, byte_index: int, value: int) -> Buffer: >>> data = bytearray(8) >>> set_ulint(data, 0, 12345) """ + _check_space(bytearray_, byte_index, 8) bytearray_[byte_index : byte_index + 8] = struct.pack(">Q", value) return bytearray_ @@ -806,6 +842,7 @@ def set_ltime(bytearray_: Buffer, byte_index: int, value: timedelta) -> Buffer: >>> data = bytearray(8) >>> set_ltime(data, 0, timedelta(seconds=1)) """ + _check_space(bytearray_, byte_index, 8) # Use integer arithmetic to avoid float precision loss nanoseconds = (value.days * 86400 + value.seconds) * 1_000_000_000 + value.microseconds * 1000 bytearray_[byte_index : byte_index + 8] = struct.pack(">q", nanoseconds) @@ -832,6 +869,7 @@ def set_ltod(bytearray_: Buffer, byte_index: int, value: timedelta) -> Buffer: >>> data = bytearray(8) >>> set_ltod(data, 0, timedelta(hours=12, minutes=30)) """ + _check_space(bytearray_, byte_index, 8) if value.days >= 1: raise ValueError("LTOD value must be less than 24 hours") nanoseconds = (value.days * 86400 + value.seconds) * 1_000_000_000 + value.microseconds * 1000 @@ -860,6 +898,7 @@ def set_ldt(bytearray_: Buffer, byte_index: int, value: datetime) -> Buffer: >>> data = bytearray(8) >>> set_ldt(data, 0, datetime(2024, 1, 15, 10, 30, 0)) """ + _check_space(bytearray_, byte_index, 8) epoch = datetime(1970, 1, 1) delta = value - epoch nanoseconds = (delta.days * 86400 + delta.seconds) * 1_000_000_000 + delta.microseconds * 1000 diff --git a/tests/test_util.py b/tests/test_util.py index 8f646b0d..6ae5d243 100644 --- a/tests/test_util.py +++ b/tests/test_util.py @@ -9,7 +9,7 @@ from snap7 import DB, Row from snap7.type import Area, WordLen from snap7.util import get_byte, get_time, get_fstring, get_int, get_string, get_wstring -from snap7.util import set_byte, set_time, set_fstring, set_int +from snap7.util import set_byte, set_time, set_fstring, set_int, setters from snap7.util.db import print_row test_spec = """ @@ -1501,6 +1501,80 @@ def test_write_with_row_offset(self) -> None: mock_client.db_write.assert_called_once() +# Setter name, arguments after the byte index, bytes needed at index 0. +_SETTER_SIZES = [ + ("set_bool", (0, True), 1), + ("set_byte", (1,), 1), + ("set_usint", (1,), 1), + ("set_sint", (1,), 1), + ("set_char", ("a",), 1), + ("set_word", (1,), 2), + ("set_int", (1,), 2), + ("set_uint", (1,), 2), + ("set_wchar", ("a",), 2), + ("set_date", (datetime.date(2024, 1, 1),), 2), + ("set_real", (1.0,), 4), + ("set_dword", (1,), 4), + ("set_dint", (1,), 4), + ("set_udint", (1,), 4), + ("set_time", ("0:0:0:1.0",), 4), + ("set_tod", (datetime.timedelta(seconds=1),), 4), + ("set_fstring", ("a", 4), 4), + ("set_string", ("ab", 4), 6), + ("set_wstring", ("ab", 4), 8), + ("set_lreal", (1.0,), 8), + ("set_lword", (1,), 8), + ("set_lint", (1,), 8), + ("set_ulint", (1,), 8), + ("set_ltime", (datetime.timedelta(seconds=1),), 8), + ("set_ltod", (datetime.timedelta(seconds=1),), 8), + ("set_ldt", (datetime.datetime(2024, 1, 1),), 8), + ("set_dt", (datetime.datetime(2024, 1, 1),), 8), + ("set_dtl", (datetime.datetime(2024, 1, 1),), 12), +] + + +@pytest.mark.util +@pytest.mark.parametrize("name, args, size", _SETTER_SIZES) +def test_setter_refuses_a_short_buffer(name: str, args: tuple[object, ...], size: int) -> None: + setter = getattr(setters, name) + short = bytearray(size - 1) + with pytest.raises(ValueError, match="do not fit"): + setter(short, 0, *args) + assert len(short) == size - 1 + + +@pytest.mark.util +@pytest.mark.parametrize("name, args, size", _SETTER_SIZES) +def test_setter_refuses_a_buffer_short_after_the_index(name: str, args: tuple[object, ...], size: int) -> None: + setter = getattr(setters, name) + with pytest.raises(ValueError, match="do not fit"): + setter(bytearray(size), 1, *args) + + +@pytest.mark.util +@pytest.mark.parametrize("name, args, size", _SETTER_SIZES) +def test_setter_accepts_an_exact_buffer(name: str, args: tuple[object, ...], size: int) -> None: + buffer = bytearray(size) + getattr(setters, name)(buffer, 0, *args) + assert len(buffer) == size + + +@pytest.mark.util +def test_set_wstring_needs_room_for_the_value_only() -> None: + buffer = bytearray(6) + setters.set_wstring(buffer, 0, "a") + assert get_wstring(buffer, 0) == "a" + + +@pytest.mark.util +def test_row_set_value_refuses_a_short_buffer() -> None: + row = Row(bytearray(1), "0 value INT") + with pytest.raises(ValueError, match="do not fit"): + row["value"] = 1 + assert len(row.get_bytearray()) == 1 + + if __name__ == "__main__": unittest.main()