From 0ad7405099ec4b5a33d39d0e05c40a8fe46a4644 Mon Sep 17 00:00:00 2001 From: Xander Date: Mon, 26 Oct 2026 09:47:29 -0400 Subject: [PATCH] Read encrypted manifest lists --- pyiceberg/encryption/io.py | 50 ++++++++++++ pyiceberg/manifest.py | 7 +- pyiceberg/table/__init__.py | 37 +++++++-- pyiceberg/table/inspect.py | 12 +-- pyiceberg/table/snapshots.py | 19 ++++- tests/conftest.py | 73 ++++++++++++++++++ tests/encryption/test_stream.py | 31 ++++++++ tests/table/encryption/README.md | 25 ++++++ .../manifest-list-v3-encrypted.avro | Bin 0 -> 1568 bytes tests/table/test_init.py | 33 ++++++++ tests/utils/test_manifest.py | 42 ++++++++++ 11 files changed, 315 insertions(+), 14 deletions(-) create mode 100644 pyiceberg/encryption/io.py create mode 100644 tests/table/encryption/README.md create mode 100644 tests/table/encryption/manifest-list-v3-encrypted.avro diff --git a/pyiceberg/encryption/io.py b/pyiceberg/encryption/io.py new file mode 100644 index 0000000000..1ba9dd5b7d --- /dev/null +++ b/pyiceberg/encryption/io.py @@ -0,0 +1,50 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, +# software distributed under the License is distributed on an +# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +# KIND, either express or implied. See the License for the +# specific language governing permissions and limitations +# under the License. +"""Input files for metadata encrypted with AGS1 streams.""" + +from __future__ import annotations + +import io + +from pyiceberg.encryption.key_metadata import StandardKeyMetadata +from pyiceberg.encryption.stream import _decrypt_ags1 +from pyiceberg.io import InputFile, InputStream + + +class EncryptedInputFile(InputFile): + """Decrypts an AGS1 encrypted file, such as a manifest list, when it is opened.""" + + def __init__(self, input_file: InputFile, key_metadata: StandardKeyMetadata) -> None: + super().__init__(input_file.location) + self._input_file = input_file + self._key_metadata = key_metadata + + def __len__(self) -> int: + """Return the length of the decrypted file, in bytes.""" + return len(self._decrypt()) + + def exists(self) -> bool: + """Check whether the encrypted file exists.""" + return self._input_file.exists() + + def open(self, seekable: bool = True) -> InputStream: + """Read and decrypt the whole file, authenticating every block before returning any of it.""" + return io.BytesIO(self._decrypt()) + + def _decrypt(self) -> bytes: + with self._input_file.open(seekable=False) as stream: + return _decrypt_ags1(stream.read(), self._key_metadata) diff --git a/pyiceberg/manifest.py b/pyiceberg/manifest.py index 88ca051015..89f7a1d7cd 100644 --- a/pyiceberg/manifest.py +++ b/pyiceberg/manifest.py @@ -34,6 +34,8 @@ from pyiceberg.avro.codecs import AVRO_CODEC_KEY, AvroCompressionCodec from pyiceberg.avro.file import AvroFile, AvroOutputFile from pyiceberg.conversions import to_bytes +from pyiceberg.encryption.io import EncryptedInputFile +from pyiceberg.encryption.key_metadata import StandardKeyMetadata from pyiceberg.exceptions import ValidationError from pyiceberg.io import FileIO, InputFile, OutputFile from pyiceberg.partitioning import PartitionSpec @@ -998,7 +1000,7 @@ def clear_manifest_cache() -> None: _manifest_cache.clear() -def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]: +def _manifests(io: FileIO, manifest_list: str, key_metadata: StandardKeyMetadata | None = None) -> tuple[ManifestFile, ...]: """Read manifests from a manifest list, reusing cached ManifestFile objects. Caches individual ManifestFile objects by manifest_path. This is memory-efficient @@ -1018,11 +1020,14 @@ def _manifests(io: FileIO, manifest_list: str) -> tuple[ManifestFile, ...]: Args: io: FileIO instance for reading the manifest list. manifest_list: Path to the manifest list file. + key_metadata: Decrypts the manifest list, if it is encrypted. Returns: A tuple of ManifestFile objects. """ file = io.new_input(manifest_list) + if key_metadata is not None: + file = EncryptedInputFile(file, key_metadata) manifest_files = list(read_manifest_list(file)) return tuple(_manifest_cache.get_or_cache(manifest_file) for manifest_file in manifest_files) diff --git a/pyiceberg/table/__init__.py b/pyiceberg/table/__init__.py index 7da0e59d26..54040ff77a 100644 --- a/pyiceberg/table/__init__.py +++ b/pyiceberg/table/__init__.py @@ -32,6 +32,7 @@ from pydantic import Field +from pyiceberg.encryption.kms import KeyManagementClient from pyiceberg.exceptions import CommitFailedException, ValidationException from pyiceberg.expressions import AlwaysFalse, AlwaysTrue, And, BooleanExpression, EqualTo, IsNull, Or, Reference from pyiceberg.expressions.visitors import ( @@ -124,6 +125,7 @@ from pyiceberg.catalog import Catalog from pyiceberg.catalog.rest.scan_planning import RESTContentFile, RESTDeleteFile, RESTFileScanTask + from pyiceberg.encryption.manager import EncryptionManager from pyiceberg.table.update.snapshot import _SnapshotProducer logger = logging.getLogger(__name__) @@ -1340,6 +1342,7 @@ def __init__( io: FileIO, catalog: Catalog, config: dict[str, str] = EMPTY_DICT, + kms_client: KeyManagementClient | None = None, ) -> None: self._identifier = identifier self.metadata = metadata @@ -1347,6 +1350,18 @@ def __init__( self.io = io self.catalog = catalog self.config = config + self._kms_client = kms_client + + @property + def encryption_manager(self) -> EncryptionManager | None: + """The table's encryption manager, or None if the table is not encrypted. + + Raises: + ValueError: If the table is encrypted but no key management client was provided. + """ + from pyiceberg.encryption.manager import EncryptionManager + + return EncryptionManager.from_table_metadata(self._kms_client, self.metadata) def transaction(self) -> Transaction: """Create a new transaction object to first stage the changes, and then commit them to the catalog. @@ -1446,6 +1461,7 @@ def scan( catalog=self.catalog, table_identifier=self._identifier, table_config=self.config, + encryption_manager=self.encryption_manager, ) def incremental_append_scan( @@ -1501,6 +1517,7 @@ def incremental_append_scan( to_snapshot_id=to_snapshot_id_inclusive, options=options, limit=limit, + encryption_manager=self.encryption_manager, ) @property @@ -2068,6 +2085,7 @@ class BaseScan(ABC): case_sensitive: bool options: Properties limit: int | None + encryption_manager: EncryptionManager | None def __init__( self, @@ -2078,6 +2096,7 @@ def __init__( case_sensitive: bool = True, options: Properties = EMPTY_DICT, limit: int | None = None, + encryption_manager: EncryptionManager | None = None, ): self.table_metadata = table_metadata self.io = io @@ -2086,6 +2105,7 @@ def __init__( self.case_sensitive = case_sensitive self.options = options self.limit = limit + self.encryption_manager = encryption_manager @abstractmethod def plan_files(self) -> Iterable[ScanTask]: ... @@ -2185,6 +2205,7 @@ def __init__( catalog: Catalog | None = None, table_identifier: Identifier | None = None, table_config: Properties = EMPTY_DICT, + encryption_manager: EncryptionManager | None = None, ): super().__init__( table_metadata=table_metadata, @@ -2194,6 +2215,7 @@ def __init__( case_sensitive=case_sensitive, options=options, limit=limit, + encryption_manager=encryption_manager, ) self.snapshot_id = snapshot_id self.catalog = catalog @@ -2449,7 +2471,7 @@ def _plan_manifest_entries(self) -> Iterator[list[ManifestEntry]]: if not snapshot: return iter([]) - return self._manifest_planner.plan_manifest_entries(snapshot.manifests(self.io)) + return self._manifest_planner.plan_manifest_entries(snapshot.manifests(self.io, self.encryption_manager)) def _should_use_server_side_planning(self) -> bool: """Check if server-side scan planning should be used for this scan.""" @@ -2486,7 +2508,7 @@ def _plan_files_local(self) -> Iterable[FileScanTask]: snapshot = self.snapshot() if not snapshot: return [] - return self._manifest_planner.plan_files(snapshot.manifests(self.io)) + return self._manifest_planner.plan_files(snapshot.manifests(self.io, self.encryption_manager)) def plan_files(self) -> Iterable[FileScanTask]: """Plans the relevant files by filtering on the PartitionSpecs. @@ -2602,6 +2624,8 @@ class IncrementalAppendScan(BaseScan): An integer representing the number of rows to return in the scan result. If None, fetches all matching rows. + encryption_manager: + Decrypts the manifest lists of encrypted snapshots. """ from_snapshot_id: int | None @@ -2620,6 +2644,7 @@ def __init__( to_snapshot_id: int | None = None, options: Properties = EMPTY_DICT, limit: int | None = None, + encryption_manager: EncryptionManager | None = None, ): super().__init__( table_metadata=table_metadata, @@ -2629,6 +2654,7 @@ def __init__( case_sensitive=case_sensitive, options=options, limit=limit, + encryption_manager=encryption_manager, ) self.from_snapshot_id = from_snapshot_id self.from_snapshot_inclusive = from_snapshot_inclusive @@ -2678,7 +2704,7 @@ def plan_files(self) -> Iterable[FileScanTask]: { manifest_file for snapshot in append_snapshots - for manifest_file in snapshot.manifests(self.io) + for manifest_file in snapshot.manifests(self.io, self.encryption_manager) if manifest_file.content == ManifestContent.DATA and manifest_file.added_snapshot_id in append_snapshot_ids } ) @@ -2691,8 +2717,9 @@ def plan_files(self) -> Iterable[FileScanTask]: options=self.options, ).plan_files( manifests=manifests, - manifest_entry_filter=lambda manifest_entry: manifest_entry.snapshot_id in append_snapshot_ids - and manifest_entry.status == ManifestEntryStatus.ADDED, + manifest_entry_filter=lambda manifest_entry: ( + manifest_entry.snapshot_id in append_snapshot_ids and manifest_entry.status == ManifestEntryStatus.ADDED + ), ) def to_arrow(self) -> pa.Table: diff --git a/pyiceberg/table/inspect.py b/pyiceberg/table/inspect.py index e24e251fa4..67325058e0 100644 --- a/pyiceberg/table/inspect.py +++ b/pyiceberg/table/inspect.py @@ -232,7 +232,7 @@ def _readable_metrics_struct(bound_type: PrimitiveType) -> pa.StructType: entries = [] snapshot = self._get_snapshot(snapshot_id) - for manifest in snapshot.manifests(self.tbl.io): + for manifest in snapshot.manifests(self.tbl.io, self.tbl.encryption_manager): for entry in manifest.fetch_manifest_entry(io=self.tbl.io, discard_deleted=False): column_sizes = entry.data_file.column_sizes or {} value_counts = entry.data_file.value_counts or {} @@ -383,7 +383,7 @@ def partitions( ) snapshot = self._get_snapshot(snapshot_id) - spec_ids = {manifest.partition_spec_id for manifest in snapshot.manifests(self.tbl.io)} + spec_ids = {manifest.partition_spec_id for manifest in snapshot.manifests(self.tbl.io, self.tbl.encryption_manager)} partition_record = self.tbl.metadata.specs_struct(spec_ids=spec_ids) has_partitions = len(partition_record.fields) > 0 @@ -590,7 +590,7 @@ def _partition_summaries_to_rows( specs = self.tbl.metadata.specs() manifests = [] if snapshot: - for manifest in snapshot.manifests(self.tbl.io): + for manifest in snapshot.manifests(self.tbl.io, self.tbl.encryption_manager): is_data_file = manifest.content == ManifestContent.DATA is_delete_file = manifest.content == ManifestContent.DELETES manifest_row = { @@ -882,7 +882,8 @@ def _files(self, snapshot_id: int | None = None, data_file_filter: set[DataFileC executor = ExecutorFactory.get_or_create() results = list( executor.map( - lambda manifest_list: self._get_files_from_manifest(manifest_list, data_file_filter), snapshot.manifests(io) + lambda manifest_list: self._get_files_from_manifest(manifest_list, data_file_filter), + snapshot.manifests(io, self.tbl.encryption_manager), ) ) return pa.concat_tables(results) @@ -970,7 +971,8 @@ def _all_files(self, data_file_filter: set[DataFileContent] | None = None) -> pa return pa.Table.from_pylist([], schema=self._get_files_schema()) executor = ExecutorFactory.get_or_create() - manifest_lists = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io), snapshots) + encryption_manager = self.tbl.encryption_manager + manifest_lists = executor.map(lambda snapshot: snapshot.manifests(self.tbl.io, encryption_manager), snapshots) unique_manifests = {(manifest.manifest_path, manifest) for manifest_list in manifest_lists for manifest in manifest_list} diff --git a/pyiceberg/table/snapshots.py b/pyiceberg/table/snapshots.py index 0450df2861..deeee8230f 100644 --- a/pyiceberg/table/snapshots.py +++ b/pyiceberg/table/snapshots.py @@ -32,6 +32,7 @@ from pyiceberg.schema import Schema if TYPE_CHECKING: + from pyiceberg.encryption.manager import EncryptionManager from pyiceberg.table.metadata import TableMetadata from pyiceberg.typedef import IcebergBaseModel @@ -289,9 +290,21 @@ def __repr__(self) -> str: filtered_fields = [field for field in fields if field is not None] return f"Snapshot({', '.join(filtered_fields)})" - def manifests(self, io: FileIO) -> list[ManifestFile]: - """Return the manifests for the given snapshot.""" - return list(_manifests(io, self.manifest_list)) + def manifests(self, io: FileIO, encryption_manager: EncryptionManager | None = None) -> list[ManifestFile]: + """Return the manifests for the given snapshot. + + Args: + io: Reads the manifest list. + encryption_manager: Decrypts the manifest list, required when the snapshot has a `key-id`. + """ + if self.key_id is None: + return list(_manifests(io, self.manifest_list)) + + if encryption_manager is None: + raise ValueError(f"Snapshot {self.snapshot_id} is encrypted, but no encryption manager was provided") + + key_metadata = encryption_manager.decrypt_manifest_list_key_metadata(self.key_id) + return list(_manifests(io, self.manifest_list, key_metadata)) class MetadataLogEntry(IcebergBaseModel): diff --git a/tests/conftest.py b/tests/conftest.py index 316499ba53..244cc2746c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -2652,6 +2652,79 @@ def table_v3(example_table_metadata_v3: dict[str, Any]) -> Table: ) +@pytest.fixture +def example_table_metadata_v3_encrypted() -> dict[str, Any]: + """Metadata of an encrypted table written by iceberg-rust, whose manifest list is in tests/table/encryption.""" + manifest_list = Path(__file__).parent / "table" / "encryption" / "manifest-list-v3-encrypted.avro" + return { + "format-version": 3, + "table-uuid": "9c12d441-03fe-4693-9a96-a0705ddf69c1", + "location": "memory:///table", + "last-sequence-number": 0, + "last-updated-ms": 1602638573590, + "last-column-id": 1, + "current-schema-id": 0, + "schemas": [{"type": "struct", "schema-id": 0, "fields": [{"id": 1, "name": "x", "required": True, "type": "long"}]}], + "default-spec-id": 0, + "partition-specs": [{"spec-id": 0, "fields": []}], + "last-partition-id": 1000, + "default-sort-order-id": 0, + "sort-orders": [{"order-id": 0, "fields": []}], + "properties": {"encryption.key-id": "master-1"}, + "current-snapshot-id": 1, + "snapshots": [ + { + "snapshot-id": 1, + "sequence-number": 0, + "timestamp-ms": 0, + "manifest-list": str(manifest_list), + "summary": {"operation": "append"}, + "schema-id": 0, + "key-id": "1fb95f04-7f98-4b1f-a08f-e810cfb907ff", + } + ], + "encryption-keys": [ + { + "key-id": "5b29941b-b3b4-4d57-8758-a9d1bf340b39", + "encrypted-key-metadata": "WVMIPwafypGJRgXOICFKcL+m8K4B40gEIJFsgC7rlitqN9PUpDeT19OZPs8=", + "encrypted-by-id": "master-1", + "properties": {"KEY_TIMESTAMP": "1781172301703"}, + }, + { + "key-id": "1fb95f04-7f98-4b1f-a08f-e810cfb907ff", + "encrypted-key-metadata": ( + "XcrqLxxyf9a7FAFXM63u3MEH8CSWqh6g1H8fJD2/6zx0EjcPzPaP/5AtRpO2s6p2isV8s7a4m77zj+GIqNmIpbA2Xw==" + ), + "encrypted-by-id": "5b29941b-b3b4-4d57-8758-a9d1bf340b39", + }, + ], + "snapshot-log": [{"snapshot-id": 1, "timestamp-ms": 0}], + "metadata-log": [], + "refs": {"main": {"snapshot-id": 1, "type": "branch"}}, + "next-row-id": 0, + } + + +@pytest.fixture +def table_v3_encrypted(example_table_metadata_v3_encrypted: dict[str, Any]) -> Table: + from pyiceberg.catalog.noop import NoopCatalog + from pyiceberg.encryption.ciphers import SecureKey + from tests.encryption.memory_kms import MemoryKeyManagementClient + + kms = MemoryKeyManagementClient() + kms.add_master_key("master-1", SecureKey(bytes(range(16)))) + + table_metadata = TableMetadataV3(**example_table_metadata_v3_encrypted) + return Table( + identifier=("database", "table"), + metadata=table_metadata, + metadata_location=f"{table_metadata.location}/uuid.metadata.json", + io=load_file_io(), + catalog=NoopCatalog("NoopCatalog"), + kms_client=kms, + ) + + @pytest.fixture def table_v2_orc(example_table_metadata_v2: dict[str, Any]) -> Table: import copy diff --git a/tests/encryption/test_stream.py b/tests/encryption/test_stream.py index 2f96e71559..5211f39d19 100644 --- a/tests/encryption/test_stream.py +++ b/tests/encryption/test_stream.py @@ -15,8 +15,12 @@ # specific language governing permissions and limitations # under the License. +from dataclasses import replace +from pathlib import Path + import pytest +from pyiceberg.encryption.io import EncryptedInputFile from pyiceberg.encryption.key_metadata import StandardKeyMetadata from pyiceberg.encryption.stream import ( _BLOCK_OVERHEAD, @@ -29,6 +33,7 @@ _encrypt_ags1, _stream_block_aad, ) +from pyiceberg.io.pyarrow import PyArrowFileIO KEY = b"0123456789012345" AAD_PREFIX = b"0123456789abcdef" @@ -203,3 +208,29 @@ def test_decrypt_rejects_the_wrong_key() -> None: with pytest.raises(ValueError, match="wrong decryption key; or corrupt/tampered data"): decrypt(stream, key=b"1123456789012345") + + +def test_encrypted_input_file(tmp_path: Path) -> None: + plaintext = b"manifest list contents" + encrypted = _encrypt_ags1(plaintext, KEY_METADATA) + path = tmp_path / "manifest-list.avro" + path.write_bytes(encrypted) + + input_file = EncryptedInputFile(PyArrowFileIO().new_input(str(path)), replace(KEY_METADATA, file_length=len(encrypted))) + + assert input_file.location == str(path) + assert input_file.exists() + assert len(input_file) == len(plaintext) + with input_file.open() as stream: + assert stream.read() == plaintext + + +def test_encrypted_input_file_rejects_a_truncated_file(tmp_path: Path) -> None: + encrypted = _encrypt_ags1(b"manifest list contents", KEY_METADATA) + path = tmp_path / "manifest-list.avro" + path.write_bytes(encrypted[:-1]) + + input_file = EncryptedInputFile(PyArrowFileIO().new_input(str(path)), replace(KEY_METADATA, file_length=len(encrypted))) + + with pytest.raises(ValueError, match="key metadata expects"): + input_file.open() diff --git a/tests/table/encryption/README.md b/tests/table/encryption/README.md new file mode 100644 index 0000000000..ae20c650c8 --- /dev/null +++ b/tests/table/encryption/README.md @@ -0,0 +1,25 @@ + + +# Encrypted manifest list test data + +`manifest-list-v3-encrypted.avro` is an empty manifest list encrypted by iceberg-rust, copied from its +`crates/iceberg/testdata/manifests_lists/`. It was written by iceberg-rust's `EncryptionManager`, with a memory KMS +holding master key `master-1` = bytes `0x00` to `0x0f`. Its keys and snapshot are in the +`example_table_metadata_v3_encrypted` fixture in `tests/conftest.py`. diff --git a/tests/table/encryption/manifest-list-v3-encrypted.avro b/tests/table/encryption/manifest-list-v3-encrypted.avro new file mode 100644 index 0000000000000000000000000000000000000000..4963f0effbf37ee40174313a313a6205be8560c0 GIT binary patch literal 1568 zcmV+*2H*KXM^iBX01yCIDK4x@+rT4V*y)*^g2c0Oij>qvN968=?U@(S)k7oR>2<7U z>3A^GZko-zPgoD!69AT$!NJ$Q+bQC25h7vrUTC*QlXj)Dc&y{?jB7)03uvO`GjZzb zFQ6bq6!6W=y?XR`uN7$*NHOz54iI~Or_;YB=WC~AY=LC}I_j!gf3kqB_wJOAjGYf1 za|IuKPeg_4NW1&anlEMjZER=iuB5!|72;ufJMIi2Ya6qECx0iTcZ%^<$^M?5T7N~D5BST$1#c$m9+Z$Y00SVbS_x?^W zLpAL^-yc;ay28QtOmzV^uN~vJ1SdbWM~i{`0*XVyUAt#o9jfTevqee9ou#}gskAi4 zNK!aC=x+DlJC_TF__s@(Ba3%vrZ5;3+D#g1ml{k);NlX>h9_;GkH`QU3?|9wz@4Qo+h8jt zGs%z*EpVkUycw#1m;6+fRm53<>NNF+(tIr-xAD|^tS#%4pu>88q4i6^Lay2WvlVxd zX6d%&(JfZnA)zb9dG!Wi^6Ui7BMoaYAi0~BZ#YEIW`rZNa$Xr&#n}859MVC)+CLt2 zaW9M8jB*Rs;|EZQA>BQ?Tl6*7^?JhZXye4V-fuq0A}Z^1S9_)Jz^z3=n(upAcH}g$ zM)p7$kZnD0ZF)nTr<$_!N6f*gdO}s-=lsmZhv}0&8c6B-~{SWCFBeAmUSy%-6*~Y0?=?{t24;qZ!%%1*t{I2Cc z7fB@4 zk)6_(`8d7_!f3Q!%$vy}!X2$M%WnzTx-$d45ohkzq1QLlkjBugaPDxjHn0UV{u$Wx zQY?0T(bfY`t_uFw;_S43^9RU+hagb`tA8nAC1f%^ugXmv0u%zhwuw+`&}~tUsR`V8 zJ_rAASEGSLU0RF3HJ6n6A2J~;4se%kwQkBde+vEESa~S)VlDYOwb9EBGom#y++X%>SL=G z&~On)SWOZ!*DMvAbQ(~sXti%4lc=ruezI?{g+(RYYQph7w``qB>kuiV-Kgw2cOuS@ zohB7CaQlfK&pn{mAtw};Gpwz2sM=T~c!R*5aNOD@1mn|eQo&Pqpm|(#Qr^+3db;*S z5GyyqPNt&7^&yqMBDIwR6u8XFy+5?y>x*VCFh&b>KCW(pBq2Yng^ None: assert TypeAdapter(TableUpdate).validate_json(remove.model_dump_json()) == remove +def test_encryption_manager_of_unencrypted_table(table_v3: Table) -> None: + assert table_v3.encryption_manager is None + + +def test_encryption_manager_requires_kms_client(example_table_metadata_v3_encrypted: dict[str, Any]) -> None: + table_metadata = TableMetadataUtil.parse_obj(example_table_metadata_v3_encrypted) + table = Table(("database", "table"), table_metadata, "s3://bucket/metadata.json", load_file_io(), NoopCatalog("noop")) + + with pytest.raises(ValueError, match="no key management client was provided"): + table.scan() + + +def test_encryption_manager_follows_current_metadata(table_v3_encrypted: Table) -> None: + assert table_v3_encrypted.encryption_manager is not None + + # Catalogs must not let encryption.key-id be removed, but removing it shows the manager follows the current metadata + table_v3_encrypted.metadata = table_v3_encrypted.metadata.model_copy(update={"properties": {}}) + + assert table_v3_encrypted.encryption_manager is None + + +def test_scan_encrypted_table(table_v3_encrypted: Table) -> None: + """The manifest list was encrypted by iceberg-rust, and lists no manifests.""" + assert list(table_v3_encrypted.scan().plan_files()) == [] + assert list(table_v3_encrypted.incremental_append_scan().plan_files()) == [] + assert table_v3_encrypted.scan().to_arrow().num_rows == 0 + + +def test_inspect_encrypted_table(table_v3_encrypted: Table) -> None: + assert table_v3_encrypted.inspect.manifests().num_rows == 0 + assert table_v3_encrypted.inspect.all_manifests().num_rows == 0 + + def test_add_snapshot_update_fails_without_first_row_id(table_v3: Table) -> None: new_snapshot = Snapshot( snapshot_id=25, diff --git a/tests/utils/test_manifest.py b/tests/utils/test_manifest.py index 45e4e894ef..accfb07f57 100644 --- a/tests/utils/test_manifest.py +++ b/tests/utils/test_manifest.py @@ -16,6 +16,7 @@ # under the License. # pylint: disable=redefined-outer-name,arguments-renamed,fixme import importlib +from dataclasses import replace from pathlib import Path from tempfile import TemporaryDirectory from typing import Any @@ -26,6 +27,8 @@ import pyiceberg.manifest as manifest_module from pyiceberg.avro.codecs import AvroCompressionCodec from pyiceberg.avro.file import DEFAULT_SYNC_INTERVAL, AvroFile, AvroOutputFile +from pyiceberg.encryption.manager import EncryptionManager +from pyiceberg.encryption.stream import _encrypt_ags1 from pyiceberg.io import load_file_io from pyiceberg.io.pyarrow import PyArrowFileIO from pyiceberg.manifest import ( @@ -52,6 +55,7 @@ from pyiceberg.table.snapshots import Operation, Snapshot, Summary from pyiceberg.typedef import Record, TableVersion from pyiceberg.types import IntegerType, ListType, LongType, NestedField, StructType +from tests.encryption.memory_kms import MemoryKeyManagementClient @pytest.fixture(autouse=True) @@ -506,6 +510,44 @@ def test_read_manifest_v2(generated_manifest_file_file_v2: str) -> None: assert entry.status == ManifestEntryStatus.ADDED +def test_read_encrypted_manifest_list(generated_manifest_file_file_v2: str, tmp_path: Path) -> None: + io = load_file_io() + kms = MemoryKeyManagementClient() + kms.add_master_key("master-1") + encryption_manager = EncryptionManager(kms, "master-1") + + key_metadata = encryption_manager.generate_key_metadata() + encrypted = _encrypt_ags1(Path(generated_manifest_file_file_v2).read_bytes(), key_metadata) + manifest_list = tmp_path / "manifest-list.avro" + manifest_list.write_bytes(encrypted) + key_id = encryption_manager.encrypt_manifest_list_key_metadata(replace(key_metadata, file_length=len(encrypted))) + + plaintext = Snapshot( + snapshot_id=25, + parent_snapshot_id=19, + timestamp_ms=1602638573590, + manifest_list=generated_manifest_file_file_v2, + summary=Summary(Operation.APPEND), + schema_id=3, + ) + snapshot = plaintext.model_copy(update={"manifest_list": str(manifest_list), "key_id": key_id}) + + assert snapshot.manifests(io, encryption_manager) == plaintext.manifests(io) + + +def test_read_encrypted_manifest_list_without_encryption_manager(generated_manifest_file_file_v2: str) -> None: + snapshot = Snapshot( + snapshot_id=25, + timestamp_ms=1602638573590, + manifest_list=generated_manifest_file_file_v2, + summary=Summary(Operation.APPEND), + key_id="key", + ) + + with pytest.raises(ValueError, match="Snapshot 25 is encrypted, but no encryption manager was provided"): + snapshot.manifests(load_file_io()) + + def test_read_manifest_cache(generated_manifest_file_file_v2: str) -> None: """Test that ManifestFile objects are cached and reused across multiple reads.