fix(parquet/encoding): validate delta binary packed headers

parquet/internal/encoding/delta_bit_packing.go

@@ -84,23 +84,46 @@ func (d *deltaBitPackDecoder[T]) SetData(nvalues int, data []byte) error {
 	if !ok {
 		return errors.New("parquet: eof exception")
 	}
+	if d.blockSize == 0 {
+		return errors.New("parquet: cannot have zero values per block")
+	}
+	if d.blockSize > math.MaxUint32 || d.blockSize%128 != 0 {
+		return fmt.Errorf("parquet: values per block must be a multiple of 128, got %d", d.blockSize)
+	}
 
 	if d.miniBlocksPerBlock, ok = d.bitdecoder.GetVlqInt(); !ok {
 		return errors.New("parquet: eof exception")
 	}
 	if d.miniBlocksPerBlock == 0 {
 		return errors.New("parquet: cannot have zero miniblock per block")
 	}
+	if d.miniBlocksPerBlock > math.MaxUint32 {
+		return fmt.Errorf("parquet: too many miniblocks per block: %d", d.miniBlocksPerBlock)
+	}
+	if d.blockSize%d.miniBlocksPerBlock != 0 {
+		return fmt.Errorf("parquet: miniblocks per block must divide the block size, got %d miniblocks for %d values", d.miniBlocksPerBlock, d.blockSize)
+	}
+
+	valuesPerMini := d.blockSize / d.miniBlocksPerBlock
+	if valuesPerMini == 0 || valuesPerMini%32 != 0 {
+		return fmt.Errorf("parquet: values per miniblock must be a nonzero multiple of 32, got %d", valuesPerMini)
+	}
 
 	if d.totalValues, ok = d.bitdecoder.GetVlqInt(); !ok {
 		return errors.New("parquet: eof exception")
 	}
+	if d.totalValues > math.MaxUint32 || nvalues < 0 || d.totalValues > uint64(nvalues) {
+		return fmt.Errorf("parquet: invalid delta value count %d for %d buffered values", d.totalValues, nvalues)
+	}
 
 	if d.lastVal, ok = d.bitdecoder.GetZigZagVlqInt(); !ok {
 		return errors.New("parquet: eof exception")
 	}
+	if d.Type() == parquet.Types.Int32 && (d.lastVal < math.MinInt32 || d.lastVal > math.MaxInt32) {
+		return fmt.Errorf("parquet: initial delta value %d exceeds INT32 range", d.lastVal)
+	}
 
-	d.valsPerMini = uint32(d.blockSize / d.miniBlocksPerBlock)
+	d.valsPerMini = uint32(valuesPerMini)
 	d.usedFirst = false
 	d.nvals = int(d.totalValues)
 	return nil
@@ -113,6 +136,9 @@ func (d *deltaBitPackDecoder[T]) initBlock() error {
 	if d.minDelta, ok = d.bitdecoder.GetZigZagVlqInt(); !ok {
 		return errors.New("parquet: eof exception")
 	}
+	if d.Type() == parquet.Types.Int32 && (d.minDelta < math.MinInt32 || d.minDelta > math.MaxInt32) {
+		return fmt.Errorf("parquet: minimum delta %d exceeds INT32 range", d.minDelta)
+	}
 
 	// ensure we have enough space for our miniblocks to decode the widths
 	d.deltaBitWidths.Resize(int(d.miniBlocksPerBlock))
@@ -137,6 +163,13 @@ func (d *deltaBitPackDecoder[T]) unpackNextMini() error {
 		d.miniBlockValues = d.miniBlockValues[:0]
 	}
 	d.deltaBitWidth = d.deltaBitWidths.Bytes()[int(d.miniBlockIdx)]
+	maxBitWidth := byte(64)
+	if d.Type() == parquet.Types.Int32 {
+		maxBitWidth = 32
+	}
+	if d.deltaBitWidth > maxBitWidth {
+		return fmt.Errorf("parquet: delta bit width %d exceeds decoder width %d", d.deltaBitWidth, maxBitWidth)
+	}
 	d.currentMiniBlockVals = d.valsPerMini
 
 	n := int(d.valsPerMini)

parquet/internal/encoding/delta_bit_packing_validation_test.go

@@ -0,0 +1,76 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+package encoding
+
+import (
+	"testing"
+
+	"github.com/apache/arrow-go/v18/arrow/memory"
+	"github.com/apache/arrow-go/v18/parquet"
+	"github.com/stretchr/testify/require"
+)
+
+func TestDeltaBitPackDecoderValidatesHeader(t *testing.T) {
+	tests := [][]byte{
+		{0, 1, 1, 0},         // zero values per block
+		{1, 1, 1, 0},         // block size is not a multiple of 128
+		{128, 1, 0, 1, 0},    // zero miniblocks
+		{128, 1, 8, 1, 0},    // 16 values per miniblock
+		{128, 26, 101, 1, 0}, // miniblocks do not divide block size
+		{128, 1, 4, 2, 0},    // encoded count exceeds page count
+	}
+
+	for _, data := range tests {
+		dec := NewDecoder(parquet.Types.Int32, parquet.Encodings.DeltaBinaryPacked, nil, memory.DefaultAllocator)
+		require.Error(t, dec.SetData(1, data))
+	}
+}
+
+func TestDeltaBitPackInt32DecoderValidatesEncodedIntegerRange(t *testing.T) {
+	// 2147483648 zigzag-encodes to the final five bytes in each payload.
+	outOfRange := []byte{128, 128, 128, 128, 16}
+
+	dec := NewDecoder(parquet.Types.Int32, parquet.Encodings.DeltaBinaryPacked, nil, memory.DefaultAllocator)
+	require.Error(t, dec.SetData(1, append([]byte{128, 1, 4, 1}, outOfRange...)))
+
+	data := append([]byte{128, 1, 4, 2, 0}, outOfRange...)
+	require.NoError(t, dec.SetData(2, data))
+	_, err := dec.Discard(2)
+	require.Error(t, err)
+}
+
+func TestDeltaBitPackDecoderValidatesUsedBitWidth(t *testing.T) {
+	// Header: 128 values/block, 4 miniblocks, 2 values, first value 0.
+	// Block: min delta 0, then the supplied bit width for the first miniblock.
+	for _, tc := range []struct {
+		name     string
+		typ      parquet.Type
+		bitWidth byte
+	}{
+		{name: "int32", typ: parquet.Types.Int32, bitWidth: 33},
+		{name: "int64", typ: parquet.Types.Int64, bitWidth: 65},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			data := []byte{128, 1, 4, 2, 0, 0, tc.bitWidth, 0, 0, 0}
+			dec := NewDecoder(tc.typ, parquet.Encodings.DeltaBinaryPacked, nil, memory.DefaultAllocator)
+			require.NoError(t, dec.SetData(2, data))
+
+			_, err := dec.Discard(2)
+			require.Error(t, err)
+		})
+	}
+}