From 1738da4d239ebd94260befbac62fa05355517ba2 Mon Sep 17 00:00:00 2001 From: Gijs Molenaar Date: Wed, 7 Oct 2026 11:43:36 +0200 Subject: [PATCH] fix: require the full 28-byte block-count table in list-blocks responses --- CHANGES.md | 3 +++ snap7/s7protocol.py | 16 ++++++------- tests/test_s7protocol.py | 49 +++++++++++++++++++++++++--------------- 3 files changed, 42 insertions(+), 26 deletions(-) diff --git a/CHANGES.md b/CHANGES.md index 3eed6453..082a6dfd 100644 --- a/CHANGES.md +++ b/CHANGES.md @@ -27,6 +27,9 @@ Unreleased * `parse_response` now fails on any nonzero S7 header error word, as native Snap7 does, instead of only when the error class is nonzero. Read and upload responses with error class 0 and a nonzero error code were previously treated as successes while the write path rejected them (#930). +* `parse_list_blocks_response` now requires exactly the seven four-byte block-type entries (28 bytes) that native Snap7 + expects, with distinct, known type codes. Empty or partial tables and duplicate entries are rejected instead of being + reported as zero block counts (#939). 3.2.1 ----- diff --git a/snap7/s7protocol.py b/snap7/s7protocol.py index 1407623b..d75fbee8 100644 --- a/snap7/s7protocol.py +++ b/snap7/s7protocol.py @@ -913,10 +913,9 @@ def parse_list_blocks_response(self, response: Dict[str, Any]) -> Dict[str, int] data_info = response.get("data", {}) raw_data = data_info.get("data", b"") - if not raw_data: - return result - if len(raw_data) % 4 != 0: - raise S7ProtocolError("Block-count response length is not a multiple of four") + # Native Snap7 requires exactly seven four-byte entries (28 bytes) + if len(raw_data) != 28: + raise S7ProtocolError("Block-count response must contain exactly seven entries (28 bytes)") # Parse block entries (4 bytes each: 0x30 | type | count_hi | count_lo) # Block type codes @@ -930,18 +929,19 @@ def parse_list_blocks_response(self, response: Dict[str, Any]) -> Dict[str, int] 0x46: "SFBCount", # System Function Block } - offset = 0 - while offset + 4 <= len(raw_data): + seen: set[int] = set() + for offset in range(0, 28, 4): indicator = raw_data[offset] block_type = raw_data[offset + 1] count = struct.unpack(">H", raw_data[offset + 2 : offset + 4])[0] if indicator != 0x30 or block_type not in type_to_name: raise S7ProtocolError("Invalid block-count response entry") + if block_type in seen: + raise S7ProtocolError("Duplicate block type in block-count response") + seen.add(block_type) result[type_to_name[block_type]] = count - offset += 4 - return result def parse_list_blocks_of_type_response(self, response: Dict[str, Any]) -> List[int]: diff --git a/tests/test_s7protocol.py b/tests/test_s7protocol.py index 0ae4161b..7ac519b6 100644 --- a/tests/test_s7protocol.py +++ b/tests/test_s7protocol.py @@ -378,37 +378,50 @@ class TestParseListBlocksResponse: def setup_method(self) -> None: self.proto = S7Protocol() + @staticmethod + def _table(entries: list[tuple[int, int]]) -> dict[str, Any]: + # Each entry: indicator(0x30) + type + count(2 bytes) + return {"data": {"data": b"".join(struct.pack(">BBH", 0x30, t, c) for t, c in entries)}} + + _ALL_TYPES = (0x38, 0x41, 0x42, 0x43, 0x44, 0x45, 0x46) + def test_valid_response(self) -> None: - # Build entries: indicator(0x30) + type + count(2 bytes) - data = b"" - data += struct.pack(">BBH", 0x30, 0x38, 5) # OB: 5 - data += struct.pack(">BBH", 0x30, 0x41, 3) # DB: 3 - data += struct.pack(">BBH", 0x30, 0x43, 7) # FC: 7 - response = {"data": {"data": data}} + counts = {0x38: 5, 0x41: 3, 0x43: 7} + response = self._table([(t, counts.get(t, 0)) for t in self._ALL_TYPES]) result = self.proto.parse_list_blocks_response(response) assert result["OBCount"] == 5 assert result["DBCount"] == 3 assert result["FCCount"] == 7 assert result["FBCount"] == 0 - def test_empty_data(self) -> None: - response = {"data": {"data": b""}} - result = self.proto.parse_list_blocks_response(response) - assert result["DBCount"] == 0 + def test_empty_data_rejected(self) -> None: + with pytest.raises(S7ProtocolError, match="28 bytes"): + self.proto.parse_list_blocks_response({"data": {"data": b""}}) - def test_no_data(self) -> None: - response: dict[str, Any] = {} - result = self.proto.parse_list_blocks_response(response) - assert all(v == 0 for v in result.values()) + def test_no_data_rejected(self) -> None: + with pytest.raises(S7ProtocolError, match="28 bytes"): + self.proto.parse_list_blocks_response({}) + + def test_partial_table_rejected(self) -> None: + with pytest.raises(S7ProtocolError, match="28 bytes"): + self.proto.parse_list_blocks_response(self._table([(0x38, 1)])) + + def test_oversized_table_rejected(self) -> None: + with pytest.raises(S7ProtocolError, match="28 bytes"): + self.proto.parse_list_blocks_response(self._table([(t, 0) for t in self._ALL_TYPES] + [(0x41, 1)])) + + def test_duplicate_block_type_rejected(self) -> None: + entries = [(t, 0) for t in self._ALL_TYPES[:-1]] + [(0x38, 9)] + with pytest.raises(S7ProtocolError, match="Duplicate"): + self.proto.parse_list_blocks_response(self._table(entries)) def test_unknown_block_type_rejected(self) -> None: - data = struct.pack(">BBH", 0x30, 0xFF, 99) # unknown type - response = {"data": {"data": data}} + entries = [(t, 0) for t in self._ALL_TYPES[:-1]] + [(0xFF, 99)] with pytest.raises(S7ProtocolError, match="entry"): - self.proto.parse_list_blocks_response(response) + self.proto.parse_list_blocks_response(self._table(entries)) def test_truncated_entry_rejected(self) -> None: - with pytest.raises(S7ProtocolError, match="multiple of four"): + with pytest.raises(S7ProtocolError, match="28 bytes"): self.proto.parse_list_blocks_response({"data": {"data": b"\x30\x41\x00"}})