GH-50314: [C++][Parquet] Reject invalid DELTA_BINARY_PACKED headers

cpp/src/parquet/decoder.cc

@@ -1610,11 +1610,24 @@ class DeltaBitPackDecoder : public TypedDecoderImpl<DType> {
     }
 
     total_values_remaining_ = total_value_count_;
-    if (delta_bit_widths_ == nullptr) {
-      delta_bit_widths_ = AllocateBuffer(pool_, mini_blocks_per_block_);
-    } else {
-      PARQUET_THROW_NOT_OK(
-          delta_bit_widths_->Resize(mini_blocks_per_block_, /*shrink_to_fit*/ false));
+    if (total_value_count_ > 1) {
+      const int64_t required_blocks = ::arrow::bit_util::CeilDiv(
+          static_cast<int64_t>(total_value_count_) - 1, values_per_block_);
+      const int64_t required_bit_widths =
+          required_blocks * static_cast<int64_t>(mini_blocks_per_block_);
+      const int64_t bytes_left = decoder_->bytes_left();
+      if (required_bit_widths > bytes_left) {
+        throw ParquetException("the required number of miniblock bit widths (" +
+                               std::to_string(required_bit_widths) +
+                               ") exceeds the number of bytes remaining in the data (" +
+                               std::to_string(bytes_left) + ")");
+      }
+      if (delta_bit_widths_ == nullptr) {
+        delta_bit_widths_ = AllocateBuffer(pool_, mini_blocks_per_block_);
+      } else {
+        PARQUET_THROW_NOT_OK(
+            delta_bit_widths_->Resize(mini_blocks_per_block_, /*shrink_to_fit*/ false));
+      }
     }
     first_block_initialized_ = false;
     values_remaining_current_mini_block_ = 0;

cpp/src/parquet/encoding_test.cc

@@ -15,6 +15,7 @@
 // specific language governing permissions and limitations
 // under the License.
 
+#include <gmock/gmock.h>
 #include <gtest/gtest.h>
 
 #include <algorithm>
@@ -31,6 +32,7 @@
 #include "arrow/array/builder_dict.h"
 #include "arrow/array/concatenate.h"
 #include "arrow/compute/cast.h"
+#include "arrow/memory_pool.h"
 #include "arrow/testing/gtest_util.h"
 #include "arrow/testing/random.h"
 #include "arrow/testing/util.h"
@@ -1902,6 +1904,94 @@ TYPED_TEST(TestDeltaBitPackEncoding, BasicRoundTrip) {
   }
 }
 
+TYPED_TEST(TestDeltaBitPackEncoding, SingleValueSkipsMiniblockAllocation) {
+  using T = typename TypeParam::c_type;
+
+  // Header: 2^25 values per block, 2^20 miniblocks, 1 value, and first value 0.
+  const std::vector<uint8_t> encoded = {0x80, 0x80, 0x80, 0x10, 0x80,
+                                        0x80, 0x40, 0x01, 0x00};
+  ::arrow::ProxyMemoryPool pool(default_memory_pool());
+  auto decoder = MakeTypedDecoder<TypeParam>(Encoding::DELTA_BINARY_PACKED,
+                                             this->descr_.get(), &pool);
+  T decoded = 1;
+
+  decoder->SetData(1, encoded.data(), static_cast<int>(encoded.size()));
+  ASSERT_EQ(decoder->Decode(&decoded, 1), 1);
+  EXPECT_EQ(decoded, 0);
+  EXPECT_EQ(pool.total_bytes_allocated(), 0);
+}
+
+TYPED_TEST(TestDeltaBitPackEncoding, RejectsMiniblockWidthsLargerThanInput) {
+  using T = typename TypeParam::c_type;
+
+  // Header: 2^25 values per block, 2^20 miniblocks, 2 values, and first value 0,
+  // followed by min delta 0 and no miniblock bit widths.
+  const std::vector<uint8_t> encoded = {0x80, 0x80, 0x80, 0x10, 0x80,
+                                        0x80, 0x40, 0x02, 0x00, 0x00};
+  ::arrow::ProxyMemoryPool pool(default_memory_pool());
+  auto decoder = MakeTypedDecoder<TypeParam>(Encoding::DELTA_BINARY_PACKED,
+                                             this->descr_.get(), &pool);
+  std::vector<T> decoded(2);
+
+  EXPECT_THROW_THAT(
+      [&] {
+        decoder->SetData(2, encoded.data(), static_cast<int>(encoded.size()));
+        decoder->Decode(decoded.data(), static_cast<int>(decoded.size()));
+      },
+      ParquetException,
+      ::testing::Property(
+          &ParquetException::what,
+          ::testing::HasSubstr("the required number of miniblock bit widths "
+                               "(1048576) exceeds the number of bytes remaining "
+                               "in the data (1)")));
+  EXPECT_EQ(pool.total_bytes_allocated(), 0);
+}
+
+TYPED_TEST(TestDeltaBitPackEncoding, RejectsInsufficientBitWidthsAcrossBlocks) {
+  using T = typename TypeParam::c_type;
+
+  // Header: 128 values per block, 1 miniblock, 130 values, and first value 0.
+  const std::vector<uint8_t> encoded = {0x80, 0x01, 0x01, 0x82, 0x01, 0x00, 0x00};
+  ::arrow::ProxyMemoryPool pool(default_memory_pool());
+  auto decoder = MakeTypedDecoder<TypeParam>(Encoding::DELTA_BINARY_PACKED,
+                                             this->descr_.get(), &pool);
+  std::vector<T> decoded(130);
+
+  EXPECT_THROW_THAT(
+      [&] {
+        decoder->SetData(static_cast<int>(decoded.size()), encoded.data(),
+                         static_cast<int>(encoded.size()));
+        decoder->Decode(decoded.data(), static_cast<int>(decoded.size()));
+      },
+      ParquetException,
+      ::testing::Property(
+          &ParquetException::what,
+          ::testing::HasSubstr("the required number of miniblock bit widths (2) "
+                               "exceeds the number of bytes remaining in the "
+                               "data (1)")));
+  EXPECT_EQ(pool.total_bytes_allocated(), 0);
+}
+
+TYPED_TEST(TestDeltaBitPackEncoding, RejectsMissingBitWidthAfterMultiByteMinDelta) {
+  using T = typename TypeParam::c_type;
+
+  // Header: 128 values per block, 1 miniblock, 2 values, and first value 0,
+  // followed by a two-byte min delta and no miniblock bit-width byte.
+  const std::vector<uint8_t> encoded = {0x80, 0x01, 0x01, 0x02, 0x00, 0x80, 0x01};
+  auto decoder =
+      MakeTypedDecoder<TypeParam>(Encoding::DELTA_BINARY_PACKED, this->descr_.get());
+  std::vector<T> decoded(2);
+
+  EXPECT_THROW_THAT(
+      [&] {
+        decoder->SetData(2, encoded.data(), static_cast<int>(encoded.size()));
+        decoder->Decode(decoded.data(), static_cast<int>(decoded.size()));
+      },
+      ParquetException,
+      ::testing::Property(&ParquetException::what,
+                          ::testing::HasSubstr("Decode bit-width EOF")));
+}
+
 TYPED_TEST(TestDeltaBitPackEncoding, NonZeroPaddedMiniblockBitWidth) {
   // GH-14923: depending on the number of encoded values, some of the miniblock
   // bitwidths are actually padding bytes that may take non-conformant values