fix: require the full 28-byte block-count table in list-blocks responses

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
 -----

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]:

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"}})