GH-51669: [C++][Parquet] Preserve cipher in plaintext footer metadata

cpp/src/parquet/encryption/internal_file_encryptor.cc

@@ -68,7 +68,9 @@ std::shared_ptr<Encryptor> InternalFileEncryptor::GetFooterSigningEncryptor() {
     return footer_signing_encryptor_;
   }
 
-  ParquetCipher::type algorithm = properties_->algorithm().algorithm;
+  // A plaintext footer is always signed with AES-GCM, regardless of the algorithm
+  // configured for encryption.
+  const auto algorithm = ParquetCipher::AES_GCM_V1;
   std::string footer_aad = encryption::CreateFooterAad(properties_->file_aad());
   const SecureString& footer_signing_key = properties_->footer_key();
   auto aes_encryptor = GetMetaAesEncryptor(algorithm, footer_signing_key.size());

cpp/src/parquet/encryption/write_configurations_test.cc

@@ -19,10 +19,18 @@
 
 #include <stdio.h>
 
+#include <stdexcept>
+#include <tuple>
+#include <vector>
+
 #include <arrow/io/file.h>
 
+#include "arrow/util/endian.h"
+#include "arrow/util/ubsan.h"
+
 #include "parquet/column_reader.h"
 #include "parquet/column_writer.h"
+#include "parquet/encryption/encryption_internal.h"
 #include "parquet/encryption/test_encryption_util.h"
 #include "parquet/file_reader.h"
 #include "parquet/file_writer.h"
@@ -195,6 +203,168 @@ TEST_F(TestEncryptionConfiguration, EncryptTwoColumnsAndFooterUseAES_GCM_CTR) {
                         "tmp_encrypt_columns_and_footer_ctr.parquet.encrypted"));
 }
 
+enum class AadPrefixMode { None, Stored, External };
+
+class TestGeneratedEncryptionConfiguration
+    : public ::testing::TestWithParam<
+          std::tuple<ParquetCipher::type, bool, AadPrefixMode>> {
+ public:
+  void SetUp() override {
+    PARQUET_ASSIGN_OR_THROW(temp_dir_, temp_data_dir());
+    file_path_ = temp_dir_->path().ToString() + "algorithm_metadata.parquet.encrypted";
+
+    std::map<std::string, std::shared_ptr<ColumnEncryptionProperties>> columns;
+    columns[kDoubleFieldName] =
+        ColumnEncryptionProperties::WithColumnKey(kColumnEncryptionKey1, "kc1");
+    columns[kFloatFieldName] =
+        ColumnEncryptionProperties::WithColumnKey(kColumnEncryptionKey2, "kc2");
+    columns[kBooleanFieldName] = ColumnEncryptionProperties::Unencrypted();
+
+    FileEncryptionProperties::Builder builder(kFooterEncryptionKey);
+    builder.footer_key_metadata("kf")
+        ->encrypted_columns(std::move(columns))
+        ->algorithm(std::get<0>(GetParam()));
+    if (std::get<1>(GetParam())) {
+      builder.set_plaintext_footer();
+    }
+    if (std::get<2>(GetParam()) != AadPrefixMode::None) {
+      builder.aad_prefix(kFileName);
+    }
+    if (std::get<2>(GetParam()) == AadPrefixMode::External) {
+      builder.disable_aad_prefix_storage();
+    }
+    auto properties = builder.build();
+    file_aad_ = properties->file_aad();
+    FileEncryptor().EncryptFile(file_path_, std::move(properties));
+  }
+
+ protected:
+  enum class Keys { All, MissingColumn, WrongFooter };
+  enum class AadPrefix { Match, Missing, Wrong };
+
+  std::shared_ptr<FileDecryptionProperties> MakeDecryptionProperties(
+      Keys keys = Keys::All, AadPrefix prefix = AadPrefix::Match) {
+    auto retriever = std::make_shared<StringKeyIdRetriever>();
+    retriever->PutKey("kf", keys == Keys::WrongFooter ? SecureString(kNewFooterMasterKey)
+                                                      : kFooterEncryptionKey);
+    if (keys != Keys::MissingColumn) {
+      retriever->PutKey("kc1", kColumnEncryptionKey1);
+    }
+    retriever->PutKey("kc2", kColumnEncryptionKey2);
+
+    FileDecryptionProperties::Builder builder;
+    builder.key_retriever(std::move(retriever));
+    if (std::get<2>(GetParam()) != AadPrefixMode::None && prefix != AadPrefix::Missing) {
+      builder.aad_prefix(prefix == AadPrefix::Wrong ? "incorrect-prefix" : kFileName);
+    }
+    return builder.build();
+  }
+
+  std::unique_ptr<ParquetFileReader> OpenReader(
+      std::shared_ptr<FileDecryptionProperties> decryption_properties) {
+    auto reader_properties = default_reader_properties();
+    reader_properties.file_decryption_properties(std::move(decryption_properties));
+    return ParquetFileReader::OpenFile(file_path_, false, reader_properties);
+  }
+
+  void CheckPlaintextFooterSignature() {
+    PARQUET_ASSIGN_OR_THROW(auto file, ::arrow::io::ReadableFile::Open(file_path_));
+    PARQUET_ASSIGN_OR_THROW(auto file_size, file->GetSize());
+    ASSERT_GT(file_size, 8);
+    PARQUET_ASSIGN_OR_THROW(auto trailer, file->ReadAt(file_size - 8, 8));
+    ASSERT_EQ(trailer->size(), 8);
+    const auto footer_size = ::arrow::bit_util::FromLittleEndian(
+        ::arrow::util::SafeLoadAs<uint32_t>(trailer->data()));
+    constexpr auto signature_size = kNonceLength + kGcmTagLength;
+    ASSERT_GT(footer_size, signature_size);
+    ASSERT_LE(footer_size, file_size - 8);
+    PARQUET_ASSIGN_OR_THROW(auto footer,
+                            file->ReadAt(file_size - 8 - footer_size, footer_size));
+    ASSERT_EQ(footer->size(), footer_size);
+    const auto footer_span = footer->span_as<uint8_t>();
+    const auto metadata = footer_span.first(footer_size - signature_size);
+    const auto nonce = footer_span.subspan(metadata.size(), kNonceLength);
+    const auto tag = footer_span.last(kGcmTagLength);
+    const auto aad = CreateFooterAad(file_aad_);
+
+    // Verify the on-disk tag with GCM explicitly, independently of reader selection.
+    auto encryptor = AesEncryptor::Make(ParquetCipher::AES_GCM_V1,
+                                        kFooterEncryptionKey.size(), /*metadata=*/true,
+                                        /*write_length=*/false);
+    std::vector<uint8_t> encrypted(encryptor->CiphertextLength(metadata.size()));
+    encryptor->SignedFooterEncrypt(metadata, kFooterEncryptionKey.as_span(),
+                                   str2span(aad), nonce, encrypted);
+    EXPECT_EQ(std::vector<uint8_t>(tag.begin(), tag.end()),
+              std::vector<uint8_t>(encrypted.end() - kGcmTagLength, encrypted.end()));
+  }
+
+  void ReadFirstSecretValue(std::shared_ptr<FileDecryptionProperties> properties) {
+    auto reader = OpenReader(std::move(properties));
+    auto column = reader->RowGroup(0)->Column(5);
+    auto* double_reader = static_cast<DoubleReader*>(column.get());
+    double value;
+    int64_t values_read;
+    double_reader->ReadBatch(1, nullptr, nullptr, &value, &values_read);
+  }
+
+  std::unique_ptr<TemporaryDir> temp_dir_;
+  std::string file_path_;
+  std::string file_aad_;
+};
+
+TEST_P(TestGeneratedEncryptionConfiguration, AlgorithmMetadataAndRoundTrip) {
+  const auto cipher = std::get<0>(GetParam());
+  const bool plaintext_footer = std::get<1>(GetParam());
+  const auto prefix_mode = std::get<2>(GetParam());
+  if (plaintext_footer) {
+    CheckPlaintextFooterSignature();
+  }
+  auto reader = OpenReader(MakeDecryptionProperties());
+  const auto metadata = reader->metadata();
+  if (plaintext_footer) {
+    ASSERT_TRUE(metadata->is_encryption_algorithm_set());
+    const auto algorithm = metadata->encryption_algorithm();
+    ASSERT_EQ(cipher, algorithm.algorithm);
+    EXPECT_FALSE(algorithm.aad.aad_file_unique.empty());
+    EXPECT_EQ(prefix_mode == AadPrefixMode::External, algorithm.aad.supply_aad_prefix);
+    EXPECT_EQ(prefix_mode == AadPrefixMode::Stored ? kFileName : "",
+              algorithm.aad.aad_prefix);
+  } else {
+    // An encrypted footer declares the algorithm in FileCryptoMetaData instead.
+    EXPECT_FALSE(metadata->is_encryption_algorithm_set());
+  }
+  reader->Close();
+
+  FileDecryptor decryptor;
+  // Verify every physical column and all row groups through sync and async readers.
+  ASSERT_NO_THROW(decryptor.DecryptFile(file_path_, MakeDecryptionProperties()));
+  if (prefix_mode == AadPrefixMode::Stored) {
+    ASSERT_NO_THROW(decryptor.DecryptFile(
+        file_path_, MakeDecryptionProperties(Keys::All, AadPrefix::Missing)));
+  }
+  if (plaintext_footer) {
+    // This reads only public columns and does not authenticate the keyless footer.
+    ASSERT_NO_THROW(decryptor.DecryptFile(file_path_, nullptr));
+  }
+  EXPECT_THROW(ReadFirstSecretValue(MakeDecryptionProperties(Keys::MissingColumn)),
+               std::out_of_range);
+  EXPECT_THROW(OpenReader(MakeDecryptionProperties(Keys::WrongFooter)), ParquetException);
+  if (prefix_mode == AadPrefixMode::External) {
+    EXPECT_THROW(OpenReader(MakeDecryptionProperties(Keys::All, AadPrefix::Missing)),
+                 ParquetException);
+    EXPECT_THROW(OpenReader(MakeDecryptionProperties(Keys::All, AadPrefix::Wrong)),
+                 ParquetException);
+  }
+}
+
+INSTANTIATE_TEST_SUITE_P(
+    GeneratedFiles, TestGeneratedEncryptionConfiguration,
+    ::testing::Combine(::testing::Values(ParquetCipher::AES_GCM_V1,
+                                         ParquetCipher::AES_GCM_CTR_V1),
+                       ::testing::Bool(),
+                       ::testing::Values(AadPrefixMode::None, AadPrefixMode::Stored,
+                                         AadPrefixMode::External)));
+
 // Set temp_dir before running the write/read tests. The encrypted files will
 // be written/read from this directory.
 void TestEncryptionConfiguration::SetUpTestCase() {

cpp/src/parquet/metadata.cc

@@ -823,9 +823,11 @@ class FileMetaData::FileMetaDataImpl {
     const SecureString& key = file_decryptor_->GetFooterKey();
     const std::string& aad = encryption::CreateFooterAad(file_decryptor_->file_aad());
 
-    auto aes_encryptor = encryption::AesEncryptor::Make(file_decryptor_->algorithm(),
-                                                        static_cast<int>(key.size()),
-                                                        true, false /*write_length*/);
+    // A plaintext footer is always signed with AES-GCM, regardless of the algorithm
+    // configured for encryption.
+    auto aes_encryptor = encryption::AesEncryptor::Make(
+        ParquetCipher::AES_GCM_V1, static_cast<int>(key.size()), /*metadata=*/true,
+        /*write_length=*/false);
 
     std::shared_ptr<Buffer> encrypted_buffer = AllocateBuffer(
         file_decryptor_->pool(), aes_encryptor->CiphertextLength(
@@ -1202,8 +1204,10 @@ bool FileMetaData::VerifySignature(std::span<const uint8_t> serialized_metadata,
   const SecureString& key = file_decryptor->GetFooterKey();
   const std::string& aad = encryption::CreateFooterAad(file_decryptor->file_aad());
 
+  // A plaintext footer is always signed with AES-GCM, regardless of the algorithm
+  // configured for encryption.
   auto aes_encryptor = encryption::AesEncryptor::Make(
-      file_decryptor->algorithm(), static_cast<int>(key.size()), /*metadata=*/true,
+      ParquetCipher::AES_GCM_V1, static_cast<int>(key.size()), /*metadata=*/true,
       /*write_length=*/false);
 
   std::shared_ptr<Buffer> encrypted_buffer =
@@ -2133,19 +2137,11 @@ class FileMetaDataBuilder::FileMetaDataBuilderImpl {
     metadata_->column_orders.resize(schema_->num_columns(), column_order);
     metadata_->__isset.column_orders = true;
 
-    // if plaintext footer, set footer signing algorithm
+    // Plaintext footers declare the file encryption algorithm and signing key metadata.
     auto file_encryption_properties = properties_->file_encryption_properties();
     if (file_encryption_properties && !file_encryption_properties->encrypted_footer()) {
-      EncryptionAlgorithm signing_algorithm;
-      EncryptionAlgorithm algo = file_encryption_properties->algorithm();
-      signing_algorithm.aad.aad_file_unique = algo.aad.aad_file_unique;
-      signing_algorithm.aad.supply_aad_prefix = algo.aad.supply_aad_prefix;
-      if (!algo.aad.supply_aad_prefix) {
-        signing_algorithm.aad.aad_prefix = algo.aad.aad_prefix;
-      }
-      signing_algorithm.algorithm = ParquetCipher::AES_GCM_V1;
-
-      metadata_->__set_encryption_algorithm(ToThrift(signing_algorithm));
+      metadata_->__set_encryption_algorithm(
+          ToThrift(file_encryption_properties->algorithm()));
       const std::string& footer_signing_key_metadata =
           file_encryption_properties->footer_key_metadata();
       if (footer_signing_key_metadata.size() > 0) {