feat(encryption): [10/N] Read encrypted manifests

pyiceberg/manifest.py

@@ -900,7 +900,7 @@ def fetch_manifest_entry(
         entry_filter: Callable[[ManifestEntry], bool] | None = None,
     ) -> list[ManifestEntry]:
         """
-        Read the manifest entries from the manifest file.
+        Read the manifest entries from the manifest file, decrypting it when it has key metadata.
 
         Args:
             io: The FileIO to fetch the file.
@@ -911,6 +911,8 @@ def fetch_manifest_entry(
             An Iterator of manifest entries.
         """
         input_file = io.new_input(self.manifest_path)
+        if self.key_metadata is not None:
+            input_file = EncryptedInputFile(input_file, StandardKeyMetadata.from_bytes(self.key_metadata))
         with AvroFile[ManifestEntry](
             input_file,
             MANIFEST_ENTRY_SCHEMAS[DEFAULT_READ_VERSION],

tests/table/encryption/README.md

@@ -17,9 +17,13 @@
   ~ under the License.
 -->
 
-# Encrypted manifest list test data
+# Encrypted metadata 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`.
+
+`manifest-v3-encrypted.avro` is a v3 data manifest with two added entries, written and encrypted by iceberg-rust's
+`ManifestWriter` with key `0123456789abcdef` and AAD prefix `fedcba9876543210`. The key metadata that its manifest list
+entry would record is in `test_read_manifest_encrypted_by_iceberg_rust` in `tests/utils/test_manifest.py`.

tests/utils/test_manifest.py

@@ -27,6 +27,7 @@
 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.key_metadata import StandardKeyMetadata
 from pyiceberg.encryption.manager import EncryptionManager
 from pyiceberg.encryption.stream import _encrypt_ags1
 from pyiceberg.io import load_file_io
@@ -548,6 +549,59 @@ def test_read_encrypted_manifest_list_without_encryption_manager(generated_manif
         snapshot.manifests(load_file_io())
 
 
+def test_read_encrypted_manifest(generated_manifest_entry_file: str, tmp_path: Path) -> None:
+    io = load_file_io()
+    key_metadata = StandardKeyMetadata.generate()
+    encrypted = _encrypt_ags1(Path(generated_manifest_entry_file).read_bytes(), key_metadata)
+    encrypted_path = tmp_path / "manifest.avro"
+    encrypted_path.write_bytes(encrypted)
+    key_metadata = replace(key_metadata, file_length=len(encrypted))
+
+    def manifest(path: str, key_metadata: bytes | None) -> ManifestFile:
+        return ManifestFile.from_args(
+            manifest_path=path,
+            manifest_length=0,
+            partition_spec_id=0,
+            added_snapshot_id=0,
+            sequence_number=0,
+            partitions=[],
+            key_metadata=key_metadata,
+        )
+
+    expected = manifest(generated_manifest_entry_file, None).fetch_manifest_entry(io, discard_deleted=False)
+    assert manifest(str(encrypted_path), key_metadata.to_bytes()).fetch_manifest_entry(io, discard_deleted=False) == expected
+
+    wrong_key = replace(key_metadata, encryption_key=bytes(16))
+    with pytest.raises(ValueError, match="GCM tag check failed"):
+        manifest(str(encrypted_path), wrong_key.to_bytes()).fetch_manifest_entry(io)
+
+    with pytest.raises(ValueError, match="No schema found in Avro file headers"):
+        manifest(str(encrypted_path), None).fetch_manifest_entry(io)
+
+
+def test_read_manifest_encrypted_by_iceberg_rust() -> None:
+    """The manifest was encrypted by iceberg-rust, see tests/table/encryption/README.md."""
+    manifest = ManifestFile.from_args(
+        manifest_path=str(Path(__file__).parent.parent / "table" / "encryption" / "manifest-v3-encrypted.avro"),
+        manifest_length=3444,
+        partition_spec_id=0,
+        content=ManifestContent.DATA,
+        sequence_number=1,
+        min_sequence_number=1,
+        added_snapshot_id=1,
+        partitions=[],
+        key_metadata=bytes.fromhex("01203031323334353637383961626364656602206665646362613938373635343332313002e835"),
+    )
+
+    entries = manifest.fetch_manifest_entry(load_file_io())
+
+    assert [(entry.data_file.file_path, entry.data_file.record_count) for entry in entries] == [
+        ("s3://bucket/table/data/00000.parquet", 3),
+        ("s3://bucket/table/data/00001.parquet", 5),
+    ]
+    assert all(entry.status == ManifestEntryStatus.ADDED and entry.snapshot_id == 1 for entry in entries)
+
+
 def test_read_manifest_cache(generated_manifest_file_file_v2: str) -> None:
     """Test that ManifestFile objects are cached and reused across multiple reads.