Validate deletion vector blob framing on read

Check the length prefix bounds, the DV magic bytes, and the CRC-32
checksum when stripping the deletion-vector-v1 framing, so corrupted
or truncated blobs fail loudly instead of silently producing wrong
delete positions. The writer added the CRC in this change; the reader
now verifies it.

Author: moomindani

pyiceberg/table/deletion_vector.py

@@ -145,9 +145,23 @@ def to_blob(self) -> PuffinBlob:
 
 
 def _extract_vector_payload(blob_payload: bytes) -> bytes:
-    """Strip deletion-vector-v1 blob framing: length(4 big-endian) + DV magic(4) ... CRC(4 big-endian)."""
+    """Strip and validate deletion-vector-v1 blob framing: length(4 big-endian) + DV magic(4) ... CRC(4 big-endian)."""
+    if len(blob_payload) < 12:
+        raise ValueError(f"Invalid deletion vector blob: too short ({len(blob_payload)} bytes)")
+
     length_prefix = int.from_bytes(blob_payload[0:4], "big")
-    return blob_payload[8 : 4 + length_prefix]
+    if length_prefix < len(DELETION_VECTOR_MAGIC) or 4 + length_prefix + 4 > len(blob_payload):
+        raise ValueError(f"Invalid deletion vector blob: inconsistent length prefix {length_prefix}")
+
+    blob_content = blob_payload[4 : 4 + length_prefix]
+    if blob_content[: len(DELETION_VECTOR_MAGIC)] != DELETION_VECTOR_MAGIC:
+        raise ValueError("Invalid deletion vector blob: magic bytes mismatch")
+
+    expected_crc = int.from_bytes(blob_payload[4 + length_prefix : 8 + length_prefix], "big")
+    if zlib.crc32(blob_content) != expected_crc:
+        raise ValueError("Invalid deletion vector blob: CRC-32 mismatch")
+
+    return blob_content[len(DELETION_VECTOR_MAGIC) :]
 
 
 def deletion_vectors_from_puffin_file(puffin_file: PuffinFile) -> list[DeletionVector]:

tests/table/test_deletion_vector.py

@@ -146,3 +146,40 @@ def test_to_blob_payload_layout() -> None:
     assert length_prefix == len(DELETION_VECTOR_MAGIC) + len(vector)
     assert len(blob.payload) == 4 + length_prefix + 4
     assert DeletionVector._deserialize_bitmap(vector) == DeletionVector.from_positions("file.parquet", [1, 2, 3])._bitmaps
+
+
+def test_extract_vector_payload_rejects_corrupted_vector() -> None:
+    from pyiceberg.table.deletion_vector import _extract_vector_payload
+
+    payload = bytearray(DeletionVector.from_positions("file.parquet", [1, 2, 3]).to_blob().payload)
+    payload[10] ^= 0xFF  # flip a byte inside the serialized vector
+
+    with pytest.raises(ValueError, match="CRC-32 mismatch"):
+        _extract_vector_payload(bytes(payload))
+
+
+def test_extract_vector_payload_rejects_bad_length_prefix() -> None:
+    from pyiceberg.table.deletion_vector import _extract_vector_payload
+
+    payload = bytearray(DeletionVector.from_positions("file.parquet", [1]).to_blob().payload)
+    payload[0:4] = (len(payload) + 100).to_bytes(4, "big")
+
+    with pytest.raises(ValueError, match="inconsistent length prefix"):
+        _extract_vector_payload(bytes(payload))
+
+
+def test_extract_vector_payload_rejects_wrong_magic() -> None:
+    from pyiceberg.table.deletion_vector import _extract_vector_payload
+
+    payload = bytearray(DeletionVector.from_positions("file.parquet", [1]).to_blob().payload)
+    payload[4:8] = b"\x00\x00\x00\x00"
+
+    with pytest.raises(ValueError, match="magic bytes mismatch"):
+        _extract_vector_payload(bytes(payload))
+
+
+def test_extract_vector_payload_rejects_truncated_blob() -> None:
+    from pyiceberg.table.deletion_vector import _extract_vector_payload
+
+    with pytest.raises(ValueError, match="too short"):
+        _extract_vector_payload(b"\x00\x00\x00")