This is an automated email from the ASF dual-hosted git repository. github-merge-queue[bot] pushed a commit to branch gh-readonly-queue/main/pr-3236-830bb886542527f4e8956f03040e7c407b621401 in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
commit c9a25b23b7645f3af0e9dfd7e85d8cf646901c55 Author: Xander <[email protected]> AuthorDate: Wed Sep 23 21:20:18 2026 +0000 fix(encryption) [18/N] wire in ags1 file length for tamper proofing (#3236) * feat(encryption) [18N] wire in ags1 file length for tamper proofing * remove debug * expect * fix test * fix opendal * make more clear * Add oversized tests * don't do unreachable * import fix * Assert length in streaming tests * DRAFT try key_metadata_with_length * fix(encryption): address key metadata length review --- CHANGELOG.md | 9 + crates/iceberg/public-api.txt | 16 +- .../src/arrow/caching_delete_file_loader.rs | 7 +- crates/iceberg/src/encryption/io.rs | 225 +++++++++++++++++++-- crates/iceberg/src/encryption/key_metadata.rs | 4 +- crates/iceberg/src/encryption/manager.rs | 8 +- crates/iceberg/src/encryption/stream.rs | 158 +++++++++++---- crates/iceberg/src/io/file_io.rs | 4 +- crates/iceberg/src/io/storage/local_fs.rs | 18 +- crates/iceberg/src/io/storage/memory.rs | 10 +- crates/iceberg/src/puffin/reader.rs | 2 +- crates/iceberg/src/puffin/writer.rs | 15 +- crates/iceberg/src/spec/manifest/writer.rs | 107 +++++----- .../src/spec/manifest_list/manifest_file.rs | 17 +- crates/iceberg/src/spec/manifest_list/writer.rs | 14 +- crates/iceberg/src/transaction/append.rs | 31 ++- crates/iceberg/src/transaction/snapshot.rs | 21 +- .../src/writer/file_writer/parquet_writer.rs | 2 + .../TableMetadataV3ValidEncryption.json | 2 +- crates/storage/opendal/src/lib.rs | 76 ++++++- 20 files changed, 568 insertions(+), 178 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 1e06bafa9..83eeeb31e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -24,6 +24,15 @@ All notable changes to this project will be documented in this file. The format is based on [Keep a Changelog](https://keepachangelog.com/) and this project adheres to [Semantic Versioning](https://semver.org/). +## [Unreleased] + +### Breaking Changes + +* `FileWrite::close`, `EncryptedOutputFile::write`, `ManifestListWriter::close`, and `PuffinWriter::close` now return `Result<io::FileMetadata>` instead of `Result<()>`. The returned size is the on-disk size, including encryption overhead. Custom `FileWrite` implementations must return the number of bytes stored. +* `EncryptedOutputFile::key_metadata()` is replaced by `key_metadata_with_saved_file_metadata(&FileMetadata)`. Pass the metadata returned by `write()` or the writer's `close()` to include the stored length before encoding key metadata. +* `EncryptedInputFile::metadata()` is now synchronous and derives the plaintext size from key metadata without a storage stat. Remove `.await` from calls to this method. +* AGS1 readers now require `StandardKeyMetadata::file_length` and reject missing or invalid lengths without falling back to a storage stat. AGS1-encrypted manifests, manifest lists, and Puffin files written by earlier development builds without this field must be rewritten using a build that can still read them before upgrading. This matches the Java client's read contract. + ## [v0.10.1] - 2026-07-28 * chore: Update crossbeam-epoch to 0.9.20, pin CI ruff version to 0.15.22 by @dannycjones in https://github.com/apache/iceberg-rust/pull/2911 diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt index f29c5efe0..e2e78073e 100644 --- a/crates/iceberg/public-api.txt +++ b/crates/iceberg/public-api.txt @@ -273,7 +273,7 @@ pub struct iceberg::encryption::AesGcmFileWrite impl iceberg::encryption::AesGcmFileWrite pub fn iceberg::encryption::AesGcmFileWrite::new(inner: alloc::boxed::Box<dyn iceberg::io::FileWrite>, cipher: alloc::sync::Arc<iceberg::encryption::AesGcmCipher>, aad_prefix: impl core::convert::Into<alloc::boxed::Box<[u8]>>) -> Self impl iceberg::io::FileWrite for iceberg::encryption::AesGcmFileWrite -pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::encryption::AesGcmFileWrite::write<'life0, 'async_trait>(&'life0 mut self, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub struct iceberg::encryption::EncryptedInputFile impl iceberg::encryption::EncryptedInputFile @@ -281,7 +281,7 @@ pub async fn iceberg::encryption::EncryptedInputFile::exists(&self) -> iceberg:: pub fn iceberg::encryption::EncryptedInputFile::into_inner(self) -> iceberg::io::InputFile pub fn iceberg::encryption::EncryptedInputFile::key_metadata(&self) -> &iceberg::encryption::StandardKeyMetadata pub fn iceberg::encryption::EncryptedInputFile::location(&self) -> &str -pub async fn iceberg::encryption::EncryptedInputFile::metadata(&self) -> iceberg::Result<iceberg::io::FileMetadata> +pub fn iceberg::encryption::EncryptedInputFile::metadata(&self) -> iceberg::Result<iceberg::io::FileMetadata> pub fn iceberg::encryption::EncryptedInputFile::new(inner: iceberg::io::InputFile, key_metadata: iceberg::encryption::StandardKeyMetadata) -> Self pub async fn iceberg::encryption::EncryptedInputFile::read(&self) -> iceberg::Result<bytes::bytes::Bytes> pub async fn iceberg::encryption::EncryptedInputFile::reader(&self) -> iceberg::Result<alloc::boxed::Box<dyn iceberg::io::FileRead>> @@ -291,10 +291,10 @@ pub struct iceberg::encryption::EncryptedOutputFile impl iceberg::encryption::EncryptedOutputFile pub async fn iceberg::encryption::EncryptedOutputFile::delete(&self) -> iceberg::Result<()> pub fn iceberg::encryption::EncryptedOutputFile::into_inner(self) -> iceberg::io::OutputFile -pub fn iceberg::encryption::EncryptedOutputFile::key_metadata(&self) -> &iceberg::encryption::StandardKeyMetadata +pub fn iceberg::encryption::EncryptedOutputFile::key_metadata_with_saved_file_metadata(&self, file_metadata: &iceberg::io::FileMetadata) -> iceberg::encryption::StandardKeyMetadata pub fn iceberg::encryption::EncryptedOutputFile::location(&self) -> &str pub fn iceberg::encryption::EncryptedOutputFile::new(inner: iceberg::io::OutputFile, key_metadata: iceberg::encryption::StandardKeyMetadata) -> Self -pub async fn iceberg::encryption::EncryptedOutputFile::write(&self, bs: bytes::bytes::Bytes) -> iceberg::Result<()> +pub async fn iceberg::encryption::EncryptedOutputFile::write(&self, bs: bytes::bytes::Bytes) -> iceberg::Result<iceberg::io::FileMetadata> pub async fn iceberg::encryption::EncryptedOutputFile::writer(&self) -> iceberg::Result<alloc::boxed::Box<dyn iceberg::io::FileWrite>> impl core::fmt::Debug for iceberg::encryption::EncryptedOutputFile pub fn iceberg::encryption::EncryptedOutputFile::fmt(&self, f: &mut core::fmt::Formatter<'_>) -> core::fmt::Result @@ -1046,10 +1046,10 @@ pub fn iceberg::encryption::AesGcmFileRead::read<'life0, 'async_trait>(&'life0 s impl<T: core::convert::AsRef<dyn iceberg::io::FileRead> + core::marker::Send + core::marker::Sync + core::marker::Unpin + 'static> iceberg::io::FileRead for T pub fn T::read<'life0, 'async_trait>(&'life0 self, range: core::ops::range::Range<u64>) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<bytes::bytes::Bytes>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub trait iceberg::io::FileWrite: core::marker::Send + core::marker::Unpin + 'static -pub fn iceberg::io::FileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +pub fn iceberg::io::FileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::io::FileWrite::write<'life0, 'async_trait>(&'life0 mut self, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait impl iceberg::io::FileWrite for iceberg::encryption::AesGcmFileWrite -pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait +pub fn iceberg::encryption::AesGcmFileWrite::close<'life0, 'async_trait>(&'life0 mut self) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<iceberg::io::FileMetadata>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub fn iceberg::encryption::AesGcmFileWrite::write<'life0, 'async_trait>(&'life0 mut self, bs: bytes::bytes::Bytes) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait pub trait iceberg::io::Storage: core::fmt::Debug + core::marker::Send + core::marker::Sync + typetag::Serialize + typetag::Deserialize pub fn iceberg::io::Storage::delete<'life0, 'life1, 'async_trait>(&'life0 self, path: &'life1 str) -> core::pin::Pin<alloc::boxed::Box<(dyn core::future::future::Future<Output = iceberg::Result<()>> + core::marker::Send + 'async_trait)>> where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait @@ -1267,7 +1267,7 @@ pub async fn iceberg::puffin::PuffinReader::new_from_encrypted(encrypted_input: pub struct iceberg::puffin::PuffinWriter impl iceberg::puffin::PuffinWriter pub async fn iceberg::puffin::PuffinWriter::add(&mut self, blob: iceberg::puffin::Blob, compression_codec: iceberg::compression::CompressionCodec) -> iceberg::Result<()> -pub async fn iceberg::puffin::PuffinWriter::close(self) -> iceberg::Result<()> +pub async fn iceberg::puffin::PuffinWriter::close(self) -> iceberg::Result<iceberg::io::FileMetadata> pub async fn iceberg::puffin::PuffinWriter::new(output_file: &iceberg::io::OutputFile, properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, compress_footer: bool) -> iceberg::Result<Self> pub async fn iceberg::puffin::PuffinWriter::new_from_encrypted(encrypted_output: &iceberg::encryption::EncryptedOutputFile, properties: std::collections::hash::map::HashMap<alloc::string::String, alloc::string::String>, compress_footer: bool) -> iceberg::Result<Self> pub const iceberg::puffin::APACHE_DATASKETCHES_THETA_V1: &str @@ -2138,7 +2138,7 @@ pub async fn iceberg::spec::ManifestListReader::load(&self) -> iceberg::Result<i pub struct iceberg::spec::ManifestListWriter impl iceberg::spec::ManifestListWriter pub fn iceberg::spec::ManifestListWriter::add_manifests(&mut self, manifests: impl core::iter::traits::iterator::Iterator<Item = iceberg::spec::ManifestFile>) -> iceberg::Result<()> -pub async fn iceberg::spec::ManifestListWriter::close(self) -> iceberg::Result<()> +pub async fn iceberg::spec::ManifestListWriter::close(self) -> iceberg::Result<iceberg::io::FileMetadata> pub fn iceberg::spec::ManifestListWriter::next_row_id(&self) -> core::option::Option<u64> pub fn iceberg::spec::ManifestListWriter::v1(writer: alloc::boxed::Box<dyn iceberg::io::FileWrite>, snapshot_id: i64, parent_snapshot_id: core::option::Option<i64>) -> Self pub fn iceberg::spec::ManifestListWriter::v2(writer: alloc::boxed::Box<dyn iceberg::io::FileWrite>, snapshot_id: i64, parent_snapshot_id: core::option::Option<i64>, sequence_number: i64) -> Self diff --git a/crates/iceberg/src/arrow/caching_delete_file_loader.rs b/crates/iceberg/src/arrow/caching_delete_file_loader.rs index 0208a678e..e1cb6de15 100644 --- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs +++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs @@ -1621,13 +1621,16 @@ mod tests { let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef") .unwrap() .with_aad_prefix(b"test-aad-prefix!"); - let encoded_key_metadata = key_metadata.encode().unwrap(); let blob = encode_dv_blob([2u64, 4]); let plaintext_size = blob.len() as i64; let dv_path = format!("{table_location}/deletes.puffin"); let output = EncryptedOutputFile::new(file_io.new_output(&dv_path).unwrap(), key_metadata); - output.write(Bytes::from(blob)).await.unwrap(); + let file_metadata = output.write(Bytes::from(blob)).await.unwrap(); + let encoded_key_metadata = output + .key_metadata_with_saved_file_metadata(&file_metadata) + .encode() + .unwrap(); // content_offset / content_size_in_bytes are in the plaintext coordinate space, distinct // from the ciphertext's on-disk size (header, nonce, and tag overhead). diff --git a/crates/iceberg/src/encryption/io.rs b/crates/iceberg/src/encryption/io.rs index 3eafbe82f..17fc1a1f2 100644 --- a/crates/iceberg/src/encryption/io.rs +++ b/crates/iceberg/src/encryption/io.rs @@ -23,8 +23,9 @@ use bytes::Bytes; use super::crypto::AesGcmCipher; use super::key_metadata::StandardKeyMetadata; -use super::stream::{AesGcmFileRead, AesGcmFileWrite}; +use super::stream::{AesGcmFileRead, AesGcmFileWrite, MIN_STREAM_LENGTH}; use crate::Result; +use crate::error::invalid_data; use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; /// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`]. @@ -54,12 +55,11 @@ impl EncryptedInputFile { self.inner.exists().await } - /// Fetch and returns metadata of file. + /// Returns file metadata from the declared encrypted length without performing I/O. /// /// The returned size is the **plaintext** size. - pub async fn metadata(&self) -> Result<FileMetadata> { - let raw_meta = self.inner.metadata().await?; - let plaintext_size = AesGcmFileRead::calculate_plaintext_length(raw_meta.size)?; + pub fn metadata(&self) -> Result<FileMetadata> { + let plaintext_size = AesGcmFileRead::calculate_plaintext_length(self.encrypted_length()?)?; Ok(FileMetadata { size: plaintext_size, }) @@ -67,21 +67,34 @@ impl EncryptedInputFile { /// Read and returns whole content of file (decrypted plaintext). pub async fn read(&self) -> Result<Bytes> { - let meta = self.metadata().await?; + let meta = self.metadata()?; let reader = self.reader().await?; reader.read(0..meta.size).await } /// Creates a reader that transparently decrypts on each read. pub async fn reader(&self) -> Result<Box<dyn FileRead>> { - let raw_meta = self.inner.metadata().await?; + let encrypted_length = self.encrypted_length()?; let raw_reader = self.inner.reader().await?; let cipher = build_cipher(&self.key_metadata)?; let aad_prefix: Box<[u8]> = self.key_metadata.aad_prefix().unwrap_or_default().into(); - let decrypting = AesGcmFileRead::new(raw_reader, cipher, aad_prefix, raw_meta.size)?; + let decrypting = AesGcmFileRead::new(raw_reader, cipher, aad_prefix, encrypted_length)?; Ok(Box::new(decrypting)) } + // A storage stat would hide truncation; require the original length from key metadata. + fn encrypted_length(&self) -> Result<u64> { + let length = self.key_metadata.file_length().ok_or_else(|| { + invalid_data!("AGS1 key metadata is missing the encrypted file length") + })?; + if length < u64::from(MIN_STREAM_LENGTH) { + return Err(invalid_data!( + "Invalid encrypted file length: {length} is less than {MIN_STREAM_LENGTH}" + )); + } + Ok(length) + } + /// Returns a reference to the file's key metadata. pub fn key_metadata(&self) -> &StandardKeyMetadata { &self.key_metadata @@ -118,9 +131,14 @@ impl EncryptedOutputFile { } } - /// Returns a reference to the file's key metadata. - pub fn key_metadata(&self) -> &StandardKeyMetadata { - &self.key_metadata + /// Returns key metadata using the encrypted size returned by [`FileWrite::close`] or [`Self::write`]. + pub fn key_metadata_with_saved_file_metadata( + &self, + file_metadata: &FileMetadata, + ) -> StandardKeyMetadata { + self.key_metadata + .clone() + .with_file_length(file_metadata.size) } /// Absolute path of the file. @@ -138,8 +156,8 @@ impl EncryptedOutputFile { ))) } - /// Write bytes to file (transparently encrypted). - pub async fn write(&self, bs: Bytes) -> Result<()> { + /// Write bytes to the file and return its encrypted size. + pub async fn write(&self, bs: Bytes) -> Result<FileMetadata> { let mut writer = self.writer().await?; writer.write(bs).await?; writer.close().await @@ -172,6 +190,10 @@ fn build_cipher(metadata: &StandardKeyMetadata) -> Result<Arc<AesGcmCipher>> { #[cfg(test)] mod tests { use super::*; + use crate::ErrorKind; + use crate::encryption::stream::{ + CIPHER_BLOCK_SIZE, GCM_STREAM_HEADER_LENGTH, PLAIN_BLOCK_SIZE, + }; use crate::io::FileIO; fn key_metadata() -> StandardKeyMetadata { @@ -187,9 +209,12 @@ mod tests { let plaintext = b"Hello from EncryptedInputFile/EncryptedOutputFile!"; let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); - output.write(Bytes::from(plaintext.to_vec())).await.unwrap(); + let file_metadata = output.write(Bytes::from(plaintext.to_vec())).await.unwrap(); - let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), key_metadata()); + let input = EncryptedInputFile::new( + fileio.new_input(path).unwrap(), + output.key_metadata_with_saved_file_metadata(&file_metadata), + ); let content = input.read().await.unwrap(); assert_eq!(&content[..], plaintext); } @@ -201,7 +226,7 @@ mod tests { let plaintext = b"some bytes to measure"; let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); - output.write(Bytes::from(plaintext.to_vec())).await.unwrap(); + let file_metadata = output.write(Bytes::from(plaintext.to_vec())).await.unwrap(); let raw_size = fileio .new_input(path) @@ -215,8 +240,172 @@ mod tests { "encrypted file should be larger than plaintext (header + nonce + tag)" ); - let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), key_metadata()); - let meta = input.metadata().await.unwrap(); + // A missing path proves the size comes from the key metadata rather than a stat call. + let input = EncryptedInputFile::new( + fileio.new_input("memory:///does-not-exist").unwrap(), + output.key_metadata_with_saved_file_metadata(&file_metadata), + ); + let meta = input.metadata().unwrap(); assert_eq!(meta.size, plaintext.len() as u64); } + + #[tokio::test] + async fn test_missing_file_length_is_rejected() { + let fileio = FileIO::new_with_memory(); + let path = "memory:///test/missing_length.bin"; + let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); + output.write(Bytes::from_static(b"data")).await.unwrap(); + let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), key_metadata()); + + for err in [ + input.metadata().err().unwrap(), + input.reader().await.err().unwrap(), + input.read().await.unwrap_err(), + ] { + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!( + err.to_string() + .contains("missing the encrypted file length") + ); + } + } + + #[tokio::test] + async fn test_invalid_file_length_is_rejected() { + let fileio = FileIO::new_with_memory(); + for length in [ + 0, + u64::from(GCM_STREAM_HEADER_LENGTH), + u64::from(MIN_STREAM_LENGTH - 1), + ] { + let input = EncryptedInputFile::new( + fileio + .new_input("memory:///test/invalid_length.bin") + .unwrap(), + key_metadata().with_file_length(length), + ); + assert_eq!( + input.metadata().err().unwrap().kind(), + ErrorKind::DataInvalid + ); + assert_eq!( + input.reader().await.err().unwrap().kind(), + ErrorKind::DataInvalid + ); + } + } + + #[tokio::test] + async fn test_oversized_file_length_is_rejected() { + let fileio = FileIO::new_with_memory(); + let path = "memory:///test/oversized_length.bin"; + let plaintext = Bytes::from_static(b"some bytes to measure"); + let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); + let file_metadata = output.write(plaintext.clone()).await.unwrap(); + + // A declared length is trusted without a stat, so an inflated one is only caught once a + // read runs off the end of the real file. Both a minimal overstatement and one spanning a + // whole extra block must fail rather than silently return short plaintext. + for excess in [1, u64::from(CIPHER_BLOCK_SIZE)] { + let input = EncryptedInputFile::new( + fileio.new_input(path).unwrap(), + key_metadata().with_file_length(file_metadata.size + excess), + ); + + let inflated_size = input.metadata().unwrap().size; + assert!(inflated_size > plaintext.len() as u64); + + assert_eq!( + input.read().await.unwrap_err().kind(), + ErrorKind::DataInvalid + ); + + // Not even the bytes that genuinely are on disk can be read back. + let reader = input.reader().await.unwrap(); + assert_eq!( + reader + .read(0..plaintext.len() as u64) + .await + .unwrap_err() + .kind(), + ErrorKind::DataInvalid + ); + } + } + + #[tokio::test] + async fn test_truncated_file_is_rejected() { + let fileio = FileIO::new_with_memory(); + let path = "memory:///test/truncated.bin"; + let plaintext = Bytes::from(vec![42; 2 * PLAIN_BLOCK_SIZE as usize + 17]); + let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); + let file_metadata = output.write(plaintext.clone()).await.unwrap(); + let metadata = key_metadata().with_file_length(file_metadata.size); + let ciphertext = fileio.new_input(path).unwrap().read().await.unwrap(); + let truncated_length = (GCM_STREAM_HEADER_LENGTH + CIPHER_BLOCK_SIZE) as usize; + fileio + .new_output(path) + .unwrap() + .write(ciphertext.slice(..truncated_length)) + .await + .unwrap(); + + let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), metadata); + assert_eq!(input.metadata().unwrap().size, plaintext.len() as u64); + let reader = input.reader().await.unwrap(); + assert_eq!( + reader.read(0..u64::from(PLAIN_BLOCK_SIZE)).await.unwrap(), + plaintext.slice(..PLAIN_BLOCK_SIZE as usize) + ); + assert_eq!( + input.read().await.unwrap_err().kind(), + ErrorKind::DataInvalid + ); + } + + #[tokio::test] + async fn test_truncated_empty_file_is_rejected() { + let fileio = FileIO::new_with_memory(); + let path = "memory:///test/truncated_empty.bin"; + let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); + let file_metadata = output.write(Bytes::new()).await.unwrap(); + let metadata = key_metadata().with_file_length(file_metadata.size); + let ciphertext = fileio.new_input(path).unwrap().read().await.unwrap(); + fileio + .new_output(path) + .unwrap() + .write(ciphertext.slice(..GCM_STREAM_HEADER_LENGTH as usize)) + .await + .unwrap(); + let input = EncryptedInputFile::new(fileio.new_input(path).unwrap(), metadata); + assert_eq!( + input.read().await.unwrap_err().kind(), + ErrorKind::DataInvalid + ); + } + + #[tokio::test] + async fn test_close_returns_encrypted_size() { + let fileio = FileIO::new_with_memory(); + let path = "memory:///test/streaming.bin"; + let output = EncryptedOutputFile::new(fileio.new_output(path).unwrap(), key_metadata()); + for plaintext in [ + Bytes::from(vec![42; PLAIN_BLOCK_SIZE as usize + 17]), + Bytes::new(), + ] { + let mut writer = output.writer().await.unwrap(); + for chunk in plaintext.chunks(1024) { + writer.write(Bytes::copy_from_slice(chunk)).await.unwrap(); + } + let metadata = writer.close().await.unwrap(); + let size = fileio + .new_input(path) + .unwrap() + .metadata() + .await + .unwrap() + .size; + assert_eq!(metadata.size, size); + } + } } diff --git a/crates/iceberg/src/encryption/key_metadata.rs b/crates/iceberg/src/encryption/key_metadata.rs index 4f8445a9a..d56c8dbd3 100644 --- a/crates/iceberg/src/encryption/key_metadata.rs +++ b/crates/iceberg/src/encryption/key_metadata.rs @@ -75,7 +75,7 @@ impl StandardKeyMetadata { self } - /// Adds a file length. + /// Sets the encrypted file length in bytes, required for AGS1 truncation protection. pub fn with_file_length(mut self, length: u64) -> Self { self.file_length = Some(length); self @@ -91,7 +91,7 @@ impl StandardKeyMetadata { self.aad_prefix.as_deref() } - /// Returns the optional file length. + /// Returns the optional encrypted file length in bytes; AGS1 readers require it to be set. pub fn file_length(&self) -> Option<u64> { self.file_length } diff --git a/crates/iceberg/src/encryption/manager.rs b/crates/iceberg/src/encryption/manager.rs index d6f4e4af6..47ff98160 100644 --- a/crates/iceberg/src/encryption/manager.rs +++ b/crates/iceberg/src/encryption/manager.rs @@ -670,12 +670,16 @@ mod tests { let encrypted_output = mgr.encrypt(output); let plaintext = b"Hello, encrypted Iceberg round-trip!"; - let serialized_metadata = encrypted_output.key_metadata().encode().unwrap(); - encrypted_output + let file_metadata = encrypted_output .write(bytes::Bytes::from(plaintext.to_vec())) .await .unwrap(); + let serialized_metadata = encrypted_output + .key_metadata_with_saved_file_metadata(&file_metadata) + .encode() + .unwrap(); + let input = io.new_input(path).unwrap(); let parsed_metadata = StandardKeyMetadata::decode(&serialized_metadata).unwrap(); let decrypted_file = EncryptedInputFile::new(input, parsed_metadata); diff --git a/crates/iceberg/src/encryption/stream.rs b/crates/iceberg/src/encryption/stream.rs index c14905f70..14c0d3844 100644 --- a/crates/iceberg/src/encryption/stream.rs +++ b/crates/iceberg/src/encryption/stream.rs @@ -50,7 +50,7 @@ use bytes::{Bytes, BytesMut}; use super::AesGcmCipher; use crate::error::invalid_data; -use crate::io::{FileRead, FileWrite}; +use crate::io::{FileMetadata, FileRead, FileWrite}; use crate::{Error, ErrorKind, Result}; /// Default plaintext block size (1 MiB), matching Java's `Ciphers.PLAIN_BLOCK_SIZE`. @@ -72,8 +72,7 @@ pub const GCM_STREAM_MAGIC: [u8; 4] = *b"AGS1"; pub const GCM_STREAM_HEADER_LENGTH: u32 = 8; /// Minimum valid AGS1 stream length (header + one empty block). -#[cfg(test)] -pub const MIN_STREAM_LENGTH: u32 = GCM_STREAM_HEADER_LENGTH + NONCE_LENGTH + GCM_TAG_LENGTH; +pub(crate) const MIN_STREAM_LENGTH: u32 = GCM_STREAM_HEADER_LENGTH + NONCE_LENGTH + GCM_TAG_LENGTH; /// Constructs the per-block AAD for AGS1 stream encryption. /// @@ -147,20 +146,14 @@ impl AesGcmFileRead { aad_prefix: Box<[u8]>, encrypted_file_length: u64, ) -> Result<Self> { + if encrypted_file_length < u64::from(MIN_STREAM_LENGTH) { + return Err(invalid_data!( + "Invalid encrypted file length: {encrypted_file_length} is less than {MIN_STREAM_LENGTH}" + )); + } let plain_stream_size = Self::calculate_plaintext_length(encrypted_file_length)?; let stream_length = encrypted_file_length - GCM_STREAM_HEADER_LENGTH as u64; - if stream_length == 0 { - return Ok(Self { - inner, - cipher, - aad_prefix, - plain_stream_size: 0, - num_blocks: 0, - last_cipher_block_size: 0, - }); - } - let num_full_blocks = stream_length / CIPHER_BLOCK_SIZE as u64; let cipher_bytes_in_last_block = (stream_length % CIPHER_BLOCK_SIZE as u64) as u32; let full_blocks_only = cipher_bytes_in_last_block == 0; @@ -280,7 +273,9 @@ impl FileRead for AesGcmFileRead { /// file. GCM authentication is verified per-block, so any tampering is detected /// at the granularity of individual blocks. async fn read(&self, range: Range<u64>) -> Result<Bytes> { - if range.start == range.end { + // An empty stream still has one block whose GCM tag must be verified, so an empty read + // cannot short-circuit there: that is what detects a file truncated to just its header. + if range.start == range.end && self.plain_stream_size != 0 { return Ok(Bytes::new()); } @@ -301,12 +296,8 @@ impl FileRead for AesGcmFileRead { )); } - if self.num_blocks == 0 { - return Ok(Bytes::new()); - } - let first_block = range.start / PLAIN_BLOCK_SIZE as u64; - let last_block = (range.end - 1) / PLAIN_BLOCK_SIZE as u64; + let last_block = range.end.saturating_sub(1) / PLAIN_BLOCK_SIZE as u64; // Read all needed encrypted blocks in a single I/O call let encrypted_start = Self::encrypted_block_offset(first_block); @@ -314,6 +305,13 @@ impl FileRead for AesGcmFileRead { Self::encrypted_block_offset(last_block) + self.cipher_block_size(last_block) as u64; let all_encrypted = self.inner.read(encrypted_start..encrypted_end).await?; + if all_encrypted.len() as u64 != encrypted_end - encrypted_start { + return Err(invalid_data!( + "Invalid encrypted read length: expected {} bytes, got {}", + encrypted_end - encrypted_start, + all_encrypted.len() + )); + } // Decrypt each block and extract the requested plaintext range let result_len = (range.end - range.start) as usize; @@ -489,7 +487,7 @@ impl FileWrite for AesGcmFileWrite { Ok(()) } - async fn close(&mut self) -> Result<()> { + async fn close(&mut self) -> Result<FileMetadata> { if self.closed { return Err(Error::new( ErrorKind::Unexpected, @@ -624,6 +622,69 @@ mod tests { assert!(result.is_empty()); } + #[tokio::test] + async fn test_short_ciphertext_read_is_rejected() { + struct ShortRead; + + #[async_trait::async_trait] + impl FileRead for ShortRead { + async fn read(&self, range: Range<u64>) -> Result<Bytes> { + let len = (range.end - range.start).saturating_sub(1) as usize; + Ok(Bytes::from(vec![0; len])) + } + } + + let reader = AesGcmFileRead::new( + Box::new(ShortRead), + Arc::new(make_cipher(b"0123456789abcdef")), + Box::default(), + u64::from(MIN_STREAM_LENGTH) + 10, + ) + .unwrap(); + let err = reader.read(0..10).await.unwrap_err(); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.to_string().contains("Invalid encrypted read length")); + } + + #[tokio::test] + async fn test_oversized_declared_length_is_rejected() { + // Object stores clamp reads at EOF rather than failing, unlike the in-memory FileIO. + struct ClampingRead(Bytes); + + #[async_trait::async_trait] + impl FileRead for ClampingRead { + async fn read(&self, range: Range<u64>) -> Result<Bytes> { + let start = (range.start as usize).min(self.0.len()); + let end = (range.end as usize).min(self.0.len()); + Ok(self.0.slice(start..end)) + } + } + + let key = b"0123456789abcdef"; + let aad_prefix = b"test-aad-prefix!"; + let plaintext = b"some bytes to measure"; + let encrypted = write_through_ags1(plaintext, key, aad_prefix).await; + + // A declared length larger than the real file must be rejected rather than yielding + // truncated plaintext, whether it overstates by one byte or by a whole extra block. + for excess in [1, u64::from(CIPHER_BLOCK_SIZE)] { + let reader = AesGcmFileRead::new( + Box::new(ClampingRead(Bytes::from(encrypted.clone()))), + Arc::new(make_cipher(key)), + aad_prefix.to_vec().into_boxed_slice(), + encrypted.len() as u64 + excess, + ) + .unwrap(); + + let err = reader + .read(0..plaintext.len() as u64) + .await + .expect_err("an inflated declared length must not read back as plaintext"); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.to_string().contains("Invalid encrypted read length")); + } + } + #[tokio::test] async fn test_small_file_roundtrip() { let key = b"0123456789abcdef"; @@ -940,15 +1001,21 @@ mod tests { assert_eq!(&aad[..], &42u32.to_le_bytes()); } - #[tokio::test] - async fn test_encrypted_file_too_short() { - let result = AesGcmFileRead::new( - memory_reader(vec![0; 4]), - Arc::new(make_cipher(b"0123456789abcdef")), - [].into(), - 4, - ); - assert!(result.is_err()); + #[test] + fn test_encrypted_file_too_short() { + for length in 0..MIN_STREAM_LENGTH { + let result = AesGcmFileRead::new( + memory_reader(vec![0; length as usize]), + Arc::new(make_cipher(b"0123456789abcdef")), + [].into(), + u64::from(length), + ); + let err = result + .err() + .expect("a stream must contain an authenticated block"); + assert_eq!(err.kind(), ErrorKind::DataInvalid); + assert!(err.to_string().contains("Invalid encrypted file length")); + } } // --- AesGcmFileWrite tests --- @@ -974,8 +1041,14 @@ mod tests { Ok(()) } - async fn close(&mut self) -> Result<()> { - Ok(()) + // A poisoned AesGcmFileWrite must reject close() before delegating, so reaching this is a + // bug. Returning an error rather than panicking surfaces it as a failed assertion on the + // expected "poisoned" message. + async fn close(&mut self) -> Result<FileMetadata> { + Err(Error::new( + ErrorKind::Unexpected, + "FailingFileWrite::close called unexpectedly", + )) } } @@ -986,8 +1059,10 @@ mod tests { Ok(()) } - async fn close(&mut self) -> Result<()> { - Ok(()) + async fn close(&mut self) -> Result<FileMetadata> { + Ok(FileMetadata { + size: self.buffer.lock().unwrap().len() as u64, + }) } } @@ -1001,9 +1076,15 @@ mod tests { let mut writer = AesGcmFileWrite::new(inner, cipher, aad_prefix.to_vec()); writer.write(Bytes::from(plaintext.to_vec())).await.unwrap(); - writer.close().await.unwrap(); + let metadata = writer.close().await.unwrap(); - buffer.lock().unwrap().clone() + let encrypted = buffer.lock().unwrap().clone(); + assert_eq!( + metadata.size, + encrypted.len() as u64, + "close() must report the full ciphertext length" + ); + encrypted } #[tokio::test] @@ -1223,10 +1304,9 @@ mod tests { ); // Close should also be rejected - let result = writer.close().await; - assert!(result.is_err()); + let err = writer.close().await.err().expect("close should fail"); assert!( - result.unwrap_err().to_string().contains("poisoned"), + err.to_string().contains("poisoned"), "expected poisoned error on close" ); } diff --git a/crates/iceberg/src/io/file_io.rs b/crates/iceberg/src/io/file_io.rs index 87ac84784..42eec4db5 100644 --- a/crates/iceberg/src/io/file_io.rs +++ b/crates/iceberg/src/io/file_io.rs @@ -375,10 +375,10 @@ pub trait FileWrite: Send + Unpin + 'static { /// TODO: we can support writing non-contiguous bytes in the future. async fn write(&mut self, bs: Bytes) -> Result<()>; - /// Close file. + /// Close the file and return its stored size, including encryption overhead for encrypted files. /// /// Calling close on closed file will generate an error. - async fn close(&mut self) -> Result<()>; + async fn close(&mut self) -> Result<FileMetadata>; } /// Output file is used for writing to files.. diff --git a/crates/iceberg/src/io/storage/local_fs.rs b/crates/iceberg/src/io/storage/local_fs.rs index d5ef0a473..cb780e86e 100644 --- a/crates/iceberg/src/io/storage/local_fs.rs +++ b/crates/iceberg/src/io/storage/local_fs.rs @@ -249,12 +249,16 @@ impl FileRead for LocalFsFileRead { #[derive(Debug)] pub struct LocalFsFileWrite { file: Option<fs::File>, + bytes_written: u64, } impl LocalFsFileWrite { /// Create a new `LocalFsFileWrite` for the given file. pub fn new(file: fs::File) -> Self { - Self { file: Some(file) } + Self { + file: Some(file), + bytes_written: 0, + } } } @@ -272,11 +276,12 @@ impl FileWrite for LocalFsFileWrite { format!("Failed to write to file: {e}"), ) })?; + self.bytes_written += bs.len() as u64; Ok(()) } - async fn close(&mut self) -> Result<()> { + async fn close(&mut self) -> Result<FileMetadata> { let file = self .file .take() @@ -285,7 +290,9 @@ impl FileWrite for LocalFsFileWrite { file.sync_all() .map_err(|e| Error::new(ErrorKind::Unexpected, format!("Failed to sync file: {e}")))?; - Ok(()) + Ok(FileMetadata { + size: self.bytes_written, + }) } } @@ -464,10 +471,11 @@ mod tests { let mut writer = storage.writer(path_str).await.unwrap(); writer.write(Bytes::from("Hello, ")).await.unwrap(); writer.write(Bytes::from("World!")).await.unwrap(); - writer.close().await.unwrap(); + let metadata = writer.close().await.unwrap(); let content = storage.read(path_str).await.unwrap(); assert_eq!(content, Bytes::from("Hello, World!")); + assert_eq!(metadata.size, content.len() as u64); } #[tokio::test] @@ -494,7 +502,7 @@ mod tests { let path_str = path.to_str().unwrap(); let mut writer = storage.writer(path_str).await.unwrap(); - writer.close().await.unwrap(); + assert_eq!(writer.close().await.unwrap().size, 0); // Write after close should fail let result = writer.write(Bytes::from("test")).await; diff --git a/crates/iceberg/src/io/storage/memory.rs b/crates/iceberg/src/io/storage/memory.rs index b39c741ea..fefa7973e 100644 --- a/crates/iceberg/src/io/storage/memory.rs +++ b/crates/iceberg/src/io/storage/memory.rs @@ -311,7 +311,7 @@ impl FileWrite for MemoryFileWrite { Ok(()) } - async fn close(&mut self) -> Result<()> { + async fn close(&mut self) -> Result<FileMetadata> { if self.closed { return Err(invalid_data!("File already closed")); } @@ -323,12 +323,13 @@ impl FileWrite for MemoryFileWrite { ) })?; + let size = self.buffer.len() as u64; data.insert( self.path.clone(), Bytes::from(std::mem::take(&mut self.buffer)), ); self.closed = true; - Ok(()) + Ok(FileMetadata { size }) } } @@ -482,10 +483,11 @@ mod tests { let mut writer = storage.writer(path).await.unwrap(); writer.write(Bytes::from("Hello, ")).await.unwrap(); writer.write(Bytes::from("World!")).await.unwrap(); - writer.close().await.unwrap(); + let metadata = writer.close().await.unwrap(); let content = storage.read(path).await.unwrap(); assert_eq!(content, Bytes::from("Hello, World!")); + assert_eq!(metadata.size, content.len() as u64); } #[tokio::test] @@ -508,7 +510,7 @@ mod tests { let path = "memory://test/file.txt"; let mut writer = storage.writer(path).await.unwrap(); - writer.close().await.unwrap(); + assert_eq!(writer.close().await.unwrap().size, 0); // Write after close should fail let result = writer.write(Bytes::from("test")).await; diff --git a/crates/iceberg/src/puffin/reader.rs b/crates/iceberg/src/puffin/reader.rs index 01601f48e..a7b178ac8 100644 --- a/crates/iceberg/src/puffin/reader.rs +++ b/crates/iceberg/src/puffin/reader.rs @@ -45,7 +45,7 @@ impl PuffinReader { /// reader operates over plaintext offsets and length, so all blob and /// footer positions match those written to the unencrypted file. pub async fn new_from_encrypted(encrypted_input: EncryptedInputFile) -> Result<Self> { - let file_length = encrypted_input.metadata().await?.size; + let file_length = encrypted_input.metadata()?.size; let file_read = encrypted_input.reader().await?; Ok(Self::from_parts(file_read, file_length)) } diff --git a/crates/iceberg/src/puffin/writer.rs b/crates/iceberg/src/puffin/writer.rs index 0437bd5bf..e6882a484 100644 --- a/crates/iceberg/src/puffin/writer.rs +++ b/crates/iceberg/src/puffin/writer.rs @@ -23,7 +23,7 @@ use super::validate_puffin_compression; use crate::Result; use crate::compression::CompressionCodec; use crate::encryption::EncryptedOutputFile; -use crate::io::{FileWrite, OutputFile}; +use crate::io::{FileMetadata as IoFileMetadata, FileWrite, OutputFile}; use crate::puffin::blob::Blob; use crate::puffin::metadata::{BlobMetadata, FileMetadata, Flag}; @@ -117,12 +117,11 @@ impl PuffinWriter { Ok(()) } - /// Finalizes the Puffin file - pub async fn close(mut self) -> Result<()> { + /// Finalizes the Puffin file and returns its stored size. + pub async fn close(mut self) -> Result<IoFileMetadata> { self.write_header_once().await?; self.write_footer().await?; - self.writer.close().await?; - Ok(()) + self.writer.close().await } async fn write(&mut self, bytes: Bytes) -> Result<()> { @@ -409,7 +408,7 @@ mod tests { for blob in blobs.clone() { writer.add(blob, CompressionCodec::None).await.unwrap(); } - writer.close().await.unwrap(); + let file_metadata = writer.close().await.unwrap(); // The ciphertext on disk must not equal a plaintext puffin file. let raw = file_io.new_input(path).unwrap().read().await.unwrap(); @@ -419,8 +418,10 @@ mod tests { ); // Read back through the decrypting reader over plaintext offsets. + let key_metadata = encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata); + assert_eq!(key_metadata.file_length(), Some(raw.len() as u64)); let encrypted_input = - EncryptedInputFile::new(file_io.new_input(path).unwrap(), key_metadata()); + EncryptedInputFile::new(file_io.new_input(path).unwrap(), key_metadata); let reader = PuffinReader::new_from_encrypted(encrypted_input) .await .unwrap(); diff --git a/crates/iceberg/src/spec/manifest/writer.rs b/crates/iceberg/src/spec/manifest/writer.rs index b80567a68..15608a63f 100644 --- a/crates/iceberg/src/spec/manifest/writer.rs +++ b/crates/iceberg/src/spec/manifest/writer.rs @@ -16,8 +16,6 @@ // under the License. use std::cmp::min; -use std::future::Future; -use std::pin::Pin; use apache_avro::{Writer as AvroWriter, to_value}; use bytes::Bytes; @@ -30,7 +28,7 @@ use super::{ }; use crate::encryption::EncryptedOutputFile; use crate::error::{Result, invalid_data}; -use crate::io::{FileWrite, OutputFile}; +use crate::io::{FileMetadata, FileWrite, OutputFile}; use crate::spec::manifest::_serde::{ManifestEntryV1, ManifestEntryV2}; use crate::spec::manifest::{manifest_schema_v1, manifest_schema_v2}; use crate::spec::{ @@ -42,14 +40,38 @@ use crate::spec::{ /// with the actual snapshot ID before it is committed. const UNASSIGNED_SNAPSHOT_ID: i64 = -1; -type WriterFuture = Pin<Box<dyn Future<Output = Result<Box<dyn FileWrite>>> + Send>>; +/// Retains the output until close provides the size required by encrypted key metadata. +pub(crate) enum ManifestOutput { + Plain(OutputFile), + Encrypted(EncryptedOutputFile), +} + +impl ManifestOutput { + async fn writer(&self) -> Result<Box<dyn FileWrite>> { + match self { + Self::Plain(output) => output.writer().await, + Self::Encrypted(output) => output.writer().await, + } + } + + fn encoded_key_metadata(&self, file_metadata: &FileMetadata) -> Result<Option<Vec<u8>>> { + match self { + Self::Plain(_) => Ok(None), + Self::Encrypted(output) => Ok(Some( + output + .key_metadata_with_saved_file_metadata(file_metadata) + .encode()? + .into_vec(), + )), + } + } +} /// The builder used to create a [`ManifestWriter`]. pub struct ManifestWriterBuilder { - writer_future: WriterFuture, + output: ManifestOutput, location: String, snapshot_id: Option<i64>, - key_metadata: Option<Vec<u8>>, schema: SchemaRef, partition_spec: PartitionSpec, } @@ -64,10 +86,9 @@ impl ManifestWriterBuilder { ) -> Self { let location = output.location().to_owned(); Self { - writer_future: Box::pin(async move { output.writer().await }), + output: ManifestOutput::Plain(output), location, snapshot_id, - key_metadata: None, schema, partition_spec, } @@ -83,12 +104,10 @@ impl ManifestWriterBuilder { partition_spec: PartitionSpec, ) -> Result<Self> { let location = encrypted_output.location().to_owned(); - let key_metadata = Some(encrypted_output.key_metadata().encode()?.to_vec()); Ok(Self { - writer_future: Box::pin(async move { encrypted_output.writer().await }), + output: ManifestOutput::Encrypted(encrypted_output), location, snapshot_id, - key_metadata, schema, partition_spec, }) @@ -103,14 +122,7 @@ impl ManifestWriterBuilder { .format_version(FormatVersion::V1) .content(ManifestContentType::Data) .build(); - ManifestWriter::new( - self.writer_future, - self.location, - self.snapshot_id, - self.key_metadata, - metadata, - None, - ) + ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None) } /// Build a [`ManifestWriter`] for format version 2, data content. @@ -123,14 +135,7 @@ impl ManifestWriterBuilder { .content(ManifestContentType::Data) .build(); - ManifestWriter::new( - self.writer_future, - self.location, - self.snapshot_id, - self.key_metadata, - metadata, - None, - ) + ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None) } /// Build a [`ManifestWriter`] for format version 2, deletes content. @@ -142,14 +147,7 @@ impl ManifestWriterBuilder { .format_version(FormatVersion::V2) .content(ManifestContentType::Deletes) .build(); - ManifestWriter::new( - self.writer_future, - self.location, - self.snapshot_id, - self.key_metadata, - metadata, - None, - ) + ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None) } /// Build a [`ManifestWriter`] for format version 2, data content. @@ -162,10 +160,9 @@ impl ManifestWriterBuilder { .content(ManifestContentType::Data) .build(); ManifestWriter::new( - self.writer_future, + self.output, self.location, self.snapshot_id, - self.key_metadata, metadata, // First row id is assigned by the [`ManifestListWriter`] when the manifest // is added to the list. @@ -182,20 +179,13 @@ impl ManifestWriterBuilder { .format_version(FormatVersion::V3) .content(ManifestContentType::Deletes) .build(); - ManifestWriter::new( - self.writer_future, - self.location, - self.snapshot_id, - self.key_metadata, - metadata, - None, - ) + ManifestWriter::new(self.output, self.location, self.snapshot_id, metadata, None) } } /// A manifest writer. pub struct ManifestWriter { - writer_future: WriterFuture, + output: ManifestOutput, location: String, snapshot_id: Option<i64>, @@ -210,8 +200,6 @@ pub struct ManifestWriter { min_seq_num: Option<i64>, - key_metadata: Option<Vec<u8>>, - manifest_entries: Vec<ManifestEntry>, metadata: ManifestMetadata, @@ -220,15 +208,14 @@ pub struct ManifestWriter { impl ManifestWriter { /// Create a new manifest writer. pub(crate) fn new( - writer_future: WriterFuture, + output: ManifestOutput, location: String, snapshot_id: Option<i64>, - key_metadata: Option<Vec<u8>>, metadata: ManifestMetadata, first_row_id: Option<u64>, ) -> Self { Self { - writer_future, + output, location, snapshot_id, added_files: 0, @@ -239,7 +226,6 @@ impl ManifestWriter { deleted_rows: 0, first_row_id, min_seq_num: None, - key_metadata, manifest_entries: Vec::new(), metadata, } @@ -497,14 +483,15 @@ impl ManifestWriter { } let content = avro_writer.into_inner()?; - let length = content.len(); - let mut writer = self.writer_future.await?; + let mut writer = self.output.writer().await?; writer.write(Bytes::from(content)).await?; - writer.close().await?; + let file_metadata = writer.close().await?; + let key_metadata = self.output.encoded_key_metadata(&file_metadata)?; Ok(ManifestFile { manifest_path: self.location, - manifest_length: length as i64, + // Manifest lengths are on-disk sizes, including encryption overhead. + manifest_length: file_metadata.size.try_into()?, partition_spec_id: self.metadata.partition_spec.spec_id(), content: self.metadata.content, // sequence_number and min_sequence_number with UNASSIGNED_SEQUENCE_NUMBER will be replace with @@ -519,7 +506,7 @@ impl ManifestWriter { existing_rows_count: Some(self.existing_rows), deleted_rows_count: Some(self.deleted_rows), partitions: Some(partition_summary), - key_metadata: self.key_metadata, + key_metadata, first_row_id: self.first_row_id, }) } @@ -822,9 +809,9 @@ mod tests { assert_eq!(manifest_file.content, ManifestContentType::Deletes); // Read back the manifest file - let actual_manifest = - Manifest::parse_avro(fs::read(&path).expect("read_file must succeed").as_slice()) - .unwrap(); + let bytes = fs::read(&path).expect("read_file must succeed"); + assert_eq!(manifest_file.manifest_length, bytes.len() as i64); + let actual_manifest = Manifest::parse_avro(&bytes).unwrap(); // Verify the content type is correctly preserved as Deletes assert_eq!( diff --git a/crates/iceberg/src/spec/manifest_list/manifest_file.rs b/crates/iceberg/src/spec/manifest_list/manifest_file.rs index 341748a41..24df6d4dc 100644 --- a/crates/iceberg/src/spec/manifest_list/manifest_file.rs +++ b/crates/iceberg/src/spec/manifest_list/manifest_file.rs @@ -286,12 +286,15 @@ mod test { let key_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef") .unwrap() .with_aad_prefix(b"test-aad-prefix!"); - let encoded_key_metadata = key_metadata.encode().unwrap().to_vec(); - let io = FileIO::new_with_memory(); let path = "memory:///test/encrypted_manifest.avro"; - let manifest_file = write_encrypted_manifest(&io, path, key_metadata).await; - assert_eq!(manifest_file.key_metadata, Some(encoded_key_metadata)); + let manifest_file = write_encrypted_manifest(&io, path, key_metadata.clone()).await; + let size = io.new_input(path).unwrap().metadata().await.unwrap().size; + assert_eq!(manifest_file.manifest_length, size as i64); + assert_eq!( + StandardKeyMetadata::decode(manifest_file.key_metadata.as_ref().unwrap()).unwrap(), + key_metadata.with_file_length(size) + ); let manifest = ManifestReader::new(io).read(&manifest_file).await.unwrap(); assert_eq!(manifest.entries().len(), 1); @@ -318,7 +321,8 @@ mod test { // returning garbage. let wrong_key_metadata = StandardKeyMetadata::try_new(b"fedcba9876543210") .unwrap() - .with_aad_prefix(b"test-aad-prefix!"); + .with_aad_prefix(b"test-aad-prefix!") + .with_file_length(manifest_file.manifest_length as u64); manifest_file.key_metadata = Some(wrong_key_metadata.encode().unwrap().to_vec()); let err = ManifestReader::new(io) @@ -343,7 +347,8 @@ mod test { // so GCM authentication must fail even though the key is right. let wrong_aad_metadata = StandardKeyMetadata::try_new(b"0123456789abcdef") .unwrap() - .with_aad_prefix(b"wrong-aad-prefix"); + .with_aad_prefix(b"wrong-aad-prefix") + .with_file_length(manifest_file.manifest_length as u64); manifest_file.key_metadata = Some(wrong_aad_metadata.encode().unwrap().to_vec()); let err = ManifestReader::new(io) diff --git a/crates/iceberg/src/spec/manifest_list/writer.rs b/crates/iceberg/src/spec/manifest_list/writer.rs index 24613770d..5ce2cfb25 100644 --- a/crates/iceberg/src/spec/manifest_list/writer.rs +++ b/crates/iceberg/src/spec/manifest_list/writer.rs @@ -26,7 +26,7 @@ use super::_const_schema::{ use super::_serde::{ManifestFileV1, ManifestFileV2, ManifestFileV3}; use super::{FormatVersion, ManifestContentType, ManifestFile, UNASSIGNED_SEQUENCE_NUMBER}; use crate::error::{Result, invalid_data}; -use crate::io::FileWrite; +use crate::io::{FileMetadata, FileWrite}; use crate::{Error, ErrorKind}; /// A manifest list writer. @@ -196,12 +196,11 @@ impl ManifestListWriter { Ok(()) } - /// Write the manifest list to the output file. - pub async fn close(mut self) -> Result<()> { + /// Write the manifest list and return its stored size. + pub async fn close(mut self) -> Result<FileMetadata> { let data = self.avro_writer.into_inner()?; self.writer.write(Bytes::from(data)).await?; - self.writer.close().await?; - Ok(()) + self.writer.close().await } /// Assign sequence numbers to manifest if they are unassigned @@ -601,7 +600,6 @@ mod test { let path = "memory:///manifest_list_v3_encrypted.avro"; let encrypted_output = mgr.encrypt(file_io.new_output(path).unwrap()); - let key_metadata = encrypted_output.key_metadata().clone(); let snapshot_id = 9_000_000_000_000_001i64; let seq_num = 7i64; @@ -636,7 +634,7 @@ mod test { writer .add_manifests(expected.entries.clone().into_iter()) .unwrap(); - writer.close().await.unwrap(); + let file_metadata = writer.close().await.unwrap(); let raw_bytes = file_io.new_input(path).unwrap().read().await.unwrap(); assert!( @@ -644,6 +642,8 @@ mod test { "raw bytes should be ciphertext, not parseable as Avro" ); + let key_metadata = encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata); + assert_eq!(key_metadata.file_length(), Some(raw_bytes.len() as u64)); let plaintext = EncryptedInputFile::new(file_io.new_input(path).unwrap(), key_metadata) .read() .await diff --git a/crates/iceberg/src/transaction/append.rs b/crates/iceberg/src/transaction/append.rs index 872a1d89c..3260f9ffa 100644 --- a/crates/iceberg/src/transaction/append.rs +++ b/crates/iceberg/src/transaction/append.rs @@ -420,6 +420,25 @@ mod tests { }) .expect("a fast append should emit an AddSnapshot update"); + let manifest_list_key_metadata = table + .encryption_manager() + .unwrap() + .decrypt_manifest_list_key_metadata(new_snapshot.encryption_key_id().unwrap()) + .await + .unwrap(); + let manifest_list_size = table + .file_io() + .new_input(new_snapshot.manifest_list()) + .unwrap() + .metadata() + .await + .unwrap() + .size; + assert_eq!( + manifest_list_key_metadata.file_length(), + Some(manifest_list_size) + ); + let manifest_list = table .manifest_list_reader(&new_snapshot) .load() @@ -436,8 +455,18 @@ mod tests { .key_metadata .as_ref() .expect("encrypted manifest must record key metadata"); - StandardKeyMetadata::decode(key_metadata_bytes) + let key_metadata = StandardKeyMetadata::decode(key_metadata_bytes) .expect("recorded key metadata must decode as StandardKeyMetadata"); + let manifest_size = table + .file_io() + .new_input(&manifest_file.manifest_path) + .unwrap() + .metadata() + .await + .unwrap() + .size; + assert_eq!(key_metadata.file_length(), Some(manifest_size)); + assert_eq!(manifest_file.manifest_length, manifest_size as i64); // The reader self-decrypts using the recorded key metadata and must // recover the entry we appended. Because the read goes through the diff --git a/crates/iceberg/src/transaction/snapshot.rs b/crates/iceberg/src/transaction/snapshot.rs index 43e8b25d5..a0ce1fba1 100644 --- a/crates/iceberg/src/transaction/snapshot.rs +++ b/crates/iceberg/src/transaction/snapshot.rs @@ -444,13 +444,13 @@ impl<'a> SnapshotProducer<'a> { .file_io() .new_output(manifest_list_path.clone())?; - let (writer, encryption_key_id) = match self.table.encryption_manager() { + let (writer, encrypted_output) = match self.table.encryption_manager() { Some(em) => { let encrypted_output = em.encrypt(raw_output); - let key_id = em - .encrypt_manifest_list_key_metadata(encrypted_output.key_metadata()) - .await?; - (encrypted_output.writer().await?, Some(key_id)) + ( + encrypted_output.writer().await?, + Some((em.clone(), encrypted_output)), + ) } None => (raw_output.writer().await?, None), }; @@ -485,7 +485,16 @@ impl<'a> SnapshotProducer<'a> { manifest_list_writer.add_manifests(new_manifests.into_iter())?; let writer_next_row_id = manifest_list_writer.next_row_id(); - manifest_list_writer.close().await?; + let file_metadata = manifest_list_writer.close().await?; + let encryption_key_id = match encrypted_output { + Some((em, encrypted_output)) => Some( + em.encrypt_manifest_list_key_metadata( + &encrypted_output.key_metadata_with_saved_file_metadata(&file_metadata), + ) + .await?, + ), + None => None, + }; let commit_ts = chrono::Utc::now().timestamp_millis(); let new_snapshot = Snapshot::builder() diff --git a/crates/iceberg/src/writer/file_writer/parquet_writer.rs b/crates/iceberg/src/writer/file_writer/parquet_writer.rs index b874cb95a..7cdc48888 100644 --- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs +++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs @@ -732,9 +732,11 @@ impl ArrowAsyncFileWriter for AsyncFileWriter { fn complete(&mut self) -> BoxFuture<'_, parquet::errors::Result<()>> { Box::pin(async { + // TODO(encryption): retain the stored file size in data-file key metadata. self.0 .close() .await + .map(|_| ()) .map_err(|err| parquet::errors::ParquetError::External(Box::new(err))) }) } diff --git a/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json b/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json index 0dc0aabc1..1c096417f 100644 --- a/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json +++ b/crates/iceberg/testdata/table_metadata/TableMetadataV3ValidEncryption.json @@ -63,7 +63,7 @@ }, { "key-id": "1fb95f04-7f98-4b1f-a08f-e810cfb907ff", - "encrypted-key-metadata": "scAmTJzgxf7vDQBetLMWSKnZmOljEzoitb6H90HQ/wjDz5/naIkg44PkAg+Ldl6RBP0FBcoS1bfBDVLToUbdrVA=", + "encrypted-key-metadata": "XcrqLxxyf9a7FAFXM63u3MEH8CSWqh6g1H8fJD2/6zx0EjcPzPaP/5AtRpO2s6p2isV8s7a4m77zj+GIqNmIpbA2Xw==", "encrypted-by-id": "5b29941b-b3b4-4d57-8758-a9d1bf340b39" } ], diff --git a/crates/storage/opendal/src/lib.rs b/crates/storage/opendal/src/lib.rs index 2ed5a60cb..fe485cb91 100644 --- a/crates/storage/opendal/src/lib.rs +++ b/crates/storage/opendal/src/lib.rs @@ -551,7 +551,7 @@ impl Storage for OpenDalStorage { async fn writer(&self, path: &str) -> Result<Box<dyn FileWrite>> { let (op, relative_path) = self.create_operator(&path)?; - Ok(Box::new(OpenDalWriter( + Ok(Box::new(OpenDalWriter::new( op.writer(relative_path).await.map_err(from_opendal_error)?, ))) } @@ -635,21 +635,51 @@ impl FileRead for OpenDalReader { } /// Wrapper around `opendal::Writer` that implements `FileWrite`. -pub(crate) struct OpenDalWriter(pub(crate) opendal::Writer); +pub(crate) struct OpenDalWriter { + inner: opendal::Writer, + bytes_written: u64, +} + +impl OpenDalWriter { + pub(crate) fn new(inner: opendal::Writer) -> Self { + Self { + inner, + bytes_written: 0, + } + } +} #[async_trait] impl FileWrite for OpenDalWriter { async fn write(&mut self, bs: Bytes) -> Result<()> { - Ok(opendal::Writer::write(&mut self.0, bs) + let len = bs.len() as u64; + opendal::Writer::write(&mut self.inner, bs) .await - .map_err(from_opendal_error)?) + .map_err(from_opendal_error)?; + self.bytes_written += len; + Ok(()) } - async fn close(&mut self) -> Result<()> { - let _ = opendal::Writer::close(&mut self.0) + async fn close(&mut self) -> Result<FileMetadata> { + let metadata = opendal::Writer::close(&mut self.inner) .await .map_err(from_opendal_error)?; - Ok(()) + + // Object stores may omit the size (reported as 0); validate only a reported nonzero size. + let reported_size = metadata.content_length(); + if reported_size != 0 && reported_size != self.bytes_written { + return Err(Error::new( + ErrorKind::Unexpected, + format!( + "Wrote {} bytes but storage reports {reported_size}", + self.bytes_written + ), + )); + } + + Ok(FileMetadata { + size: self.bytes_written, + }) } } @@ -695,6 +725,38 @@ mod tests { assert_eq!(op.info().scheme().to_string(), "memory"); } + #[cfg(feature = "opendal-memory")] + #[tokio::test] + async fn test_writer_close_returns_stored_size() { + use iceberg::encryption::{EncryptedOutputFile, StandardKeyMetadata}; + + // Note: the memory service does report a content length, so this only pins the happy + // path. The counter in `OpenDalWriter` is what covers services that don't, such as S3. + let storage = Arc::new(OpenDalStorage::Memory(default_memory_operator())); + let path = "memory:///stored-size"; + for plaintext in [ + Bytes::new(), + Bytes::from_static(b"test data"), + Bytes::from(vec![7; 3 * 1024]), + ] { + let mut writer = storage.writer(path).await.unwrap(); + for chunk in plaintext.chunks(1024) { + writer.write(Bytes::copy_from_slice(chunk)).await.unwrap(); + } + let metadata = writer.close().await.unwrap(); + assert_eq!(metadata.size, plaintext.len() as u64); + assert_eq!(metadata.size, storage.metadata(path).await.unwrap().size); + + let output = EncryptedOutputFile::new( + OutputFile::new(storage.clone(), path.to_string()), + StandardKeyMetadata::try_new(b"0123456789abcdef").unwrap(), + ); + let metadata = output.write(plaintext.clone()).await.unwrap(); + assert!(metadata.size > plaintext.len() as u64); + assert_eq!(metadata.size, storage.metadata(path).await.unwrap().size); + } + } + #[cfg(feature = "opendal-memory")] #[test] fn test_relativize_path_memory() {
