This is an automated email from the ASF dual-hosted git repository.
zeroshade pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/arrow-go.git
The following commit(s) were added to refs/heads/main by this push:
new 038c0b44 fix(parquet): return errors for malformed encrypted payloads
(#945)
038c0b44 is described below
commit 038c0b444b70547ceecd230d1e019fe176407d4e
Author: Minh Vu <[email protected]>
AuthorDate: Wed Jul 15 18:16:05 2026 +0200
fix(parquet): return errors for malformed encrypted payloads (#945)
---
parquet/file/file_reader_test.go | 4 +-
parquet/file/page_reader.go | 26 +++++++-
parquet/internal/encryption/aes.go | 105 ++++++++++++++++++-------------
parquet/internal/encryption/aes_test.go | 77 +++++++++++++++++++++++
parquet/internal/encryption/decryptor.go | 10 +--
parquet/metadata/bloom_filter.go | 10 ++-
parquet/metadata/column_chunk.go | 9 ++-
parquet/metadata/file.go | 6 +-
parquet/metadata/page_index.go | 20 ++++--
9 files changed, 207 insertions(+), 60 deletions(-)
diff --git a/parquet/file/file_reader_test.go b/parquet/file/file_reader_test.go
index 5ae9ae74..9fa38c3c 100644
--- a/parquet/file/file_reader_test.go
+++ b/parquet/file/file_reader_test.go
@@ -282,9 +282,9 @@ func (m *mockDecryptor) Allocator() memory.Allocator {
func (m *mockDecryptor) CiphertextSizeDelta() int { return 0 }
-func (m *mockDecryptor) Decrypt(src []byte) []byte { return m.decrypt(src) }
+func (m *mockDecryptor) Decrypt(src []byte) ([]byte, error) { return
m.decrypt(src), nil }
-func (m *mockDecryptor) DecryptFrom(r io.Reader) []byte {
+func (m *mockDecryptor) DecryptFrom(r io.Reader, _ int64) ([]byte, error) {
decrypted, _ := io.ReadAll(r)
return m.Decrypt(decrypted)
}
diff --git a/parquet/file/page_reader.go b/parquet/file/page_reader.go
index 70898d2f..2b748282 100644
--- a/parquet/file/page_reader.go
+++ b/parquet/file/page_reader.go
@@ -608,7 +608,10 @@ func (p *serializedPageReader) decompress(rd io.Reader,
lenCompressed int, buf [
data := p.decompressBuffer.Bytes()[:lenCompressed]
if p.cryptoCtx.DataDecryptor != nil {
- data = p.cryptoCtx.DataDecryptor.Decrypt(data)
+ data, err = p.cryptoCtx.DataDecryptor.Decrypt(data)
+ if err != nil {
+ return nil, fmt.Errorf("decrypting page data: %w", err)
+ }
}
decoded, err := compress.Decode(p.codec, buf, data)
@@ -630,7 +633,10 @@ func (p *serializedPageReader) readV2Encrypted(rd
io.Reader, lenCompressed int,
return n, fmt.Errorf("parquet: expected to read %d compressed
bytes, got %d", lenCompressed, n)
}
- data :=
p.cryptoCtx.DataDecryptor.Decrypt(p.decompressBuffer.Bytes()[:lenCompressed])
+ data, err :=
p.cryptoCtx.DataDecryptor.Decrypt(p.decompressBuffer.Bytes()[:lenCompressed])
+ if err != nil {
+ return n, fmt.Errorf("decrypting data page: %w", err)
+ }
// encrypted + uncompressed -> just copy the decrypted data to output
buffer
if !compressed {
@@ -802,7 +808,21 @@ func (p *serializedPageReader) readPageHeader(rd
parquet.BufferedReader, hdr *fo
extra := 0
if p.cryptoCtx.MetaDecryptor != nil {
p.updateDecryption(p.cryptoCtx.MetaDecryptor,
encryption.DictPageHeaderModule, p.dataPageHeaderAad)
- view = p.cryptoCtx.MetaDecryptor.Decrypt(view)
+ ciphertext, frameErr :=
encryption.FramedCiphertext(view)
+ if errors.Is(frameErr, io.ErrUnexpectedEOF) {
+ allowedPgSz *= 2
+ if allowedPgSz > p.maxPageHeaderSize {
+ return errors.New("parquet:
deserializing page header failed")
+ }
+ continue
+ }
+ if frameErr != nil {
+ return fmt.Errorf("reading encrypted page
header: %w", frameErr)
+ }
+ view, err =
p.cryptoCtx.MetaDecryptor.Decrypt(ciphertext)
+ if err != nil {
+ return fmt.Errorf("decrypting page header: %w",
err)
+ }
extra = p.cryptoCtx.MetaDecryptor.CiphertextSizeDelta()
}
diff --git a/parquet/internal/encryption/aes.go
b/parquet/internal/encryption/aes.go
index 06c9c90c..e6bed554 100644
--- a/parquet/internal/encryption/aes.go
+++ b/parquet/internal/encryption/aes.go
@@ -193,36 +193,58 @@ func newAesDecryptor(alg parquet.Cipher, metadata bool)
*aesDecryptor {
// the length of the plaintext after decryption.
func (a *aesDecryptor) CiphertextSizeDelta() int { return
a.ciphertextSizeDelta }
-// DecryptFrom
-func (a *aesDecryptor) DecryptFrom(r io.Reader, key, aad []byte) []byte {
+// DecryptFrom decrypts a length-prefixed ciphertext read from r. maxCiphertext
+// limits the framed ciphertext payload before it is allocated.
+func (a *aesDecryptor) DecryptFrom(r io.Reader, key, aad []byte, maxCiphertext
int64) ([]byte, error) {
block, err := aes.NewCipher(key)
if err != nil {
- panic(err)
+ return nil, err
}
var writtenCiphertextLen uint32
if err := binary.Read(r, binary.LittleEndian, &writtenCiphertextLen);
err != nil {
- panic(err)
+ return nil, fmt.Errorf("reading encrypted payload length: %w",
err)
+ }
+ if int64(writtenCiphertextLen) > maxCiphertext {
+ return nil, fmt.Errorf("encrypted payload length %d exceeds
limit %d", writtenCiphertextLen, maxCiphertext)
+ }
+ if err := a.validateCiphertextLength(int(writtenCiphertextLen)); err !=
nil {
+ return nil, err
}
cipherText := make([]byte, writtenCiphertextLen)
- if n, err := io.ReadFull(r, cipherText); n != int(writtenCiphertextLen)
|| err != nil {
- panic(err)
+ if _, err := io.ReadFull(r, cipherText); err != nil {
+ return nil, fmt.Errorf("reading encrypted payload: %w", err)
+ }
+
+ return a.decryptPayload(block, cipherText, aad)
+}
+
+func (a *aesDecryptor) validateCiphertextLength(length int) error {
+ minimum := NonceLength
+ if a.mode == gcmMode {
+ minimum += GcmTagLength
}
+ if length < minimum {
+ return fmt.Errorf("encrypted payload length %d is smaller than
minimum %d", length, minimum)
+ }
+ return nil
+}
+func (a *aesDecryptor) decryptPayload(block cipher.Block, cipherText, aad
[]byte) ([]byte, error) {
nonce := cipherText[:NonceLength]
cipherText = cipherText[NonceLength:]
if a.mode == gcmMode {
aead, err := cipher.NewGCM(block)
if err != nil {
- panic(err)
+ return nil, err
}
- plain, err := aead.Open(cipherText[:0], nonce, cipherText, aad)
+ plain, err := aead.Open(nil, nonce, cipherText, aad)
if err != nil {
- panic(err)
+ return nil, fmt.Errorf("decrypting encrypted payload:
%w", err)
}
- return plain
+ return plain, nil
}
// Parquet CTR IVs are comprised of a 12-byte nonce and a 4-byte initial
@@ -234,48 +256,47 @@ func (a *aesDecryptor) DecryptFrom(r io.Reader, key, aad
[]byte) []byte {
iv[ctrIVLen-1] = 1
stream := cipher.NewCTR(block, iv)
- // dst := make([]byte, len(cipherText))
- stream.XORKeyStream(cipherText, cipherText)
- return cipherText
+ plain := make([]byte, len(cipherText))
+ stream.XORKeyStream(plain, cipherText)
+ return plain, nil
}
// Decrypt returns the plaintext version of the given ciphertext when decrypted
// with the provided key and AAD security bytes.
-func (a *aesDecryptor) Decrypt(cipherText, key, aad []byte) []byte {
+func (a *aesDecryptor) Decrypt(cipherText, key, aad []byte) ([]byte, error) {
block, err := aes.NewCipher(key)
if err != nil {
- panic(err)
+ return nil, err
}
- writtenCiphertextLen := binary.LittleEndian.Uint32(cipherText)
- cipherLen := writtenCiphertextLen + bufferSizeLength
- nonce := cipherText[bufferSizeLength : bufferSizeLength+NonceLength]
-
- if a.mode == gcmMode {
- aead, err := cipher.NewGCM(block)
- if err != nil {
- panic(err)
- }
-
- plain, err := aead.Open(nil, nonce,
cipherText[bufferSizeLength+NonceLength:cipherLen], aad)
- if err != nil {
- panic(err)
- }
- return plain
+ frame, err := FramedCiphertext(cipherText)
+ if err != nil {
+ return nil, err
}
+ if len(frame) != len(cipherText) {
+ return nil, fmt.Errorf("encrypted payload has %d trailing
bytes", len(cipherText)-len(frame))
+ }
+ writtenCiphertextLen := len(frame) - bufferSizeLength
+ if err := a.validateCiphertextLength(int(writtenCiphertextLen)); err !=
nil {
+ return nil, err
+ }
+ return a.decryptPayload(block, frame[bufferSizeLength:], aad)
+}
- // Parquet CTR IVs are comprised of a 12-byte nonce and a 4-byte initial
- // counter field.
- // The first 31 bits of the initial counter field are set to 0, the
last bit
- // is set to 1.
- iv := make([]byte, ctrIVLen)
- copy(iv, nonce)
- iv[ctrIVLen-1] = 1
-
- stream := cipher.NewCTR(block, iv)
- dst := make([]byte, len(cipherText)-bufferSizeLength-NonceLength)
- stream.XORKeyStream(dst, cipherText[bufferSizeLength+NonceLength:])
- return dst
+// FramedCiphertext returns the complete first length-prefixed ciphertext frame
+// in src. Additional bytes are left out of the returned slice.
+func FramedCiphertext(src []byte) ([]byte, error) {
+ if len(src) < bufferSizeLength {
+ return nil, fmt.Errorf("encrypted payload is missing its
%d-byte length prefix: %w",
+ bufferSizeLength, io.ErrUnexpectedEOF)
+ }
+ writtenCiphertextLen := binary.LittleEndian.Uint32(src)
+ frameLen := uint64(writtenCiphertextLen) + bufferSizeLength
+ if frameLen > uint64(len(src)) {
+ return nil, fmt.Errorf("encrypted payload length prefix is %d,
available length is %d: %w",
+ writtenCiphertextLen, len(src)-bufferSizeLength,
io.ErrUnexpectedEOF)
+ }
+ return src[:frameLen], nil
}
// CreateModuleAad creates the section AAD security bytes for the file,
module, row group, column and page.
diff --git a/parquet/internal/encryption/aes_test.go
b/parquet/internal/encryption/aes_test.go
new file mode 100644
index 00000000..f36f0e80
--- /dev/null
+++ b/parquet/internal/encryption/aes_test.go
@@ -0,0 +1,77 @@
+// 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 encryption
+
+import (
+ "bytes"
+ "encoding/binary"
+ "testing"
+
+ "github.com/apache/arrow-go/v18/parquet"
+ "github.com/stretchr/testify/assert"
+)
+
+func TestAESDecryptRejectsMalformedCiphertext(t *testing.T) {
+ decryptor := newAesDecryptor(parquet.AesGcm, false)
+ key := make([]byte, 16)
+
+ tests := map[string][]byte{
+ "missing length": nil,
+ "short length": {0x01, 0x02, 0x03},
+ "short payload": framedCiphertext(NonceLength, NonceLength),
+ "truncated": framedCiphertext(NonceLength+GcmTagLength, 1),
+ "trailing data": framedCiphertext(NonceLength+GcmTagLength,
NonceLength+GcmTagLength+1),
+ "invalid tag": framedCiphertext(NonceLength+GcmTagLength,
NonceLength+GcmTagLength),
+ }
+
+ for name, ciphertext := range tests {
+ t.Run(name, func(t *testing.T) {
+ _, err := decryptor.Decrypt(ciphertext, key, nil)
+ assert.Error(t, err)
+ })
+ }
+}
+
+func TestAESDecryptFromValidatesLengthBeforeAllocation(t *testing.T) {
+ decryptor := newAesDecryptor(parquet.AesGcm, false)
+ key := make([]byte, 16)
+
+ t.Run("exceeds limit", func(t *testing.T) {
+ var input [4]byte
+ binary.LittleEndian.PutUint32(input[:], 1<<30)
+ _, err := decryptor.DecryptFrom(bytes.NewReader(input[:]), key,
nil, 1024)
+ assert.ErrorContains(t, err, "exceeds limit")
+ })
+
+ t.Run("short payload", func(t *testing.T) {
+ input := framedCiphertext(NonceLength, NonceLength)
+ _, err := decryptor.DecryptFrom(bytes.NewReader(input), key,
nil, 1024)
+ assert.ErrorContains(t, err, "smaller than minimum")
+ })
+
+ t.Run("truncated payload", func(t *testing.T) {
+ input := framedCiphertext(NonceLength+GcmTagLength, 1)
+ _, err := decryptor.DecryptFrom(bytes.NewReader(input), key,
nil, 1024)
+ assert.ErrorContains(t, err, "reading encrypted payload")
+ })
+}
+
+func framedCiphertext(declared, actual int) []byte {
+ data := make([]byte, bufferSizeLength+actual)
+ binary.LittleEndian.PutUint32(data, uint32(declared))
+ return data
+}
diff --git a/parquet/internal/encryption/decryptor.go
b/parquet/internal/encryption/decryptor.go
index 50a1a6f8..d36bb521 100644
--- a/parquet/internal/encryption/decryptor.go
+++ b/parquet/internal/encryption/decryptor.go
@@ -245,9 +245,9 @@ type Decryptor interface {
// returns the CiphertextSizeDelta from the decryptor
CiphertextSizeDelta() int
// Decrypt just returns the decrypted plaintext from the src ciphertext
- Decrypt(src []byte) []byte
+ Decrypt(src []byte) ([]byte, error)
// Decrypt just returns the decrypted plaintext from the src ciphertext
- DecryptFrom(r io.Reader) []byte
+ DecryptFrom(r io.Reader, maxCiphertext int64) ([]byte, error)
// set the AAD bytes of the decryptor to the provided string
UpdateAad(string)
}
@@ -264,11 +264,11 @@ func (d *decryptor) Allocator() memory.Allocator { return
d.mem }
func (d *decryptor) FileAad() string { return string(d.fileAad) }
func (d *decryptor) UpdateAad(aad string) { d.aad = []byte(aad) }
func (d *decryptor) CiphertextSizeDelta() int { return
d.decryptor.CiphertextSizeDelta() }
-func (d *decryptor) Decrypt(src []byte) []byte {
+func (d *decryptor) Decrypt(src []byte) ([]byte, error) {
return d.decryptor.Decrypt(src, d.key, d.aad)
}
-func (d *decryptor) DecryptFrom(r io.Reader) []byte {
- return d.decryptor.DecryptFrom(r, d.key, d.aad)
+func (d *decryptor) DecryptFrom(r io.Reader, maxCiphertext int64) ([]byte,
error) {
+ return d.decryptor.DecryptFrom(r, d.key, d.aad, maxCiphertext)
}
func getColumnDecryptor(cryptoMetadata *format.ColumnCryptoMetaData,
fileDecryptor FileDecryptor, metadata bool) (Decryptor, error) {
diff --git a/parquet/metadata/bloom_filter.go b/parquet/metadata/bloom_filter.go
index bd9faaec..72371aab 100644
--- a/parquet/metadata/bloom_filter.go
+++ b/parquet/metadata/bloom_filter.go
@@ -464,7 +464,10 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i
int) (BloomFilter, er
encryption.UpdateDecryptor(decryptor, r.rgOrdinal, int16(i),
encryption.BloomFilterHeaderModule)
- hdr := decryptor.DecryptFrom(sectionRdr)
+ hdr, err := decryptor.DecryptFrom(sectionRdr,
maximumBloomFilterBytes)
+ if err != nil {
+ return nil, fmt.Errorf("decrypting bloom filter header:
%w", err)
+ }
if _, err = thrift.DeserializeThrift(&header, hdr); err != nil {
return nil, err
}
@@ -475,7 +478,10 @@ func (r *RowGroupBloomFilterReader) GetColumnBloomFilter(i
int) (BloomFilter, er
encryption.UpdateDecryptor(decryptor, r.rgOrdinal, int16(i),
encryption.BloomFilterBitsetModule)
- bitset := decryptor.DecryptFrom(sectionRdr)
+ bitset, err := decryptor.DecryptFrom(sectionRdr,
int64(header.NumBytes)+int64(decryptor.CiphertextSizeDelta()))
+ if err != nil {
+ return nil, fmt.Errorf("decrypting bloom filter bitset:
%w", err)
+ }
if len(bitset) != int(header.NumBytes) {
return nil, fmt.Errorf("wrong length of decrypted bloom
filter bitset: %d vs %d",
len(bitset), header.NumBytes)
diff --git a/parquet/metadata/column_chunk.go b/parquet/metadata/column_chunk.go
index e3533fb2..b387a442 100644
--- a/parquet/metadata/column_chunk.go
+++ b/parquet/metadata/column_chunk.go
@@ -20,6 +20,7 @@ import (
"bytes"
"context"
"errors"
+ "fmt"
"io"
"reflect"
@@ -103,7 +104,13 @@ func NewColumnChunkMetaData(column *format.ColumnChunk,
descr *schema.Column, wr
keyMetadata :=
ccmd.ENCRYPTION_WITH_COLUMN_KEY.GetKeyMetadata()
aadColumnMetadata :=
encryption.CreateModuleAad(fileDecryptor.FileAad(),
encryption.ColumnMetaModule, rowGroupOrdinal, columnOrdinal, -1)
decryptor :=
fileDecryptor.GetColumnMetaDecryptor(path.String(), string(keyMetadata),
aadColumnMetadata)
- thrift.DeserializeThrift(&c.decryptedMeta,
decryptor.Decrypt(column.GetEncryptedColumnMetadata()))
+ decrypted, err :=
decryptor.Decrypt(column.GetEncryptedColumnMetadata())
+ if err != nil {
+ return nil, fmt.Errorf("decrypting
column metadata: %w", err)
+ }
+ if _, err :=
thrift.DeserializeThrift(&c.decryptedMeta, decrypted); err != nil {
+ return nil, err
+ }
c.columnMeta = &c.decryptedMeta
} else {
return nil, errors.New("cannot decrypt column
metadata. file decryption not setup correctly")
diff --git a/parquet/metadata/file.go b/parquet/metadata/file.go
index 0b5d1284..c3969640 100644
--- a/parquet/metadata/file.go
+++ b/parquet/metadata/file.go
@@ -298,7 +298,11 @@ func NewFileMetaData(data []byte, fileDecryptor
encryption.FileDecryptor) (*File
meta := format.NewFileMetaData()
if fileDecryptor != nil {
footerDecryptor := fileDecryptor.GetFooterDecryptor()
- data = footerDecryptor.Decrypt(data)
+ var err error
+ data, err = footerDecryptor.Decrypt(data)
+ if err != nil {
+ return nil, fmt.Errorf("decrypting file metadata: %w",
err)
+ }
}
remain, err := thrift.DeserializeThrift(meta, data)
diff --git a/parquet/metadata/page_index.go b/parquet/metadata/page_index.go
index 030dd49f..f6b7e2b7 100644
--- a/parquet/metadata/page_index.go
+++ b/parquet/metadata/page_index.go
@@ -102,7 +102,7 @@ func mustArg[T any](val T, err error) T {
// TypedColumnIndex.
func NewColumnIndex(descr *schema.Column, serializedIndex []byte, props
*parquet.ReaderProperties, decryptor encryption.Decryptor) ColumnIndex {
if decryptor != nil {
- serializedIndex = decryptor.Decrypt(serializedIndex)
+ serializedIndex = mustArg(decryptor.Decrypt(serializedIndex))
}
var colidx format.ColumnIndex
@@ -233,7 +233,7 @@ func (p PageIndexSelection) String() string {
// optionally decrypting it if it was encrypted.
func NewOffsetIndex(serializedIndex []byte, _ *parquet.ReaderProperties,
decryptor encryption.Decryptor) OffsetIndex {
if decryptor != nil {
- serializedIndex = decryptor.Decrypt(serializedIndex)
+ serializedIndex = mustArg(decryptor.Decrypt(serializedIndex))
}
var offsetIndex format.OffsetIndex
@@ -347,12 +347,18 @@ func (r *RowGroupPageIndexReader) GetColumnIndex(i int)
(ColumnIndex, error) {
return nil, err
}
+ serializedIndex := r.colIndexBuffer[bufferOffset :
bufferOffset+int64(colIndexLocation.Length)]
if decryptor != nil {
encryption.UpdateDecryptor(decryptor, int16(r.rgOrdinal),
int16(i), encryption.ColumnIndexModule)
+ decrypted, err := decryptor.Decrypt(serializedIndex)
+ if err != nil {
+ return nil, fmt.Errorf("decrypting column index: %w",
err)
+ }
+ serializedIndex = decrypted
}
- idx := NewColumnIndex(descr, r.colIndexBuffer[bufferOffset:], r.props,
decryptor)
+ idx := NewColumnIndex(descr, serializedIndex, r.props, nil)
r.colIndexes[i] = idx
return idx, nil
}
@@ -401,12 +407,18 @@ func (r *RowGroupPageIndexReader) GetOffsetIndex(i int)
(OffsetIndex, error) {
return nil, err
}
+ serializedIndex := r.offsetIndexBuffer[bufferOffset :
bufferOffset+int64(offsetIndexLocation.Length)]
if decryptor != nil {
encryption.UpdateDecryptor(decryptor, int16(r.rgOrdinal),
int16(i), encryption.OffsetIndexModule)
+ decrypted, err := decryptor.Decrypt(serializedIndex)
+ if err != nil {
+ return nil, fmt.Errorf("decrypting offset index: %w",
err)
+ }
+ serializedIndex = decrypted
}
- oidx := NewOffsetIndex(r.offsetIndexBuffer[bufferOffset:], r.props,
decryptor)
+ oidx := NewOffsetIndex(serializedIndex, r.props, nil)
r.offsetIndices[i] = oidx
return oidx, nil
}
