fix(parquet): preserve CTR algorithm in plaintext footers

parquet/file/file_reader_test.go

@@ -1043,3 +1043,103 @@ func TestListColumns(t *testing.T) {
 		}
 	}
 }
+
+func TestEncryptedFileCipherAndAAD(t *testing.T) {
+	const footerKey = "0123456789012345"
+	values := []int32{3, 7, 11}
+	sc, err := schema.NewGroupNode("schema", parquet.Repetitions.Required,
+		schema.FieldList{schema.NewInt32Node("value", parquet.Repetitions.Required, -1)}, -1)
+	require.NoError(t, err)
+	for _, cipher := range []struct {
+		name      string
+		algorithm parquet.Cipher
+	}{
+		{"GCM", parquet.AesGcm}, {"CTR", parquet.AesCtr},
+	} {
+		for _, footer := range []struct {
+			name        string
+			plaintext   bool
+			aad         string
+			externalAAD bool
+		}{
+			{"encrypted", false, "", false},
+			{"plaintext", true, "", false},
+			{"plaintext_stored_aad", true, "test-file", false},
+			{"plaintext_external_aad", true, "test-file", true},
+		} {
+			t.Run(fmt.Sprintf("%s/%s", cipher.name, footer.name), func(t *testing.T) {
+				opts := []parquet.EncryptOption{parquet.WithAlg(cipher.algorithm), parquet.WithFooterKeyMetadata("footer-key")}
+				if footer.plaintext {
+					opts = append(opts, parquet.WithPlaintextFooter())
+				}
+				if footer.aad != "" {
+					opts = append(opts, parquet.WithAadPrefix(footer.aad))
+				}
+				if footer.externalAAD {
+					opts = append(opts, parquet.DisableAadPrefixStorage())
+				}
+				encryption := parquet.NewFileEncryptionProperties(footerKey, opts...)
+				props := parquet.NewWriterProperties(parquet.WithEncryptionProperties(encryption), parquet.WithDictionaryDefault(false))
+				var buf bytes.Buffer
+				writer, err := file.NewParquetWriterWithError(&buf, sc, file.WithWriterProps(props))
+				require.NoError(t, err)
+				group, err := writer.AppendRowGroupChecked()
+				require.NoError(t, err)
+				column, err := group.NextColumn()
+				require.NoError(t, err)
+				n, err := column.(*file.Int32ColumnChunkWriter).WriteBatch(values, nil, nil)
+				require.NoError(t, err)
+				require.EqualValues(t, len(values), n)
+				require.NoError(t, group.Close())
+				require.NoError(t, writer.Close())
+
+				if footer.plaintext {
+					// Inspect the serialized footer, including the AAD storage policy.
+					data := buf.Bytes()
+					footerLength := int(binary.LittleEndian.Uint32(data[len(data)-8:]))
+					meta, err := metadata.NewFileMetaData(data[len(data)-8-footerLength:len(data)-8], nil)
+					require.NoError(t, err)
+					assert.Equal(t, encryption.Algorithm(), meta.EncryptionAlgorithm())
+					assert.Equal(t, []byte("footer-key"), meta.GetFooterSigningKeyMetadata())
+				}
+
+				alloc := memory.NewCheckedAllocator(memory.NewGoAllocator())
+				defer alloc.AssertSize(t, 0)
+
+				readerProps := func() *parquet.ReaderProperties {
+					decryption := []parquet.FileDecryptionOption{parquet.WithFooterKey(footerKey)}
+					if footer.externalAAD {
+						decryption = append(decryption, parquet.WithDecryptAadPrefix(footer.aad))
+					}
+					p := parquet.NewReaderProperties(alloc)
+					p.FileDecryptProps = parquet.NewFileDecryptionProperties(decryption...)
+					return p
+				}
+				reader, err := file.NewParquetReader(bytes.NewReader(buf.Bytes()), file.WithReadProps(readerProps()))
+				require.NoError(t, err)
+				defer reader.Close()
+				require.EqualValues(t, len(values), reader.NumRows())
+				readColumn, err := reader.RowGroup(0).Column(0)
+				require.NoError(t, err)
+				defer readColumn.Close()
+				actual := make([]int32, len(values))
+				total, count, err := readColumn.(*file.Int32ColumnChunkReader).ReadBatch(int64(len(values)), actual, nil, nil)
+				require.NoError(t, err)
+				require.NoError(t, readColumn.Err())
+				require.EqualValues(t, len(values), total)
+				require.Equal(t, len(values), count)
+				assert.Equal(t, values, actual)
+				assert.False(t, readColumn.HasNext())
+				require.NoError(t, readColumn.Err())
+
+				if cipher.algorithm == parquet.AesCtr && footer.plaintext && footer.aad == "" {
+					// CTR data pages still use a GCM signature for a plaintext footer.
+					corrupted := bytes.Clone(buf.Bytes())
+					corrupted[len(corrupted)-9] ^= 1
+					_, err := file.NewParquetReader(bytes.NewReader(corrupted), file.WithReadProps(readerProps()))
+					require.ErrorContains(t, err, "signature verification failed")
+				}
+			})
+		}
+	}
+}

parquet/metadata/file.go

@@ -153,15 +153,7 @@ func (f *FileMetaDataBuilder) Snapshot() (*FileMetaData, error) {
 
 	encryptProps := f.props.FileEncryptionProperties()
 	if encryptProps != nil && !encryptProps.EncryptedFooter() {
-		var signingAlgo parquet.Algorithm
-		algo := encryptProps.Algorithm()
-		signingAlgo.Aad.AadFileUnique = algo.Aad.AadFileUnique
-		signingAlgo.Aad.SupplyAadPrefix = algo.Aad.SupplyAadPrefix
-		if !algo.Aad.SupplyAadPrefix {
-			signingAlgo.Aad.AadPrefix = algo.Aad.AadPrefix
-		}
-		signingAlgo.Algo = parquet.AesGcm
-		f.metadata.EncryptionAlgorithm = signingAlgo.ToThrift()
+		f.metadata.EncryptionAlgorithm = encryptProps.Algorithm().ToThrift()
 		footerSigningMetadata := f.props.FileEncryptionProperties().FooterKeyMetadata()
 		if footerSigningMetadata != "" {
 			f.metadata.FooterSigningKeyMetadata = []byte(footerSigningMetadata)