Support writing encrypted Parquet files with plaintext footers

parquet/src/encryption/ciphers.rs

@@ -23,12 +23,14 @@ use ring::rand::{SecureRandom, SystemRandom};
 use std::fmt::Debug;
 
 const RIGHT_TWELVE: u128 = 0x0000_0000_ffff_ffff_ffff_ffff_ffff_ffff;
-const NONCE_LEN: usize = 12;
-const TAG_LEN: usize = 16;
-const SIZE_LEN: usize = 4;
+pub(crate) const NONCE_LEN: usize = 12;
+pub(crate) const TAG_LEN: usize = 16;
+pub(crate) const SIZE_LEN: usize = 4;
 
 pub(crate) trait BlockDecryptor: Debug + Send + Sync {
     fn decrypt(&self, length_and_ciphertext: &[u8], aad: &[u8]) -> Result<Vec<u8>>;
+
+    fn compute_plaintext_tag(&self, aad: &[u8], plaintext: &[u8]) -> Result<Vec<u8>>;
 }
 
 #[derive(Debug, Clone)]
@@ -63,6 +65,19 @@ impl BlockDecryptor for RingGcmBlockDecryptor {
         result.resize(result.len() - TAG_LEN, 0u8);
         Ok(result)
     }
+
+    fn compute_plaintext_tag(&self, aad: &[u8], plaintext: &[u8]) -> Result<Vec<u8>> {
+        let mut plaintext = plaintext.to_vec();
+        let nonce = &plaintext[plaintext.len() - NONCE_LEN - TAG_LEN..plaintext.len() - TAG_LEN];
+        let nonce = ring::aead::Nonce::try_assume_unique_for_key(nonce)?;
+        let plaintext_end = plaintext.len() - NONCE_LEN - TAG_LEN;
+        let tag = self.key.seal_in_place_separate_tag(
+            nonce,
+            Aad::from(aad),
+            &mut plaintext[..plaintext_end],
+        )?;
+        Ok(tag.as_ref().to_vec())
+    }
 }
 
 pub(crate) trait BlockEncryptor: Debug + Send + Sync {

parquet/src/encryption/decrypt.rs

@@ -17,8 +17,8 @@
 
 //! Configuration and utilities for decryption of files using Parquet Modular Encryption
 
-use crate::encryption::ciphers::{BlockDecryptor, RingGcmBlockDecryptor};
-use crate::encryption::modules::{create_module_aad, ModuleType};
+use crate::encryption::ciphers::{BlockDecryptor, RingGcmBlockDecryptor, TAG_LEN};
+use crate::encryption::modules::{create_footer_aad, create_module_aad, ModuleType};
 use crate::errors::{ParquetError, Result};
 use crate::file::column_crypto_metadata::ColumnCryptoMetaData;
 use std::borrow::Cow;
@@ -331,6 +331,7 @@ impl PartialEq for DecryptionKeys {
 pub struct FileDecryptionProperties {
     keys: DecryptionKeys,
     aad_prefix: Option<Vec<u8>>,
+    footer_signature_verification: bool,
 }
 
 impl FileDecryptionProperties {
@@ -351,6 +352,11 @@ impl FileDecryptionProperties {
         self.aad_prefix.as_ref()
     }
 
+    /// Returns true if footer signature verification is enabled for files with plaintext footers.
+    pub fn check_plaintext_footer_integrity(&self) -> bool {
+        self.footer_signature_verification
+    }
+
     /// Get the encryption key for decrypting a file's footer,
     /// and also column data if uniform encryption is used.
     pub fn footer_key(&self, key_metadata: Option<&[u8]>) -> Result<Cow<Vec<u8>>> {
@@ -415,6 +421,7 @@ pub struct DecryptionPropertiesBuilder {
     key_retriever: Option<Arc<dyn KeyRetriever>>,
     column_keys: HashMap<String, Vec<u8>>,
     aad_prefix: Option<Vec<u8>>,
+    footer_signature_verification: bool,
 }
 
 impl DecryptionPropertiesBuilder {
@@ -426,6 +433,7 @@ impl DecryptionPropertiesBuilder {
             key_retriever: None,
             column_keys: HashMap::default(),
             aad_prefix: None,
+            footer_signature_verification: true,
         }
     }
 
@@ -439,6 +447,7 @@ impl DecryptionPropertiesBuilder {
             key_retriever: Some(key_retriever),
             column_keys: HashMap::default(),
             aad_prefix: None,
+            footer_signature_verification: true,
         }
     }
 
@@ -464,6 +473,7 @@ impl DecryptionPropertiesBuilder {
         Ok(FileDecryptionProperties {
             keys,
             aad_prefix: self.aad_prefix,
+            footer_signature_verification: self.footer_signature_verification,
         })
     }
 
@@ -496,6 +506,13 @@ impl DecryptionPropertiesBuilder {
         }
         Ok(self)
     }
+
+    /// Disable verification of footer tags for files that use plaintext footers.
+    /// Signature verification is enabled by default.
+    pub fn disable_footer_signature_verification(mut self) -> Self {
+        self.footer_signature_verification = false;
+        self
+    }
 }
 
 #[derive(Clone, Debug)]
@@ -538,6 +555,25 @@ impl FileDecryptor {
         Ok(self.footer_decryptor.clone())
     }
 
+    /// Verify the signature of the footer
+    pub(crate) fn verify_plaintext_footer_signature(&self, plaintext_footer: &[u8]) -> Result<()> {
+        // Plaintext footer format is: [plaintext metadata, nonce, authentication tag]
+        let tag = &plaintext_footer[plaintext_footer.len() - TAG_LEN..];
+        let aad = create_footer_aad(self.file_aad())?;
+        let footer_decryptor = self.get_footer_decryptor()?;
+
+        let computed_tag = footer_decryptor.compute_plaintext_tag(&aad, plaintext_footer)?;
+
+        if computed_tag != tag {
+            return Err(general_err!(
+                "Footer signature verification failed. Computed: {:?}, Expected: {:?}",
+                computed_tag,
+                tag
+            ));
+        }
+        Ok(())
+    }
+
     pub(crate) fn get_column_data_decryptor(
         &self,
         column_name: &str,

parquet/src/encryption/encrypt.rs

@@ -17,7 +17,9 @@
 
 //! Configuration and utilities for Parquet Modular Encryption
 
-use crate::encryption::ciphers::{BlockEncryptor, RingGcmBlockEncryptor};
+use crate::encryption::ciphers::{
+    BlockEncryptor, RingGcmBlockEncryptor, NONCE_LEN, SIZE_LEN, TAG_LEN,
+};
 use crate::errors::{ParquetError, Result};
 use crate::file::column_crypto_metadata::{ColumnCryptoMetaData, EncryptionWithColumnKey};
 use crate::schema::types::{ColumnDescPtr, SchemaDescriptor};
@@ -374,6 +376,29 @@ pub(crate) fn encrypt_object<T: TSerializable, W: Write>(
     Ok(())
 }
 
+pub(crate) fn write_signed_plaintext_object<T: TSerializable, W: Write>(
+    object: &T,
+    encryptor: &mut Box<dyn BlockEncryptor>,
+    sink: &mut W,
+    module_aad: &[u8],
+) -> Result<()> {
+    let mut buffer: Vec<u8> = vec![];
+    {
+        let mut protocol = TCompactOutputProtocol::new(&mut buffer);
+        object.write_to_out_protocol(&mut protocol)?;
+    }
+    sink.write_all(&buffer)?;
+    buffer = encryptor.encrypt(buffer.as_ref(), module_aad)?;
+
+    // Format of encrypted buffer is: [ciphertext size, nonce, ciphertext, authentication tag]
+    let nonce = &buffer[SIZE_LEN..SIZE_LEN + NONCE_LEN];
+    let tag = &buffer[buffer.len() - TAG_LEN..];
+    sink.write_all(nonce)?;
+    sink.write_all(tag)?;
+
+    Ok(())
+}
+
 /// Encrypt a Thrift serializable object to a byte vector
 pub(crate) fn encrypt_object_to_vec<T: TSerializable>(
     object: &T,

parquet/src/file/metadata/mod.rs

@@ -1999,7 +1999,7 @@ mod tests {
         #[cfg(not(feature = "encryption"))]
         let base_expected_size = 2312;
         #[cfg(feature = "encryption")]
-        let base_expected_size = 2640;
+        let base_expected_size = 2648;
 
         assert_eq!(parquet_meta.memory_size(), base_expected_size);
 
@@ -2029,7 +2029,7 @@ mod tests {
         #[cfg(not(feature = "encryption"))]
         let bigger_expected_size = 2816;
         #[cfg(feature = "encryption")]
-        let bigger_expected_size = 3144;
+        let bigger_expected_size = 3152;
 
         // more set fields means more memory usage
         assert!(bigger_expected_size > base_expected_size);

parquet/src/file/metadata/reader.rs

@@ -17,14 +17,13 @@
 
 use std::{io::Read, ops::Range, sync::Arc};
 
-use bytes::Bytes;
-
 use crate::basic::ColumnOrder;
 #[cfg(feature = "encryption")]
 use crate::encryption::{
     decrypt::{FileDecryptionProperties, FileDecryptor},
     modules::create_footer_aad,
 };
+use bytes::Bytes;
 
 use crate::errors::{ParquetError, Result};
 use crate::file::metadata::{ColumnChunkMetaData, FileMetaData, ParquetMetaData, RowGroupMetaData};
@@ -967,11 +966,15 @@ impl ParquetMetaDataReader {
             file_decryption_properties,
         ) {
             // File has a plaintext footer but encryption algorithm is set
-            file_decryptor = Some(get_file_decryptor(
+            let file_decryptor_value = get_file_decryptor(
                 algo,
                 t_file_metadata.footer_signing_key_metadata.as_deref(),
                 file_decryption_properties,
-            )?);
+            )?;
+            if file_decryption_properties.check_plaintext_footer_integrity() && !encrypted_footer {
+                file_decryptor_value.verify_plaintext_footer_signature(buf)?;
+            }
+            file_decryptor = Some(file_decryptor_value);
         }
 
         let mut row_groups = Vec::new();

parquet/src/file/metadata/writer.rs

@@ -16,17 +16,21 @@
 // under the License.
 
 #[cfg(feature = "encryption")]
-use crate::encryption::encrypt::{encrypt_object, encrypt_object_to_vec, FileEncryptor};
-#[cfg(feature = "encryption")]
-use crate::encryption::modules::{create_footer_aad, create_module_aad, ModuleType};
+use crate::encryption::{
+    encrypt::{
+        encrypt_object, encrypt_object_to_vec, write_signed_plaintext_object, FileEncryptor,
+    },
+    modules::{create_footer_aad, create_module_aad, ModuleType},
+};
 #[cfg(feature = "encryption")]
 use crate::errors::ParquetError;
 use crate::errors::Result;
 use crate::file::metadata::{KeyValue, ParquetMetaData};
 use crate::file::page_index::index::Index;
 use crate::file::writer::{get_file_magic, TrackedWrite};
+use crate::format::EncryptionAlgorithm;
 #[cfg(feature = "encryption")]
-use crate::format::{AesGcmV1, ColumnCryptoMetaData, EncryptionAlgorithm};
+use crate::format::{AesGcmV1, ColumnCryptoMetaData};
 use crate::format::{ColumnChunk, ColumnIndex, FileMetaData, OffsetIndex, RowGroup};
 use crate::schema::types;
 use crate::schema::types::{SchemaDescPtr, SchemaDescriptor, TypePtr};
@@ -149,7 +153,7 @@ impl<'a, W: Write> ThriftMetadataWriter<'a, W> {
             schema: types::to_thrift(self.schema.as_ref())?,
             created_by: self.created_by.clone(),
             column_orders,
-            encryption_algorithm: None,
+            encryption_algorithm: self.object_writer.get_footer_encryption_algorithm(),
             footer_signing_key_metadata: None,
         };
 
@@ -474,6 +478,10 @@ impl MetadataObjectWriter {
     pub fn get_file_magic(&self) -> &[u8; 4] {
         get_file_magic()
     }
+
+    fn get_footer_encryption_algorithm(&self) -> Option<EncryptionAlgorithm> {
+        None
+    }
 }
 
 /// Implementations of [`MetadataObjectWriter`] methods that rely on encryption being enabled
@@ -503,6 +511,11 @@ impl MetadataObjectWriter {
                 let mut encryptor = file_encryptor.get_footer_encryptor()?;
                 encrypt_object(file_metadata, &mut encryptor, &mut sink, &aad)
             }
+            Some(file_encryptor) if file_metadata.encryption_algorithm.is_some() => {
+                let aad = create_footer_aad(file_encryptor.file_aad())?;
+                let mut encryptor = file_encryptor.get_footer_encryptor()?;
+                write_signed_plaintext_object(file_metadata, &mut encryptor, &mut sink, &aad)
+            }
             _ => Self::write_object(file_metadata, &mut sink),
         }
     }
@@ -622,25 +635,36 @@ impl MetadataObjectWriter {
         }
     }
 
-    fn file_crypto_metadata(
-        file_encryptor: &FileEncryptor,
-    ) -> Result<crate::format::FileCryptoMetaData> {
-        let properties = file_encryptor.properties();
-        let supply_aad_prefix = properties
+    fn get_footer_encryption_algorithm(&self) -> Option<EncryptionAlgorithm> {
+        if let Some(file_encryptor) = &self.file_encryptor {
+            return Some(Self::encryption_algorithm_from_encryptor(file_encryptor));
+        }
+        None
+    }
+
+    fn encryption_algorithm_from_encryptor(file_encryptor: &FileEncryptor) -> EncryptionAlgorithm {
+        let supply_aad_prefix = file_encryptor
+            .properties()
             .aad_prefix()
-            .map(|_| !properties.store_aad_prefix());
-        let encryption_algorithm = AesGcmV1 {
-            aad_prefix: if properties.store_aad_prefix() {
-                properties.aad_prefix().cloned()
-            } else {
-                None
-            },
+            .map(|_| !file_encryptor.properties().store_aad_prefix());
+        let aad_prefix = if file_encryptor.properties().store_aad_prefix() {
+            file_encryptor.properties().aad_prefix().cloned()
+        } else {
+            None
+        };
+        EncryptionAlgorithm::AESGCMV1(AesGcmV1 {
+            aad_prefix,
             aad_file_unique: Some(file_encryptor.aad_file_unique().clone()),
             supply_aad_prefix,
-        };
+        })
+    }
 
+    fn file_crypto_metadata(
+        file_encryptor: &FileEncryptor,
+    ) -> Result<crate::format::FileCryptoMetaData> {
+        let properties = file_encryptor.properties();
         Ok(crate::format::FileCryptoMetaData {
-            encryption_algorithm: EncryptionAlgorithm::AESGCMV1(encryption_algorithm),
+            encryption_algorithm: Self::encryption_algorithm_from_encryptor(file_encryptor),
             key_metadata: properties.footer_key_metadata().cloned(),
         })
     }

parquet/src/file/writer.rs

@@ -212,12 +212,6 @@ impl<W: Write + Send> SerializedFileWriter<W> {
         if let Some(file_encryption_properties) = &properties.file_encryption_properties {
             file_encryption_properties.validate_encrypted_column_names(schema_descriptor)?;
 
-            if !file_encryption_properties.encrypt_footer() {
-                return Err(general_err!(
-                    "Writing encrypted files with plaintext footers is not supported yet"
-                ));
-            }
-
             Ok(Some(Arc::new(FileEncryptor::new(
                 file_encryption_properties.clone(),
             )?)))