blackmwk commented on code in PR #2383: URL: https://github.com/apache/iceberg-rust/pull/2383#discussion_r3193971642
########## crates/iceberg/src/encryption/encrypted_io.rs: ########## Review Comment: For simplicity should we just rename this module to `io`? ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## @@ -0,0 +1,706 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encryption manager for file-level encryption and two-layer envelope key management. +//! +//! [`EncryptionManager`] provides file-level `decrypt` / `encrypt` +//! operations matching Java's `org.apache.iceberg.encryption.EncryptionManager`, +//! using envelope encryption: +//! - A master key (in KMS) wraps a Key Encryption Key (KEK) +//! - The KEK wraps Data Encryption Keys (DEKs) locally + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use std::time::Duration; + +use aes_gcm::aead::OsRng; +use aes_gcm::aead::rand_core::RngCore; +use chrono::Utc; +use uuid::Uuid; + +const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000; + +use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes}; +use super::encrypted_io::{EncryptedInputFile, EncryptedOutputFile}; +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use super::key_metadata::StandardKeyMetadata; +use super::kms::KeyManagementClient; +use crate::io::{InputFile, OutputFile}; +use crate::spec::EncryptedKey; +use crate::{Error, ErrorKind, Result}; + +/// Property key for the KEK creation timestamp (milliseconds since epoch). +/// Matches Java's `StandardEncryptionManager.KEY_TIMESTAMP`. +pub const KEK_CREATED_AT_PROPERTY: &str = "KEY_TIMESTAMP"; + +/// Default KEK lifespan in days, per NIST SP 800-57. +const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730; + +/// Default cache TTL for unwrapped KEKs. +const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600); + +/// Default AAD prefix length in bytes. +/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`. +const AAD_PREFIX_LENGTH: usize = 16; + +/// File-level encryption manager using two-layer envelope encryption. +/// +/// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls. +#[derive(typed_builder::TypedBuilder)] +pub struct EncryptionManager { + kms_client: Arc<dyn KeyManagementClient>, + #[builder( + default = moka::future::Cache::builder().time_to_live(DEFAULT_CACHE_TTL).build(), + setter(skip) + )] + kek_cache: moka::future::Cache<String, SensitiveBytes>, Review Comment: Please don't use fully qualified name here, use them in imports as much as possible. ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## Review Comment: Should we just rename this file to `manager.rs`? ########## crates/iceberg/src/encryption/encrypted_io.rs: ########## @@ -0,0 +1,154 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encrypted file wrappers for InputFile / OutputFile. + +use std::sync::Arc; + +use bytes::Bytes; + +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; + +/// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`]. +/// +/// Transparently decrypts on read. +pub struct EncryptedInputFile { + inner: InputFile, + decryptor: Arc<AesGcmFileDecryptor>, Review Comment: Another question is do we really need `AesGcmFileDecryptor`/`AesGcmFileEncryptor`? ########## crates/iceberg/src/encryption/encrypted_io.rs: ########## @@ -0,0 +1,154 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encrypted file wrappers for InputFile / OutputFile. + +use std::sync::Arc; + +use bytes::Bytes; + +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; + +/// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`]. +/// +/// Transparently decrypts on read. +pub struct EncryptedInputFile { + inner: InputFile, + decryptor: Arc<AesGcmFileDecryptor>, Review Comment: I think we should just store `StandardKeyMetadata` here? ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## @@ -0,0 +1,706 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encryption manager for file-level encryption and two-layer envelope key management. +//! +//! [`EncryptionManager`] provides file-level `decrypt` / `encrypt` +//! operations matching Java's `org.apache.iceberg.encryption.EncryptionManager`, +//! using envelope encryption: +//! - A master key (in KMS) wraps a Key Encryption Key (KEK) +//! - The KEK wraps Data Encryption Keys (DEKs) locally + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use std::time::Duration; + +use aes_gcm::aead::OsRng; +use aes_gcm::aead::rand_core::RngCore; +use chrono::Utc; +use uuid::Uuid; + +const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000; + +use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes}; +use super::encrypted_io::{EncryptedInputFile, EncryptedOutputFile}; +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use super::key_metadata::StandardKeyMetadata; +use super::kms::KeyManagementClient; +use crate::io::{InputFile, OutputFile}; +use crate::spec::EncryptedKey; +use crate::{Error, ErrorKind, Result}; + +/// Property key for the KEK creation timestamp (milliseconds since epoch). +/// Matches Java's `StandardEncryptionManager.KEY_TIMESTAMP`. +pub const KEK_CREATED_AT_PROPERTY: &str = "KEY_TIMESTAMP"; + +/// Default KEK lifespan in days, per NIST SP 800-57. +const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730; + +/// Default cache TTL for unwrapped KEKs. +const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600); + +/// Default AAD prefix length in bytes. +/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`. +const AAD_PREFIX_LENGTH: usize = 16; + +/// File-level encryption manager using two-layer envelope encryption. +/// +/// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls. +#[derive(typed_builder::TypedBuilder)] +pub struct EncryptionManager { + kms_client: Arc<dyn KeyManagementClient>, + #[builder( + default = moka::future::Cache::builder().time_to_live(DEFAULT_CACHE_TTL).build(), + setter(skip) + )] + kek_cache: moka::future::Cache<String, SensitiveBytes>, + /// AES key size for DEK generation. Defaults to 128-bit. + #[builder(default = AesKeySize::default())] + key_size: AesKeySize, + /// Master key ID from table property `encryption.key-id`. + #[builder(setter(into))] + table_key_id: String, + /// All encryption keys from table metadata (KEKs and wrapped key metadata entries). + #[builder(default)] + encryption_keys: HashMap<String, EncryptedKey>, +} + +impl fmt::Debug for EncryptionManager { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("EncryptionManager") + .field("key_size", &self.key_size) + .field("table_key_id", &self.table_key_id) + .finish_non_exhaustive() + } +} + +impl EncryptionManager { + /// Encrypt a file with AGS1 stream encryption. + /// + /// Returns an [`EncryptedOutputFile`] that transparently encrypts on + /// write, along with key metadata for later decryption. + pub fn encrypt(&self, raw_output: OutputFile) -> Result<EncryptedOutputFile> { + let dek = SecureKey::generate(self.key_size); + let aad_prefix = Self::generate_aad_prefix(); + + let key_metadata_bytes = StandardKeyMetadata::new(dek.as_bytes()) + .with_aad_prefix(&aad_prefix) + .encode()?; + + let encryptor = Arc::new(AesGcmFileEncryptor::new(dek.as_bytes(), aad_prefix)?); + + Ok(EncryptedOutputFile::new( + raw_output, + key_metadata_bytes, + encryptor, + )) + } + + /// Decrypt an encrypted input file, returning an [`EncryptedInputFile`] + /// that transparently decrypts on read. + pub fn decrypt(&self, input: InputFile, key_metadata: &[u8]) -> Result<EncryptedInputFile> { + let metadata = StandardKeyMetadata::decode(key_metadata)?; + + let decryptor = Arc::new(AesGcmFileDecryptor::new( + metadata.encryption_key().as_bytes(), + metadata.aad_prefix().unwrap_or_default(), + )?); + + Ok(EncryptedInputFile::new(input, decryptor)) + } + + /// Generate key material for Parquet Modular Encryption (PME). + /// + /// Returns a [`StandardKeyMetadata`] containing a fresh DEK and AAD prefix. + /// The caller should pass this to the Parquet writer to configure + /// `FileEncryptionProperties`, and serialize it for storage in the manifest. + pub fn generate_native_key_metadata(&self) -> Result<StandardKeyMetadata> { + let dek = SecureKey::generate(self.key_size); + let aad_prefix = Self::generate_aad_prefix(); + Ok(StandardKeyMetadata::new(dek.as_bytes()).with_aad_prefix(&aad_prefix)) + } + + /// Wrap key metadata bytes with a KEK for storage in table metadata. + /// + /// Returns `(wrapped_entry, optional_new_kek)`. The wrapped entry + /// contains the key metadata encrypted by the KEK, and should be stored + /// in `TableMetadata.encryption_keys`. The optional second element is a + /// newly created KEK — present only when no active KEK existed (first + /// write) or the existing KEK expired (rotation). When `Some`, the + /// caller must also persist this KEK in table metadata so that future + /// `unwrap_key_metadata` calls can find it. + pub async fn wrap_key_metadata( + &self, + key_metadata: &[u8], + ) -> Result<(EncryptedKey, Option<EncryptedKey>)> { + let (kek, new_kek) = match self.find_active_kek(&self.encryption_keys) { + Some(existing) => (existing.clone(), None), + None => { + let created = self.create_kek().await?; + let cloned = created.clone(); + (created, Some(cloned)) + } + }; + + let kek_bytes = self.unwrap_kek(&kek).await?; + + // Use the KEK timestamp as AAD to prevent timestamp tampering attacks. + let aad = Self::kek_timestamp_aad(&kek)?; + let wrapped_metadata = self.wrap_dek_with_kek(key_metadata, &kek_bytes, Some(aad))?; + + let wrapped_key = EncryptedKey::builder() + .key_id(Uuid::new_v4().to_string()) + .encrypted_key_metadata(wrapped_metadata) + .encrypted_by_id(kek.key_id()) + .build(); + + Ok((wrapped_key, new_kek)) + } + + /// Unwrap key metadata that was KEK-wrapped and stored in table metadata. + /// + /// Given an `EncryptedKey` entry (from a manifest list or snapshot) and + /// the full map of encryption keys from `TableMetadata`, returns the + /// unwrapped key metadata bytes (e.g. serialized `StandardKeyMetadata`). + pub async fn unwrap_key_metadata( Review Comment: Ditto. ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## @@ -0,0 +1,706 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encryption manager for file-level encryption and two-layer envelope key management. +//! +//! [`EncryptionManager`] provides file-level `decrypt` / `encrypt` +//! operations matching Java's `org.apache.iceberg.encryption.EncryptionManager`, +//! using envelope encryption: +//! - A master key (in KMS) wraps a Key Encryption Key (KEK) +//! - The KEK wraps Data Encryption Keys (DEKs) locally + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use std::time::Duration; + +use aes_gcm::aead::OsRng; +use aes_gcm::aead::rand_core::RngCore; +use chrono::Utc; +use uuid::Uuid; + +const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000; + +use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes}; +use super::encrypted_io::{EncryptedInputFile, EncryptedOutputFile}; +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use super::key_metadata::StandardKeyMetadata; +use super::kms::KeyManagementClient; +use crate::io::{InputFile, OutputFile}; +use crate::spec::EncryptedKey; +use crate::{Error, ErrorKind, Result}; + +/// Property key for the KEK creation timestamp (milliseconds since epoch). +/// Matches Java's `StandardEncryptionManager.KEY_TIMESTAMP`. +pub const KEK_CREATED_AT_PROPERTY: &str = "KEY_TIMESTAMP"; + +/// Default KEK lifespan in days, per NIST SP 800-57. +const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730; + +/// Default cache TTL for unwrapped KEKs. +const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600); + +/// Default AAD prefix length in bytes. +/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`. +const AAD_PREFIX_LENGTH: usize = 16; + +/// File-level encryption manager using two-layer envelope encryption. +/// +/// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls. +#[derive(typed_builder::TypedBuilder)] +pub struct EncryptionManager { + kms_client: Arc<dyn KeyManagementClient>, + #[builder( + default = moka::future::Cache::builder().time_to_live(DEFAULT_CACHE_TTL).build(), + setter(skip) + )] + kek_cache: moka::future::Cache<String, SensitiveBytes>, + /// AES key size for DEK generation. Defaults to 128-bit. + #[builder(default = AesKeySize::default())] + key_size: AesKeySize, + /// Master key ID from table property `encryption.key-id`. + #[builder(setter(into))] + table_key_id: String, + /// All encryption keys from table metadata (KEKs and wrapped key metadata entries). + #[builder(default)] + encryption_keys: HashMap<String, EncryptedKey>, +} + +impl fmt::Debug for EncryptionManager { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("EncryptionManager") + .field("key_size", &self.key_size) + .field("table_key_id", &self.table_key_id) + .finish_non_exhaustive() + } +} + +impl EncryptionManager { + /// Encrypt a file with AGS1 stream encryption. + /// + /// Returns an [`EncryptedOutputFile`] that transparently encrypts on + /// write, along with key metadata for later decryption. + pub fn encrypt(&self, raw_output: OutputFile) -> Result<EncryptedOutputFile> { + let dek = SecureKey::generate(self.key_size); + let aad_prefix = Self::generate_aad_prefix(); + + let key_metadata_bytes = StandardKeyMetadata::new(dek.as_bytes()) + .with_aad_prefix(&aad_prefix) + .encode()?; + + let encryptor = Arc::new(AesGcmFileEncryptor::new(dek.as_bytes(), aad_prefix)?); + + Ok(EncryptedOutputFile::new( + raw_output, + key_metadata_bytes, + encryptor, + )) + } + + /// Decrypt an encrypted input file, returning an [`EncryptedInputFile`] + /// that transparently decrypts on read. + pub fn decrypt(&self, input: InputFile, key_metadata: &[u8]) -> Result<EncryptedInputFile> { Review Comment: ```suggestion pub fn decrypt(&self, input: EncryptedInputeFile) -> Result<EncryptedInputFile> { ``` ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## @@ -0,0 +1,706 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encryption manager for file-level encryption and two-layer envelope key management. +//! +//! [`EncryptionManager`] provides file-level `decrypt` / `encrypt` +//! operations matching Java's `org.apache.iceberg.encryption.EncryptionManager`, +//! using envelope encryption: +//! - A master key (in KMS) wraps a Key Encryption Key (KEK) +//! - The KEK wraps Data Encryption Keys (DEKs) locally + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use std::time::Duration; + +use aes_gcm::aead::OsRng; +use aes_gcm::aead::rand_core::RngCore; +use chrono::Utc; +use uuid::Uuid; + +const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000; + +use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes}; +use super::encrypted_io::{EncryptedInputFile, EncryptedOutputFile}; +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use super::key_metadata::StandardKeyMetadata; +use super::kms::KeyManagementClient; +use crate::io::{InputFile, OutputFile}; +use crate::spec::EncryptedKey; +use crate::{Error, ErrorKind, Result}; + +/// Property key for the KEK creation timestamp (milliseconds since epoch). +/// Matches Java's `StandardEncryptionManager.KEY_TIMESTAMP`. +pub const KEK_CREATED_AT_PROPERTY: &str = "KEY_TIMESTAMP"; + +/// Default KEK lifespan in days, per NIST SP 800-57. +const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730; + +/// Default cache TTL for unwrapped KEKs. +const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600); + +/// Default AAD prefix length in bytes. +/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`. +const AAD_PREFIX_LENGTH: usize = 16; + +/// File-level encryption manager using two-layer envelope encryption. +/// +/// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls. +#[derive(typed_builder::TypedBuilder)] +pub struct EncryptionManager { + kms_client: Arc<dyn KeyManagementClient>, + #[builder( + default = moka::future::Cache::builder().time_to_live(DEFAULT_CACHE_TTL).build(), + setter(skip) + )] + kek_cache: moka::future::Cache<String, SensitiveBytes>, + /// AES key size for DEK generation. Defaults to 128-bit. + #[builder(default = AesKeySize::default())] + key_size: AesKeySize, + /// Master key ID from table property `encryption.key-id`. + #[builder(setter(into))] + table_key_id: String, + /// All encryption keys from table metadata (KEKs and wrapped key metadata entries). + #[builder(default)] + encryption_keys: HashMap<String, EncryptedKey>, Review Comment: nit: For collections like this, we could use mutations in TypedBuilder so that user facing api is like `add_encryption_key(EncryptedKey)` . ########## crates/iceberg/src/encryption/encrypted_io.rs: ########## @@ -0,0 +1,154 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encrypted file wrappers for InputFile / OutputFile. + +use std::sync::Arc; + +use bytes::Bytes; + +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; + +/// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`]. +/// +/// Transparently decrypts on read. +pub struct EncryptedInputFile { + inner: InputFile, + decryptor: Arc<AesGcmFileDecryptor>, +} + +impl EncryptedInputFile { + /// Creates a new encrypted input file. + pub fn new(inner: InputFile, decryptor: Arc<AesGcmFileDecryptor>) -> Self { + Self { inner, decryptor } + } + + /// Absolute path of the file. + pub fn location(&self) -> &str { + self.inner.location() + } + + /// Check if file exists. + pub async fn exists(&self) -> crate::Result<bool> { + self.inner.exists().await + } + + /// Fetch and returns metadata of file. + /// + /// The returned size is the **plaintext** size. + pub async fn metadata(&self) -> crate::Result<FileMetadata> { + let raw_meta = self.inner.metadata().await?; + let plaintext_size = self.decryptor.plaintext_length(raw_meta.size)?; + Ok(FileMetadata { + size: plaintext_size, + }) + } + + /// Read and returns whole content of file (decrypted plaintext). + pub async fn read(&self) -> crate::Result<Bytes> { + let meta = self.metadata().await?; + 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) -> crate::Result<Box<dyn FileRead>> { + let raw_meta = self.inner.metadata().await?; + let raw_reader = self.inner.reader().await?; + self.decryptor.wrap_reader(raw_reader, raw_meta.size) + } + + /// Consumes self and returns the underlying plain input file. + pub fn into_inner(self) -> InputFile { + self.inner + } +} + +impl std::fmt::Debug for EncryptedInputFile { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("EncryptedInputFile") + .field("path", &self.inner.location()) + .finish_non_exhaustive() + } +} + +/// An AGS1 stream-encrypted output file wrapping a plain [`OutputFile`]. +/// +/// Transparently encrypts on write. +pub struct EncryptedOutputFile { + inner: OutputFile, + key_metadata: Box<[u8]>, + encryptor: Arc<AesGcmFileEncryptor>, +} + +impl EncryptedOutputFile { + /// Creates a new encrypted output file. + pub fn new( + inner: OutputFile, + key_metadata: Box<[u8]>, + encryptor: Arc<AesGcmFileEncryptor>, Review Comment: I think for constructor, we only need a data key length? ########## crates/iceberg/src/encryption/encrypted_io.rs: ########## @@ -0,0 +1,154 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encrypted file wrappers for InputFile / OutputFile. + +use std::sync::Arc; + +use bytes::Bytes; + +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use crate::io::{FileMetadata, FileRead, FileWrite, InputFile, OutputFile}; + +/// An AGS1 stream-encrypted input file wrapping a plain [`InputFile`]. +/// +/// Transparently decrypts on read. +pub struct EncryptedInputFile { + inner: InputFile, + decryptor: Arc<AesGcmFileDecryptor>, +} + +impl EncryptedInputFile { + /// Creates a new encrypted input file. + pub fn new(inner: InputFile, decryptor: Arc<AesGcmFileDecryptor>) -> Self { + Self { inner, decryptor } + } + + /// Absolute path of the file. + pub fn location(&self) -> &str { + self.inner.location() + } + + /// Check if file exists. + pub async fn exists(&self) -> crate::Result<bool> { + self.inner.exists().await + } + + /// Fetch and returns metadata of file. + /// + /// The returned size is the **plaintext** size. + pub async fn metadata(&self) -> crate::Result<FileMetadata> { + let raw_meta = self.inner.metadata().await?; + let plaintext_size = self.decryptor.plaintext_length(raw_meta.size)?; + Ok(FileMetadata { + size: plaintext_size, + }) + } + + /// Read and returns whole content of file (decrypted plaintext). + pub async fn read(&self) -> crate::Result<Bytes> { + let meta = self.metadata().await?; + 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) -> crate::Result<Box<dyn FileRead>> { + let raw_meta = self.inner.metadata().await?; + let raw_reader = self.inner.reader().await?; + self.decryptor.wrap_reader(raw_reader, raw_meta.size) + } + + /// Consumes self and returns the underlying plain input file. + pub fn into_inner(self) -> InputFile { + self.inner + } +} + +impl std::fmt::Debug for EncryptedInputFile { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("EncryptedInputFile") + .field("path", &self.inner.location()) + .finish_non_exhaustive() + } +} + +/// An AGS1 stream-encrypted output file wrapping a plain [`OutputFile`]. +/// +/// Transparently encrypts on write. +pub struct EncryptedOutputFile { + inner: OutputFile, + key_metadata: Box<[u8]>, + encryptor: Arc<AesGcmFileEncryptor>, +} + +impl EncryptedOutputFile { + /// Creates a new encrypted output file. + pub fn new( + inner: OutputFile, + key_metadata: Box<[u8]>, + encryptor: Arc<AesGcmFileEncryptor>, + ) -> Self { + Self { + inner, + key_metadata, + encryptor, + } + } + + /// Returns the key metadata bytes (for storage in manifest/data files). + pub fn key_metadata(&self) -> &[u8] { Review Comment: We should return `StandardKeyMetadata` ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## @@ -0,0 +1,706 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encryption manager for file-level encryption and two-layer envelope key management. +//! +//! [`EncryptionManager`] provides file-level `decrypt` / `encrypt` +//! operations matching Java's `org.apache.iceberg.encryption.EncryptionManager`, +//! using envelope encryption: +//! - A master key (in KMS) wraps a Key Encryption Key (KEK) +//! - The KEK wraps Data Encryption Keys (DEKs) locally + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use std::time::Duration; + +use aes_gcm::aead::OsRng; +use aes_gcm::aead::rand_core::RngCore; +use chrono::Utc; +use uuid::Uuid; + +const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000; + +use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes}; +use super::encrypted_io::{EncryptedInputFile, EncryptedOutputFile}; +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use super::key_metadata::StandardKeyMetadata; +use super::kms::KeyManagementClient; +use crate::io::{InputFile, OutputFile}; +use crate::spec::EncryptedKey; +use crate::{Error, ErrorKind, Result}; + +/// Property key for the KEK creation timestamp (milliseconds since epoch). +/// Matches Java's `StandardEncryptionManager.KEY_TIMESTAMP`. +pub const KEK_CREATED_AT_PROPERTY: &str = "KEY_TIMESTAMP"; + +/// Default KEK lifespan in days, per NIST SP 800-57. +const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730; + +/// Default cache TTL for unwrapped KEKs. +const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600); + +/// Default AAD prefix length in bytes. +/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`. +const AAD_PREFIX_LENGTH: usize = 16; + +/// File-level encryption manager using two-layer envelope encryption. +/// +/// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls. +#[derive(typed_builder::TypedBuilder)] +pub struct EncryptionManager { + kms_client: Arc<dyn KeyManagementClient>, + #[builder( + default = moka::future::Cache::builder().time_to_live(DEFAULT_CACHE_TTL).build(), + setter(skip) + )] + kek_cache: moka::future::Cache<String, SensitiveBytes>, + /// AES key size for DEK generation. Defaults to 128-bit. + #[builder(default = AesKeySize::default())] + key_size: AesKeySize, + /// Master key ID from table property `encryption.key-id`. + #[builder(setter(into))] + table_key_id: String, + /// All encryption keys from table metadata (KEKs and wrapped key metadata entries). + #[builder(default)] + encryption_keys: HashMap<String, EncryptedKey>, +} + +impl fmt::Debug for EncryptionManager { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("EncryptionManager") + .field("key_size", &self.key_size) + .field("table_key_id", &self.table_key_id) + .finish_non_exhaustive() + } +} + +impl EncryptionManager { + /// Encrypt a file with AGS1 stream encryption. + /// + /// Returns an [`EncryptedOutputFile`] that transparently encrypts on + /// write, along with key metadata for later decryption. + pub fn encrypt(&self, raw_output: OutputFile) -> Result<EncryptedOutputFile> { + let dek = SecureKey::generate(self.key_size); + let aad_prefix = Self::generate_aad_prefix(); + + let key_metadata_bytes = StandardKeyMetadata::new(dek.as_bytes()) + .with_aad_prefix(&aad_prefix) + .encode()?; + + let encryptor = Arc::new(AesGcmFileEncryptor::new(dek.as_bytes(), aad_prefix)?); + + Ok(EncryptedOutputFile::new( + raw_output, + key_metadata_bytes, + encryptor, + )) + } + + /// Decrypt an encrypted input file, returning an [`EncryptedInputFile`] + /// that transparently decrypts on read. + pub fn decrypt(&self, input: InputFile, key_metadata: &[u8]) -> Result<EncryptedInputFile> { + let metadata = StandardKeyMetadata::decode(key_metadata)?; + + let decryptor = Arc::new(AesGcmFileDecryptor::new( + metadata.encryption_key().as_bytes(), + metadata.aad_prefix().unwrap_or_default(), + )?); + + Ok(EncryptedInputFile::new(input, decryptor)) + } + + /// Generate key material for Parquet Modular Encryption (PME). + /// + /// Returns a [`StandardKeyMetadata`] containing a fresh DEK and AAD prefix. + /// The caller should pass this to the Parquet writer to configure + /// `FileEncryptionProperties`, and serialize it for storage in the manifest. + pub fn generate_native_key_metadata(&self) -> Result<StandardKeyMetadata> { Review Comment: Let's make it mod private first/ ########## crates/iceberg/src/encryption/encryption_manager.rs: ########## @@ -0,0 +1,706 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Encryption manager for file-level encryption and two-layer envelope key management. +//! +//! [`EncryptionManager`] provides file-level `decrypt` / `encrypt` +//! operations matching Java's `org.apache.iceberg.encryption.EncryptionManager`, +//! using envelope encryption: +//! - A master key (in KMS) wraps a Key Encryption Key (KEK) +//! - The KEK wraps Data Encryption Keys (DEKs) locally + +use std::collections::HashMap; +use std::fmt; +use std::sync::Arc; +use std::time::Duration; + +use aes_gcm::aead::OsRng; +use aes_gcm::aead::rand_core::RngCore; +use chrono::Utc; +use uuid::Uuid; + +const MILLIS_IN_DAY: i64 = 24 * 60 * 60 * 1000; + +use super::crypto::{AesGcmCipher, AesKeySize, SecureKey, SensitiveBytes}; +use super::encrypted_io::{EncryptedInputFile, EncryptedOutputFile}; +use super::file_decryptor::AesGcmFileDecryptor; +use super::file_encryptor::AesGcmFileEncryptor; +use super::key_metadata::StandardKeyMetadata; +use super::kms::KeyManagementClient; +use crate::io::{InputFile, OutputFile}; +use crate::spec::EncryptedKey; +use crate::{Error, ErrorKind, Result}; + +/// Property key for the KEK creation timestamp (milliseconds since epoch). +/// Matches Java's `StandardEncryptionManager.KEY_TIMESTAMP`. +pub const KEK_CREATED_AT_PROPERTY: &str = "KEY_TIMESTAMP"; + +/// Default KEK lifespan in days, per NIST SP 800-57. +const DEFAULT_KEK_LIFESPAN_DAYS: i64 = 730; + +/// Default cache TTL for unwrapped KEKs. +const DEFAULT_CACHE_TTL: Duration = Duration::from_secs(3600); + +/// Default AAD prefix length in bytes. +/// Matches Java's `TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT`. +const AAD_PREFIX_LENGTH: usize = 16; + +/// File-level encryption manager using two-layer envelope encryption. +/// +/// Uses an async cache for unwrapped KEK bytes to avoid repeated KMS calls. +#[derive(typed_builder::TypedBuilder)] +pub struct EncryptionManager { + kms_client: Arc<dyn KeyManagementClient>, + #[builder( + default = moka::future::Cache::builder().time_to_live(DEFAULT_CACHE_TTL).build(), + setter(skip) + )] + kek_cache: moka::future::Cache<String, SensitiveBytes>, + /// AES key size for DEK generation. Defaults to 128-bit. + #[builder(default = AesKeySize::default())] + key_size: AesKeySize, + /// Master key ID from table property `encryption.key-id`. + #[builder(setter(into))] + table_key_id: String, + /// All encryption keys from table metadata (KEKs and wrapped key metadata entries). + #[builder(default)] + encryption_keys: HashMap<String, EncryptedKey>, +} + +impl fmt::Debug for EncryptionManager { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + f.debug_struct("EncryptionManager") + .field("key_size", &self.key_size) + .field("table_key_id", &self.table_key_id) + .finish_non_exhaustive() + } +} + +impl EncryptionManager { + /// Encrypt a file with AGS1 stream encryption. + /// + /// Returns an [`EncryptedOutputFile`] that transparently encrypts on + /// write, along with key metadata for later decryption. + pub fn encrypt(&self, raw_output: OutputFile) -> Result<EncryptedOutputFile> { + let dek = SecureKey::generate(self.key_size); + let aad_prefix = Self::generate_aad_prefix(); + + let key_metadata_bytes = StandardKeyMetadata::new(dek.as_bytes()) + .with_aad_prefix(&aad_prefix) + .encode()?; + + let encryptor = Arc::new(AesGcmFileEncryptor::new(dek.as_bytes(), aad_prefix)?); + + Ok(EncryptedOutputFile::new( + raw_output, + key_metadata_bytes, + encryptor, + )) + } + + /// Decrypt an encrypted input file, returning an [`EncryptedInputFile`] + /// that transparently decrypts on read. + pub fn decrypt(&self, input: InputFile, key_metadata: &[u8]) -> Result<EncryptedInputFile> { + let metadata = StandardKeyMetadata::decode(key_metadata)?; + + let decryptor = Arc::new(AesGcmFileDecryptor::new( + metadata.encryption_key().as_bytes(), + metadata.aad_prefix().unwrap_or_default(), + )?); + + Ok(EncryptedInputFile::new(input, decryptor)) + } + + /// Generate key material for Parquet Modular Encryption (PME). + /// + /// Returns a [`StandardKeyMetadata`] containing a fresh DEK and AAD prefix. + /// The caller should pass this to the Parquet writer to configure + /// `FileEncryptionProperties`, and serialize it for storage in the manifest. + pub fn generate_native_key_metadata(&self) -> Result<StandardKeyMetadata> { + let dek = SecureKey::generate(self.key_size); + let aad_prefix = Self::generate_aad_prefix(); + Ok(StandardKeyMetadata::new(dek.as_bytes()).with_aad_prefix(&aad_prefix)) + } + + /// Wrap key metadata bytes with a KEK for storage in table metadata. + /// + /// Returns `(wrapped_entry, optional_new_kek)`. The wrapped entry + /// contains the key metadata encrypted by the KEK, and should be stored + /// in `TableMetadata.encryption_keys`. The optional second element is a + /// newly created KEK — present only when no active KEK existed (first + /// write) or the existing KEK expired (rotation). When `Some`, the + /// caller must also persist this KEK in table metadata so that future + /// `unwrap_key_metadata` calls can find it. + pub async fn wrap_key_metadata( Review Comment: It's deprecated in java, remove it. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
