You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iceberg.apache.org by sc...@apache.org on 2026/10/09 22:24:29 UTC
(iceberg-rust) branch 0.11.x updated: 0.11.x backport: fix(encryption) [18/N] wire in ags1 file length for tamper proofing #3236 (#3270)
This is an automated email from the ASF dual-hosted git repository.
CTTY pushed a commit to branch 0.11.x
in repository https://gitbox.apache.org/repos/asf/iceberg-rust.git
The following commit(s) were added to refs/heads/0.11.x by this push:
new f98cdf3ad 0.11.x backport: fix(encryption) [18/N] wire in ags1 file length for tamper proofing #3236 (#3270)
f98cdf3ad is described below
commit f98cdf3ad9f0d325ca129157fe1f65aac55a1a02
Author: Xander <za...@googlemail.com>
AuthorDate: Fri Oct 9 23:24:23 2026 +0100
0.11.x backport: fix(encryption) [18/N] wire in ags1 file length for tamper proofing #3236 (#3270)
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
(cherry picked from commit c9a25b23b7645f3af0e9dfd7e85d8cf646901c55)
---
CHANGELOG.md | 6 +
crates/iceberg/public-api.txt | 16 +-
.../src/arrow/caching_delete_file_loader.rs | 7 +-
crates/iceberg/src/encryption/io.rs | 230 +++++++++++++++++++--
crates/iceberg/src/encryption/key_metadata.rs | 4 +-
crates/iceberg/src/encryption/manager.rs | 8 +-
crates/iceberg/src/encryption/stream.rs | 164 +++++++++++----
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, 575 insertions(+), 179 deletions(-)
diff --git a/CHANGELOG.md b/CHANGELOG.md
index 63052b44f..28571832e 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -40,6 +40,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
* feat(parquet): wire in additional parquet writer settings by @xanderbailey in https://github.com/apache/iceberg-rust/pull/2887
* fix(gcs): Fix GCS host property by @xanderbailey in https://github.com/apache/iceberg-rust/pull/2965
* refactor(scan): extract reusable scan functions by @xanderbailey in https://github.com/apache/iceberg-rust/pull/3093
+ * fix(encryption) [18/N] wire in ags1 file length for tamper proofing by @xanderbailey in https://github.com/apache/iceberg-rust/pull/3236
+ * `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.
* **Compatibility:**
* chore: Update MSRV to Rust 1.95 by @dannycjones in https://github.com/apache/iceberg-rust/pull/3083
@@ -230,6 +235,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
* fix: break dev-dependency cycle that blocks publishing iceberg-property-macro by @kevinjqliu in https://github.com/apache/iceberg-rust/pull/3184
* ci: check new crates are reserved on crates.io before merge by @kevinjqliu in https://github.com/apache/iceberg-rust/pull/3185
* ci: pull MinIO images from Quay by @smaheshwar-pltr in https://github.com/apache/iceberg-rust/pull/3202
+* fix(encryption) [18/N] wire in ags1 file length for tamper proofing by @xanderbailey in https://github.com/apache/iceberg-rust/pull/3236
## [v0.10.1] - 2026-07-28
diff --git a/crates/iceberg/public-api.txt b/crates/iceberg/public-api.txt
index 57edfd642..bfe2d2429 100644
--- a/crates/iceberg/public-api.txt
+++ b/crates/iceberg/public-api.txt
@@ -272,7 +272,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
@@ -280,7 +280,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>>
@@ -290,10 +290,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
@@ -1045,10 +1045,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
@@ -1266,7 +1266,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
@@ -2133,7 +2133,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 905fa86d8..5b12491e6 100644
--- a/crates/iceberg/src/arrow/caching_delete_file_loader.rs
+++ b/crates/iceberg/src/arrow/caching_delete_file_loader.rs
@@ -1647,13 +1647,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..200162d06 100644
--- a/crates/iceberg/src/encryption/io.rs
+++ b/crates/iceberg/src/encryption/io.rs
@@ -23,9 +23,9 @@ use bytes::Bytes;
use super::crypto::AesGcmCipher;
use super::key_metadata::StandardKeyMetadata;
-use super::stream::{AesGcmFileRead, AesGcmFileWrite};
-use crate::Result;
+use super::stream::{AesGcmFileRead, AesGcmFileWrite, MIN_STREAM_LENGTH};
use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile};
+use crate::{Error, ErrorKind, Result};
/// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`].
///
@@ -54,12 +54,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 +66,38 @@ 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(|| {
+ Error::new(
+ ErrorKind::DataInvalid,
+ "AGS1 key metadata is missing the encrypted file length",
+ )
+ })?;
+ if length < u64::from(MIN_STREAM_LENGTH) {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!("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 +134,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 +159,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 +193,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 +212,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 +229,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 +243,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 1169698e2..2e754e4ac 100644
--- a/crates/iceberg/src/encryption/key_metadata.rs
+++ b/crates/iceberg/src/encryption/key_metadata.rs
@@ -74,7 +74,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
@@ -90,7 +90,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 f54ef464d..5f26d080d 100644
--- a/crates/iceberg/src/encryption/manager.rs
+++ b/crates/iceberg/src/encryption/manager.rs
@@ -688,12 +688,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 7971a3df5..745914cb0 100644
--- a/crates/iceberg/src/encryption/stream.rs
+++ b/crates/iceberg/src/encryption/stream.rs
@@ -49,7 +49,7 @@ use std::sync::Arc;
use bytes::{Bytes, BytesMut};
use super::AesGcmCipher;
-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`.
@@ -71,8 +71,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.
///
@@ -146,20 +145,17 @@ impl AesGcmFileRead {
aad_prefix: Box<[u8]>,
encrypted_file_length: u64,
) -> Result<Self> {
+ if encrypted_file_length < u64::from(MIN_STREAM_LENGTH) {
+ return Err(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "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;
@@ -288,7 +284,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());
}
@@ -312,12 +310,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);
@@ -325,6 +319,16 @@ 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(Error::new(
+ ErrorKind::DataInvalid,
+ format!(
+ "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;
@@ -502,7 +506,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,
@@ -640,6 +644,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";
@@ -956,15 +1023,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 ---
@@ -990,8 +1063,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",
+ ))
}
}
@@ -1002,8 +1081,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,
+ })
}
}
@@ -1017,9 +1098,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]
@@ -1239,10 +1326,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 836ea74d4..f744e9b94 100644
--- a/crates/iceberg/src/io/storage/local_fs.rs
+++ b/crates/iceberg/src/io/storage/local_fs.rs
@@ -268,12 +268,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,
+ }
}
}
@@ -291,11 +295,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()
@@ -304,7 +309,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,
+ })
}
}
@@ -483,10 +490,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]
@@ -513,7 +521,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 dc9ca10d6..929d864a7 100644
--- a/crates/iceberg/src/io/storage/memory.rs
+++ b/crates/iceberg/src/io/storage/memory.rs
@@ -325,7 +325,7 @@ impl FileWrite for MemoryFileWrite {
Ok(())
}
- async fn close(&mut self) -> Result<()> {
+ async fn close(&mut self) -> Result<FileMetadata> {
if self.closed {
return Err(Error::new(ErrorKind::DataInvalid, "File already closed"));
}
@@ -337,12 +337,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 })
}
}
@@ -496,10 +497,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]
@@ -522,7 +524,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 725669295..0922d3ec1 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;
-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::{
@@ -43,14 +41,38 @@ use crate::{Error, ErrorKind};
/// 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,
}
@@ -65,10 +87,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,
}
@@ -84,12 +105,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,
})
@@ -104,14 +123,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.
@@ -124,14 +136,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.
@@ -143,14 +148,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.
@@ -163,10 +161,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.
@@ -183,20 +180,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>,
@@ -211,8 +201,6 @@ pub struct ManifestWriter {
min_seq_num: Option<i64>,
- key_metadata: Option<Vec<u8>>,
-
manifest_entries: Vec<ManifestEntry>,
metadata: ManifestMetadata,
@@ -221,15 +209,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,
@@ -240,7 +227,6 @@ impl ManifestWriter {
deleted_rows: 0,
first_row_id,
min_seq_num: None,
- key_metadata,
manifest_entries: Vec::new(),
metadata,
}
@@ -508,14 +494,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
@@ -530,7 +517,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,
})
}
@@ -836,9 +823,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 567e78e3a..3c7d5d9c1 100644
--- a/crates/iceberg/src/spec/manifest_list/manifest_file.rs
+++ b/crates/iceberg/src/spec/manifest_list/manifest_file.rs
@@ -290,12 +290,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);
@@ -322,7 +325,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)
@@ -347,7 +351,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 710471780..6850fe8d3 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;
-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
@@ -616,7 +615,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;
@@ -651,7 +649,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!(
@@ -659,6 +657,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 1f09a79a0..e4b4f370f 100644
--- a/crates/iceberg/src/transaction/snapshot.rs
+++ b/crates/iceberg/src/transaction/snapshot.rs
@@ -451,13 +451,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),
};
@@ -492,7 +492,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 55defaa5d..8f05723a3 100644
--- a/crates/iceberg/src/writer/file_writer/parquet_writer.rs
+++ b/crates/iceberg/src/writer/file_writer/parquet_writer.rs
@@ -743,9 +743,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() {