Verify footer tags when reading encrypted Parquet files with plaintext footers (#7459)

* Initial commit

* Lint and clippy

* Plaintext layout is different to encrypted one

* Lint and expected memory size at decryption

* Apply suggestions from code review

Co-authored-by: Adam Reeve <adreeve@gmail.com>

* Review feedback

* Lint

* Update parquet/tests/encryption/encryption.rs

Co-authored-by: Adam Reeve <adreeve@gmail.com>

* Review feedback

---------

Co-authored-by: Adam Reeve <adreeve@gmail.com>

Author: rok

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/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/tests/encryption/encryption.rs

@@ -60,6 +60,43 @@ fn test_non_uniform_encryption_plaintext_footer() {
     verify_encryption_test_file_read(file, decryption_properties);
 }
 
+#[test]
+fn test_plaintext_footer_signature_verification() {
+    let test_data = arrow::util::test_util::parquet_test_data();
+    let path = format!("{test_data}/encrypt_columns_plaintext_footer.parquet.encrypted");
+    let file = File::open(path.clone()).unwrap();
+
+    let footer_key = "0000000000000000".as_bytes(); // 128bit/16
+    let column_1_key = "1234567890123450".as_bytes();
+    let column_2_key = "1234567890123451".as_bytes();
+
+    let decryption_properties = FileDecryptionProperties::builder(footer_key.to_vec())
+        .disable_footer_signature_verification()
+        .with_column_key("double_field", column_1_key.to_vec())
+        .with_column_key("float_field", column_2_key.to_vec())
+        .build()
+        .unwrap();
+
+    verify_encryption_test_file_read(file, decryption_properties);
+
+    let file = File::open(path.clone()).unwrap();
+
+    let decryption_properties = FileDecryptionProperties::builder(footer_key.to_vec())
+        .with_column_key("double_field", column_1_key.to_vec())
+        .with_column_key("float_field", column_2_key.to_vec())
+        .build()
+        .unwrap();
+
+    let options = ArrowReaderOptions::default()
+        .with_file_decryption_properties(decryption_properties.clone());
+    let result = ArrowReaderMetadata::load(&file, options.clone());
+    assert!(result.is_err());
+    assert!(result
+        .unwrap_err()
+        .to_string()
+        .starts_with("Parquet error: Footer signature verification failed. Computed: ["));
+}
+
 #[test]
 fn test_non_uniform_encryption_disabled_aad_storage() {
     let test_data = arrow::util::test_util::parquet_test_data();