parquet: tighten encrypted bloom filter checks

Author: fenfeng9

cpp/src/parquet/bloom_filter.cc

@@ -37,8 +37,8 @@ namespace {
 
 constexpr int32_t kCiphertextLengthSize = 4;
 
-// Parse the little-endian length prefix and return the total ciphertext size
-// including the 4-byte length field itself.
+/// Parse the 4-byte little-endian length prefix and return the total ciphertext size,
+/// including the 4-byte length field itself.
 int64_t ParseCiphertextTotalLength(const uint8_t* data, int64_t length) {
   if (length < kCiphertextLengthSize) {
     throw ParquetException("Ciphertext length buffer is too small");
@@ -130,7 +130,8 @@ BlockSplitBloomFilter BlockSplitBloomFilter::Deserialize(
     Decryptor* bitset_decryptor) {
   if (header_decryptor != nullptr || bitset_decryptor != nullptr) {
     if (header_decryptor == nullptr || bitset_decryptor == nullptr) {
-      throw ParquetException("Bloom filter decryptors must be both provided");
+      throw ParquetException(
+          "Bloom filter decryptors must be both provided or both null");
     }
 
     // Encrypted path: header and bitset are encrypted separately.

cpp/src/parquet/bloom_filter_reader.cc

@@ -90,6 +90,16 @@ std::unique_ptr<BloomFilter> RowGroupBloomFilterReaderImpl::GetColumnBloomFilter
   std::unique_ptr<Decryptor> bitset_decryptor =
       InternalFileDecryptor::GetColumnDataDecryptorFactory(file_decryptor_.get(),
                                                            crypto_metadata.get())();
+  if (header_decryptor != nullptr || bitset_decryptor != nullptr) {
+    constexpr auto kEncryptedOrdinalLimit = 32767;
+    if (ARROW_PREDICT_FALSE(row_group_ordinal_ > kEncryptedOrdinalLimit)) {
+      throw ParquetException("Encrypted files cannot contain more than 32767 row groups");
+    }
+    if (ARROW_PREDICT_FALSE(i > kEncryptedOrdinalLimit)) {
+      throw ParquetException("Encrypted files cannot contain more than 32767 columns");
+    }
+  }
+
   if (header_decryptor != nullptr) {
     UpdateDecryptor(header_decryptor.get(), row_group_ordinal_, static_cast<int16_t>(i),
                     encryption::kBloomFilterHeader);
@@ -98,6 +108,7 @@ std::unique_ptr<BloomFilter> RowGroupBloomFilterReaderImpl::GetColumnBloomFilter
     UpdateDecryptor(bitset_decryptor.get(), row_group_ordinal_, static_cast<int16_t>(i),
                     encryption::kBloomFilterBitset);
   }
+
   const int64_t stream_length =
       bloom_filter_length ? *bloom_filter_length : file_size - *bloom_filter_offset;
   auto stream = ::arrow::io::RandomAccessFile::GetStream(input_, *bloom_filter_offset,