This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 9a097c57cc [python][daft] Validate BLOB descriptor parsing (#8462)
9a097c57cc is described below
commit 9a097c57cc089de47b229d1889c424c09f535406
Author: QuakeWang <[email protected]>
AuthorDate: Sun Jul 5 15:19:54 2026 +0800
[python][daft] Validate BLOB descriptor parsing (#8462)
Daft BLOB File conversion parsed serialized BlobDescriptor bytes
manually. That bypassed the shared descriptor validation, so malformed
bytes could leak low-level parsing errors, and descriptors with
unsupported versions or invalid magic could be accepted.
This PR reuses `BlobDescriptor.deserialize` in the Daft conversion path
and adds regression coverage for malformed descriptors while preserving
normal descriptor-to-File conversion behavior.
---
paimon-python/pypaimon/daft/daft_blob.py | 24 +++---------
.../pypaimon/tests/daft/daft_blob_test.py | 43 ++++++++++++++++++++++
2 files changed, 49 insertions(+), 18 deletions(-)
diff --git a/paimon-python/pypaimon/daft/daft_blob.py
b/paimon-python/pypaimon/daft/daft_blob.py
index d249352e03..79931ea6fb 100644
--- a/paimon-python/pypaimon/daft/daft_blob.py
+++ b/paimon-python/pypaimon/daft/daft_blob.py
@@ -25,6 +25,7 @@ import struct
import pyarrow as pa
from pypaimon.daft.daft_compat import file_range_position_field,
file_range_size_field
+from pypaimon.table.row.blob import BlobDescriptor
FILE_PHYSICAL_TYPE = pa.struct(
@@ -39,24 +40,11 @@ FILE_PHYSICAL_TYPE = pa.struct(
def _deserialize_one(data: bytes) -> tuple[str, int, int]:
"""Deserialize a single BlobDescriptor -> (url, offset, length)."""
- pos = 0
- version = data[pos]
- pos += 1
-
- if version > 1:
- pos += 8 # skip magic
-
- uri_len = struct.unpack_from("<I", data, pos)[0]
- pos += 4
-
- uri = data[pos:pos + uri_len].decode("utf-8")
- pos += uri_len
-
- offset = struct.unpack_from("<q", data, pos)[0]
- pos += 8
-
- length = struct.unpack_from("<q", data, pos)[0]
- return uri, offset, length
+ try:
+ descriptor = BlobDescriptor.deserialize(data)
+ except (struct.error, UnicodeDecodeError) as e:
+ raise ValueError("Invalid BlobDescriptor data") from e
+ return descriptor.uri, descriptor.offset, descriptor.length
def blob_column_to_file_array(column: pa.Array, io_config_bytes: bytes | None
= None) -> pa.Array:
diff --git a/paimon-python/pypaimon/tests/daft/daft_blob_test.py
b/paimon-python/pypaimon/tests/daft/daft_blob_test.py
index 594278ca45..4b50c73600 100644
--- a/paimon-python/pypaimon/tests/daft/daft_blob_test.py
+++ b/paimon-python/pypaimon/tests/daft/daft_blob_test.py
@@ -15,6 +15,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
################################################################################
+import struct
import unittest
import pyarrow as pa
@@ -46,6 +47,23 @@ def _descriptor_column(specs):
return pa.array(out, type=pa.large_binary())
+def _descriptor_bytes(version=BlobDescriptor.CURRENT_VERSION,
+ magic=BlobDescriptor.MAGIC,
+ uri=b"oss://b/k",
+ offset=0,
+ length=1):
+ data = bytes([version])
+ if version > 1:
+ data += struct.pack("<Q", magic)
+ data += struct.pack("<I", len(uri))
+ data += uri
+ if offset is not None:
+ data += struct.pack("<q", offset)
+ if length is not None:
+ data += struct.pack("<q", length)
+ return data
+
+
class BlobColumnToFileArrayTest(unittest.TestCase):
# Pure config/arrow tests run on any installed Daft; only File-cast needs
file ranges.
@@ -62,6 +80,31 @@ class BlobColumnToFileArrayTest(unittest.TestCase):
self.assertEqual(blob_column_to_file_array(col,
blob).field("io_config").to_pylist(),
[blob, None, blob])
+ def test_malformed_blob_descriptor_raises_value_error(self):
+ cases = [
+ ("empty", b"", "too short"),
+ ("unknown version",
+ _descriptor_bytes(version=BlobDescriptor.CURRENT_VERSION + 1),
+ "version"),
+ ("bad magic", _descriptor_bytes(magic=0), "magic header"),
+ ("truncated uri",
+ bytes([BlobDescriptor.CURRENT_VERSION])
+ + struct.pack("<Q", BlobDescriptor.MAGIC)
+ + struct.pack("<I", 5)
+ + b"ab",
+ "URI length exceeds data size"),
+ ("truncated offset", _descriptor_bytes(uri=b"", offset=None,
length=None) + b"\x00" * 7,
+ "missing offset/length"),
+ ("truncated length", _descriptor_bytes(uri=b"", length=None) +
b"\x00" * 7,
+ "missing offset/length"),
+ ("invalid utf8 uri", _descriptor_bytes(uri=b"\xff"), "Invalid
BlobDescriptor data"),
+ ]
+
+ for name, raw, message in cases:
+ with self.subTest(name=name):
+ with self.assertRaisesRegex(ValueError, message):
+ blob_column_to_file_array(pa.array([raw],
type=pa.large_binary()))
+
def test_serialize_io_config_roundtrips(self):
s3 = IOConfig(s3=S3Config(key_id="AK", access_key="SK",
session_token="TOK"))
self.assertEqual(IOConfig._from_serialized(serialize_io_config(s3)).s3.session_token,
"TOK")
