diff --git a/parquet-hadoop/README.md b/parquet-hadoop/README.md index 2025243143..3b210c742d 100644 --- a/parquet-hadoop/README.md +++ b/parquet-hadoop/README.md @@ -451,6 +451,10 @@ Note: nested column names must be specified as full dot-separated paths for each **Property:** `parquet.encryption.algorithm` **Description:** Parquet encryption algorithm. Can be `AES_GCM_V1` or `AES_GCM_CTR_V1`. +Since 1.19.0, DataPageV2 pages of encrypted columns encrypt the repetition and definition levels with the values, +as the specification requires. Earlier versions wrote the levels in plaintext. Version 1.19.0 reads both layouts. +Versions 1.18 and earlier cannot read the new pages. With `AES_GCM_V1`, decryption fails. With `AES_GCM_CTR_V1`, +the levels are incorrect. **Default value:** `AES_GCM_V1` --- diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java index a21fec9ab8..8f0e967083 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageReadStore.java @@ -218,85 +218,163 @@ public DataPage visit(DataPageV1 dataPageV1) { @Override public DataPage visit(DataPageV2 dataPageV2) { - if (!dataPageV2.isCompressed() && offsetIndex == null && null == blockDecryptor) { - return dataPageV2; + if (dataPageV2 instanceof SingleModuleDataPageV2) { + return readPageV2WithEncryptedLevels((SingleModuleDataPageV2) dataPageV2, currentPageIndex); } - BytesInput pageBytes = dataPageV2.getData(); - try { - if (options.getAllocator().isDirect() && options.useOffHeapDecryptBuffer()) { - ByteBuffer byteBuffer = pageBytes.toByteBuffer(releaser); - if (!byteBuffer.isDirect()) { - throw new ParquetDecodingException("Expected a direct buffer"); - } - if (blockDecryptor != null) { - byteBuffer = blockDecryptor.decrypt(byteBuffer, dataPageAAD); - } - long compressedSize = byteBuffer.limit(); - if (dataPageV2.isCompressed()) { - int uncompressedSize = Math.toIntExact(dataPageV2.getUncompressedSize() - - dataPageV2.getDefinitionLevels().size() - - dataPageV2.getRepetitionLevels().size()); - ByteBuffer decompressedBuffer = - options.getAllocator().allocate(uncompressedSize); - releaser.releaseLater(decompressedBuffer); - long start = System.nanoTime(); - decompressor.decompress( - byteBuffer, (int) compressedSize, decompressedBuffer, uncompressedSize); - setDecompressMetrics(pageBytes, start); - decompressedBuffer.flip(); - pageBytes = BytesInput.from(decompressedBuffer); - } else { - pageBytes = BytesInput.from(byteBuffer); - } - } else { - if (null != blockDecryptor) { - pageBytes = - BytesInput.from(blockDecryptor.decrypt(pageBytes.toByteArray(), dataPageAAD)); - } - if (dataPageV2.isCompressed()) { - int uncompressedSize = Math.toIntExact(dataPageV2.getUncompressedSize() - - dataPageV2.getDefinitionLevels().size() - - dataPageV2.getRepetitionLevels().size()); - long start = System.nanoTime(); - pageBytes = BytesInput.copy(decompressor.decompress(pageBytes, uncompressedSize)); - setDecompressMetrics(pageBytes, start); - } - } - } catch (IOException e) { - throw new ParquetDecodingException("could not decompress page", e); - } - - final DataPageV2 decompressedPage; - if (offsetIndex == null) { - decompressedPage = DataPageV2.uncompressed( - dataPageV2.getRowCount(), - dataPageV2.getNullCount(), - dataPageV2.getValueCount(), - dataPageV2.getRepetitionLevels(), - dataPageV2.getDefinitionLevels(), - dataPageV2.getDataEncoding(), - pageBytes, - dataPageV2.getStatistics()); - } else { - decompressedPage = DataPageV2.uncompressed( - dataPageV2.getRowCount(), - dataPageV2.getNullCount(), - dataPageV2.getValueCount(), - offsetIndex.getFirstRowIndex(currentPageIndex), - dataPageV2.getRepetitionLevels(), - dataPageV2.getDefinitionLevels(), - dataPageV2.getDataEncoding(), - pageBytes, - dataPageV2.getStatistics()); - } - if (dataPageV2.getCrc().isPresent()) { - decompressedPage.setCrc(dataPageV2.getCrc().getAsInt()); - } - return decompressedPage; + return readPageV2WithPlaintextLevels(dataPageV2, currentPageIndex); } }); } + /** + * Reads a page whose levels are not encrypted. If the column is encrypted, only the values are an + * encrypted module. + * + *

An encrypted page with plaintext levels does not follow the Parquet encryption spec, which + * encrypts the levels and the values as one module. Such pages are written only by parquet-java + * 1.18.0 and earlier, and are read here for backward compatibility only. + */ + private DataPageV2 readPageV2WithPlaintextLevels(DataPageV2 page, int currentPageIndex) { + if (!page.isCompressed() && offsetIndex == null && null == blockDecryptor) { + return page; + } + boolean offHeap = decryptsOffHeap(); + BytesInput values = page.getData(); + try { + if (null != blockDecryptor) { + values = BytesInput.from(decryptDataPageModule(values, offHeap)); + } else if (offHeap) { + values = BytesInput.from(toDirectBuffer(values)); + } + if (page.isCompressed()) { + int levelsByteLength = + Math.toIntExact(page.getRepetitionLevels().size() + + page.getDefinitionLevels().size()); + values = decompressValues(values, valuesUncompressedSize(page, levelsByteLength), offHeap); + } + } catch (IOException e) { + throw new ParquetDecodingException("could not decompress page", e); + } + return uncompressedPageV2( + page, page.getRepetitionLevels(), page.getDefinitionLevels(), values, currentPageIndex); + } + + /** + * Reads a page whose levels and values are one encrypted module: decrypts the module, splits it + * into levels and values, and decompresses the values. + */ + private DataPageV2 readPageV2WithEncryptedLevels(SingleModuleDataPageV2 page, int currentPageIndex) { + boolean offHeap = decryptsOffHeap(); + int repetitionLevelsByteLength = page.getRepetitionLevelsByteLength(); + int definitionLevelsByteLength = page.getDefinitionLevelsByteLength(); + int levelsByteLength = repetitionLevelsByteLength + definitionLevelsByteLength; + try { + ByteBuffer decryptedPage = decryptDataPageModule(page.getEncryptedModule(), offHeap); + checkDecryptedPageSize(page, decryptedPage.limit(), levelsByteLength); + BytesInput repetitionLevels = BytesInput.from(decryptedPage.slice(0, repetitionLevelsByteLength)); + BytesInput definitionLevels = + BytesInput.from(decryptedPage.slice(repetitionLevelsByteLength, definitionLevelsByteLength)); + BytesInput values = BytesInput.from( + decryptedPage.slice(levelsByteLength, decryptedPage.limit() - levelsByteLength)); + if (page.isCompressed()) { + values = decompressValues(values, valuesUncompressedSize(page, levelsByteLength), offHeap); + } + return uncompressedPageV2(page, repetitionLevels, definitionLevels, values, currentPageIndex); + } catch (IOException e) { + throw new ParquetDecodingException("could not decompress page", e); + } + } + + private void checkDecryptedPageSize(DataPageV2 page, int decryptedSize, int levelsByteLength) { + if (decryptedSize < levelsByteLength) { + throw new ParquetDecodingException("Decrypted DataPageV2 page has " + decryptedSize + + " bytes, which is less than the " + levelsByteLength + + " bytes of repetition and definition levels in the page header"); + } + if (!page.isCompressed() && decryptedSize != page.getUncompressedSize()) { + throw new ParquetDecodingException("Decrypted DataPageV2 page has " + decryptedSize + + " bytes, but the page header gives an uncompressed page size of " + + page.getUncompressedSize() + " bytes"); + } + } + + private static int valuesUncompressedSize(DataPageV2 page, int levelsByteLength) { + return Math.toIntExact(page.getUncompressedSize() - levelsByteLength); + } + + private boolean decryptsOffHeap() { + return options.getAllocator().isDirect() && options.useOffHeapDecryptBuffer(); + } + + /** Decrypts in place if off heap, otherwise into a new heap buffer. */ + private ByteBuffer decryptDataPageModule(BytesInput module, boolean offHeap) throws IOException { + if (offHeap) { + return blockDecryptor.decrypt(toDirectBuffer(module), dataPageAAD); + } + return ByteBuffer.wrap(blockDecryptor.decrypt(module.toByteArray(), dataPageAAD)); + } + + private ByteBuffer toDirectBuffer(BytesInput bytes) { + ByteBuffer buffer = bytes.toByteBuffer(releaser); + if (!buffer.isDirect()) { + throw new ParquetDecodingException("Expected a direct buffer"); + } + return buffer; + } + + private BytesInput decompressValues(BytesInput compressed, int uncompressedSize, boolean offHeap) + throws IOException { + long start = System.nanoTime(); + BytesInput decompressed; + if (offHeap) { + ByteBuffer input = toDirectBuffer(compressed); + ByteBuffer output = options.getAllocator().allocate(uncompressedSize); + releaser.releaseLater(output); + decompressor.decompress(input, input.limit(), output, uncompressedSize); + output.flip(); + decompressed = BytesInput.from(output); + } else { + decompressed = BytesInput.copy(decompressor.decompress(compressed, uncompressedSize)); + } + setDecompressMetrics(compressed, start); + return decompressed; + } + + private DataPageV2 uncompressedPageV2( + DataPageV2 source, + BytesInput repetitionLevels, + BytesInput definitionLevels, + BytesInput values, + int currentPageIndex) { + final DataPageV2 decompressedPage; + if (offsetIndex == null) { + decompressedPage = DataPageV2.uncompressed( + source.getRowCount(), + source.getNullCount(), + source.getValueCount(), + repetitionLevels, + definitionLevels, + source.getDataEncoding(), + values, + source.getStatistics()); + } else { + decompressedPage = DataPageV2.uncompressed( + source.getRowCount(), + source.getNullCount(), + source.getValueCount(), + offsetIndex.getFirstRowIndex(currentPageIndex), + repetitionLevels, + definitionLevels, + source.getDataEncoding(), + values, + source.getStatistics()); + } + if (source.getCrc().isPresent()) { + decompressedPage.setCrc(source.getCrc().getAsInt()); + } + return decompressedPage; + } + private void setDecompressMetrics(BytesInput bytes, long start) { final ParquetMetricsCallback metricsCallback = options.getMetricsCallback(); if (metricsCallback != null) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java index 9f5bc48f8d..e22dc7427a 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ColumnChunkPageWriteStore.java @@ -319,7 +319,13 @@ public void writePageV2( } if (null != pageBlockEncryptor) { AesCipher.quickUpdatePageAAD(dataPageAAD, pageOrdinal); - compressedData = BytesInput.from(pageBlockEncryptor.encrypt(compressedData.toByteArray(), dataPageAAD)); + // The levels are encrypted with the values as one module, which is written in place of all three + compressedData = BytesInput.from(pageBlockEncryptor.encrypt( + BytesInput.concat(repetitionLevels, definitionLevels, compressedData) + .toByteArray(), + dataPageAAD)); + repetitionLevels = BytesInput.empty(); + definitionLevels = BytesInput.empty(); } int compressedSize = toIntWithCheck(compressedData.size() + repetitionLevels.size() + definitionLevels.size()); diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java index 5a611b41d1..d211159eb4 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileReader.java @@ -2215,34 +2215,7 @@ public ColumnChunkPageReader readAllPages( break; case DATA_PAGE_V2: DataPageHeaderV2 dataHeaderV2 = pageHeader.getData_page_header_v2(); - int dataSize = compressedPageSize - - dataHeaderV2.getRepetition_levels_byte_length() - - dataHeaderV2.getDefinition_levels_byte_length(); - final BytesInput repetitionLevels = - this.readAsBytesInput(dataHeaderV2.getRepetition_levels_byte_length()); - final BytesInput definitionLevels = - this.readAsBytesInput(dataHeaderV2.getDefinition_levels_byte_length()); - final BytesInput values = this.readAsBytesInput(dataSize); - if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { - verifyCrc( - pageHeader.getCrc(), - "could not verify page integrity, CRC checksum verification failed", - repetitionLevels, - definitionLevels, - values); - } - DataPageV2 dataPageV2 = new DataPageV2( - dataHeaderV2.getNum_rows(), - dataHeaderV2.getNum_nulls(), - dataHeaderV2.getNum_values(), - repetitionLevels, - definitionLevels, - converter.getEncoding(dataHeaderV2.getEncoding()), - values, - uncompressedPageSize, - converter.fromParquetStatistics( - getFileMetaData().getCreatedBy(), dataHeaderV2.getStatistics(), type), - dataHeaderV2.isIs_compressed()); + DataPageV2 dataPageV2 = readDataPageV2(pageHeader, type, null != pageBlockDecryptor); // Copy crc to new page, used for testing if (pageHeader.isSetCrc()) { dataPageV2.setCrc(pageHeader.getCrc()); @@ -2294,6 +2267,118 @@ private int getPageOrdinal(int dataPageCountReadSoFar) { return offsetIndex.getPageOrdinal(dataPageCountReadSoFar); } + /** + * Reads the bytes after a DataPageV2 header. For an encrypted column with levels, returns a {@link + * SingleModuleDataPageV2} if the levels are inside the encrypted module, or a {@link DataPageV2} + * with the plaintext levels split out otherwise. + */ + private DataPageV2 readDataPageV2(PageHeader pageHeader, PrimitiveType type, boolean encryptedColumn) + throws IOException { + DataPageHeaderV2 header = pageHeader.getData_page_header_v2(); + int compressedPageSize = pageHeader.getCompressed_page_size(); + int uncompressedPageSize = pageHeader.getUncompressed_page_size(); + int repetitionLevelsByteLength = header.getRepetition_levels_byte_length(); + int definitionLevelsByteLength = header.getDefinition_levels_byte_length(); + org.apache.parquet.column.Encoding encoding = converter.getEncoding(header.getEncoding()); + org.apache.parquet.column.statistics.Statistics statistics = + converter.fromParquetStatistics(getFileMetaData().getCreatedBy(), header.getStatistics(), type); + + if (encryptedColumn && repetitionLevelsByteLength + definitionLevelsByteLength != 0) { + if (repetitionLevelsByteLength < 0 + || definitionLevelsByteLength < 0 + || (long) repetitionLevelsByteLength + definitionLevelsByteLength + > Math.min(compressedPageSize, uncompressedPageSize)) { + throw new ParquetDecodingException("Invalid DataPageV2 header in column " + descriptor.col + + ": repetition levels of " + repetitionLevelsByteLength + + " bytes and definition levels of " + + definitionLevelsByteLength + " bytes in a page of " + compressedPageSize + + " bytes, of " + uncompressedPageSize + " bytes uncompressed"); + } + int levelsByteLength = repetitionLevelsByteLength + definitionLevelsByteLength; + BytesInput pageBytes = this.readAsBytesInput(compressedPageSize); + if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { + verifyCrc( + pageHeader.getCrc(), + "could not verify page integrity, CRC checksum verification failed", + pageBytes); + } + switch (SingleModuleDataPageV2.detectLayout( + pageBytes, levelsByteLength, getFileMetaData().getCreatedBy())) { + case PLAINTEXT_LEVELS: + return pageWithPlaintextLevels(pageHeader, pageBytes, encoding, statistics); + case SINGLE_MODULE: + default: + return singleModulePage(pageHeader, pageBytes, encoding, statistics); + } + } + + int dataSize = compressedPageSize - repetitionLevelsByteLength - definitionLevelsByteLength; + BytesInput repetitionLevels = this.readAsBytesInput(repetitionLevelsByteLength); + BytesInput definitionLevels = this.readAsBytesInput(definitionLevelsByteLength); + BytesInput values = this.readAsBytesInput(dataSize); + if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { + verifyCrc( + pageHeader.getCrc(), + "could not verify page integrity, CRC checksum verification failed", + repetitionLevels, + definitionLevels, + values); + } + return new DataPageV2( + header.getNum_rows(), + header.getNum_nulls(), + header.getNum_values(), + repetitionLevels, + definitionLevels, + encoding, + values, + uncompressedPageSize, + statistics, + header.isIs_compressed()); + } + + /** Splits the plaintext levels off the front of the bytes after a page header. */ + private DataPageV2 pageWithPlaintextLevels( + PageHeader pageHeader, + BytesInput pageBytes, + org.apache.parquet.column.Encoding encoding, + org.apache.parquet.column.statistics.Statistics statistics) + throws IOException { + DataPageHeaderV2 header = pageHeader.getData_page_header_v2(); + ByteBufferInputStream pageStream = pageBytes.toInputStream(); + return new DataPageV2( + header.getNum_rows(), + header.getNum_nulls(), + header.getNum_values(), + BytesInput.from(pageStream.sliceBuffers(header.getRepetition_levels_byte_length())), + BytesInput.from(pageStream.sliceBuffers(header.getDefinition_levels_byte_length())), + encoding, + BytesInput.from(pageStream.remainingBuffers()), + pageHeader.getUncompressed_page_size(), + statistics, + header.isIs_compressed()); + } + + /** Keeps the encrypted module of a page whose levels and values are one encrypted module. */ + private SingleModuleDataPageV2 singleModulePage( + PageHeader pageHeader, + BytesInput pageBytes, + org.apache.parquet.column.Encoding encoding, + org.apache.parquet.column.statistics.Statistics statistics) { + DataPageHeaderV2 header = pageHeader.getData_page_header_v2(); + return new SingleModuleDataPageV2( + header.getNum_rows(), + header.getNum_nulls(), + header.getNum_values(), + encoding, + pageBytes, + header.getRepetition_levels_byte_length(), + header.getDefinition_levels_byte_length(), + pageHeader.getUncompressed_page_size(), + statistics, + header.isIs_compressed()); + } + /** * @param size the size of the page * @return the page diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java index a453d4e756..677a7fd078 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/ParquetFileWriter.java @@ -1163,11 +1163,14 @@ public void writeDataPageV2( statistics, null, null, + null, + null, null); } /** - * Writes a single v2 data page + * Writes a single v2 data page. The levels and data bytes are written as given, without encryption. + * For encrypted columns, use the overload that takes a page block encryptor. * * @param rowCount count of rows * @param nullCount count of nulls @@ -1206,12 +1209,19 @@ public void writeDataPageV2( statistics, null, null, + null, + null, null); } /** * Writes a single v2 data page * + *

Do not use for encrypted columns. If {@code metadataBlockEncryptor} is set, only the page + * header is encrypted by this method: the repetition and definition levels are written in plaintext, + * and the data bytes are written as given. Such pages do not follow the Parquet encryption spec, + * cannot be read by other Parquet implementations, and leave the levels unprotected. + * * @param rowCount count of rows * @param nullCount count of nulls * @param valueCount count of values @@ -1221,10 +1231,12 @@ public void writeDataPageV2( * @param compressedData compressed data bytes * @param uncompressedDataSize the size of uncompressed data * @param statistics the statistics of the page - * @param metadataBlockEncryptor encryptor for block data + * @param metadataBlockEncryptor encryptor for the page header * @param pageHeaderAAD pageHeader AAD * @throws IOException if any I/O error occurs during writing the file - * @deprecated will be removed in 2.0.0. Use {@link ParquetFileWriter#writeDataPageV2(int, int, int, BytesInput, BytesInput, Encoding, BytesInput, boolean, int, Statistics, BlockCipher.Encryptor, byte[])} instead + * @deprecated will be removed in 2.0.0. Use {@link #writeDataPageV2(int, int, int, BytesInput, + * BytesInput, Encoding, BytesInput, boolean, int, Statistics, BlockCipher.Encryptor, byte[], + * SizeStatistics, BlockCipher.Encryptor, byte[])} instead */ @Deprecated public void writeDataPageV2( @@ -1259,6 +1271,11 @@ public void writeDataPageV2( /** * Writes a single v2 data page * + *

Do not use for encrypted columns. If {@code metadataBlockEncryptor} is set, only the page + * header is encrypted by this method: the repetition and definition levels are written in plaintext, + * and the data bytes are written as given. Such pages do not follow the Parquet encryption spec, + * cannot be read by other Parquet implementations, and leave the levels unprotected. + * * @param rowCount count of rows * @param nullCount count of nulls * @param valueCount count of values @@ -1269,10 +1286,14 @@ public void writeDataPageV2( * @param compressed whether the data bytes is compressed * @param uncompressedDataSize the size of uncompressed data * @param statistics the statistics of the page - * @param metadataBlockEncryptor encryptor for block data + * @param metadataBlockEncryptor encryptor for the page header * @param pageHeaderAAD pageHeader AAD * @throws IOException if any I/O error occurs during writing the file + * @deprecated will be removed in 2.0.0. Use {@link #writeDataPageV2(int, int, int, BytesInput, + * BytesInput, Encoding, BytesInput, boolean, int, Statistics, BlockCipher.Encryptor, byte[], + * SizeStatistics, BlockCipher.Encryptor, byte[])} instead */ + @Deprecated public void writeDataPageV2( int rowCount, int nullCount, @@ -1306,6 +1327,11 @@ public void writeDataPageV2( /** * Writes a single v2 data page * + *

Do not use for encrypted columns. If {@code metadataBlockEncryptor} is set, only the page + * header is encrypted by this method: the repetition and definition levels are written in plaintext, + * and the data bytes are written as given. Such pages do not follow the Parquet encryption spec, + * cannot be read by other Parquet implementations, and leave the levels unprotected. + * * @param rowCount count of rows * @param nullCount count of nulls * @param valueCount count of values @@ -1315,11 +1341,13 @@ public void writeDataPageV2( * @param compressedData compressed data bytes * @param uncompressedDataSize the size of uncompressed data * @param statistics the statistics of the page - * @param metadataBlockEncryptor encryptor for block data + * @param metadataBlockEncryptor encryptor for the page header * @param pageHeaderAAD pageHeader AAD * @param sizeStatistics size statistics for the page * @throws IOException if any I/O error occurs during writing the file - * @deprecated will be removed in 2.0.0. Use {@link ParquetFileWriter#writeDataPageV2(int, int, int, BytesInput, BytesInput, Encoding, BytesInput, boolean, int, Statistics, BlockCipher.Encryptor, byte[], SizeStatistics)} instead + * @deprecated will be removed in 2.0.0. Use {@link #writeDataPageV2(int, int, int, BytesInput, + * BytesInput, Encoding, BytesInput, boolean, int, Statistics, BlockCipher.Encryptor, byte[], + * SizeStatistics, BlockCipher.Encryptor, byte[])} instead */ @Deprecated public void writeDataPageV2( @@ -1355,6 +1383,11 @@ public void writeDataPageV2( /** * Writes a single v2 data page * + *

Do not use for encrypted columns. If {@code metadataBlockEncryptor} is set, only the page + * header is encrypted by this method: the repetition and definition levels are written in plaintext, + * and the data bytes are written as given. Such pages do not follow the Parquet encryption spec, + * cannot be read by other Parquet implementations, and leave the levels unprotected. + * * @param rowCount count of rows * @param nullCount count of nulls * @param valueCount count of values @@ -1365,11 +1398,15 @@ public void writeDataPageV2( * @param compressed whether the data bytes is compressed * @param uncompressedDataSize the size of uncompressed data * @param statistics the statistics of the page - * @param metadataBlockEncryptor encryptor for block data + * @param metadataBlockEncryptor encryptor for the page header * @param pageHeaderAAD pageHeader AAD * @param sizeStatistics size statistics for the page * @throws IOException if any I/O error occurs during writing the file + * @deprecated will be removed in 2.0.0. Use {@link #writeDataPageV2(int, int, int, BytesInput, + * BytesInput, Encoding, BytesInput, boolean, int, Statistics, BlockCipher.Encryptor, byte[], + * SizeStatistics, BlockCipher.Encryptor, byte[])} instead */ + @Deprecated public void writeDataPageV2( int rowCount, int nullCount, @@ -1385,16 +1422,126 @@ public void writeDataPageV2( byte[] pageHeaderAAD, SizeStatistics sizeStatistics) throws IOException { + writeDataPageV2Parts( + rowCount, + nullCount, + valueCount, + toIntWithCheck(repetitionLevels.size(), "page repetition levels"), + toIntWithCheck(definitionLevels.size(), "page definition levels"), + repetitionLevels, + definitionLevels, + dataEncoding, + bytes, + compressed, + uncompressedDataSize, + statistics, + metadataBlockEncryptor, + pageHeaderAAD, + sizeStatistics); + } + + /** + * Writes a single v2 data page. If {@code pageBlockEncryptor} is not null, the repetition levels, + * definition levels and data bytes are encrypted together as one module. + * + * @param rowCount count of rows + * @param nullCount count of nulls + * @param valueCount count of values + * @param repetitionLevels repetition level bytes + * @param definitionLevels definition level bytes + * @param dataEncoding encoding for data + * @param bytes data bytes, not yet encrypted + * @param compressed whether the data bytes is compressed + * @param uncompressedDataSize the size of uncompressed data + * @param statistics the statistics of the page + * @param metadataBlockEncryptor encryptor for the page header + * @param pageHeaderAAD pageHeader AAD + * @param sizeStatistics size statistics for the page + * @param pageBlockEncryptor encryptor for the levels and data bytes, or null if the column is not + * encrypted; null if and only if {@code metadataBlockEncryptor} is null + * @param dataPageAAD data page AAD, updated for this page's ordinal; required if + * {@code pageBlockEncryptor} is not null + * @throws IOException if any I/O error occurs during writing the file + */ + public void writeDataPageV2( + int rowCount, + int nullCount, + int valueCount, + BytesInput repetitionLevels, + BytesInput definitionLevels, + Encoding dataEncoding, + BytesInput bytes, + boolean compressed, + int uncompressedDataSize, + Statistics statistics, + BlockCipher.Encryptor metadataBlockEncryptor, + byte[] pageHeaderAAD, + SizeStatistics sizeStatistics, + BlockCipher.Encryptor pageBlockEncryptor, + byte[] dataPageAAD) + throws IOException { + Preconditions.checkArgument( + (metadataBlockEncryptor == null) == (pageBlockEncryptor == null), + "Both the page header and the page of an encrypted column must be encrypted"); + int rlByteLength = toIntWithCheck(repetitionLevels.size(), "page repetition levels"); + int dlByteLength = toIntWithCheck(definitionLevels.size(), "page definition levels"); + if (pageBlockEncryptor != null) { + Preconditions.checkArgument(dataPageAAD != null, "dataPageAAD is required to encrypt the page"); + // The levels are encrypted with the data bytes as one module, which is written in place of all three + bytes = BytesInput.from(pageBlockEncryptor.encrypt( + BytesInput.concat(repetitionLevels, definitionLevels, bytes).toByteArray(), dataPageAAD)); + repetitionLevels = BytesInput.empty(); + definitionLevels = BytesInput.empty(); + } + writeDataPageV2Parts( + rowCount, + nullCount, + valueCount, + rlByteLength, + dlByteLength, + repetitionLevels, + definitionLevels, + dataEncoding, + bytes, + compressed, + uncompressedDataSize, + statistics, + metadataBlockEncryptor, + pageHeaderAAD, + sizeStatistics); + } + + /** + * @param rlByteLength the repetition levels length written in the page header + * @param dlByteLength the definition levels length written in the page header + * @param repetitionLevels the repetition level bytes written after the page header, empty if they + * are encrypted in {@code bytes} + * @param definitionLevels the definition level bytes written after the page header, empty if they + * are encrypted in {@code bytes} + */ + private void writeDataPageV2Parts( + int rowCount, + int nullCount, + int valueCount, + int rlByteLength, + int dlByteLength, + BytesInput repetitionLevels, + BytesInput definitionLevels, + Encoding dataEncoding, + BytesInput bytes, + boolean compressed, + int uncompressedDataSize, + Statistics statistics, + BlockCipher.Encryptor metadataBlockEncryptor, + byte[] pageHeaderAAD, + SizeStatistics sizeStatistics) + throws IOException { withAbortOnFailure(() -> { state = state.write(); - int rlByteLength = toIntWithCheck(repetitionLevels.size(), "page repetition levels"); - int dlByteLength = toIntWithCheck(definitionLevels.size(), "page definition levels"); - int compressedSize = toIntWithCheck(bytes.size() + repetitionLevels.size() + definitionLevels.size(), "page"); - int uncompressedSize = - toIntWithCheck(uncompressedDataSize + repetitionLevels.size() + definitionLevels.size(), "page"); + int uncompressedSize = toIntWithCheck((long) uncompressedDataSize + rlByteLength + dlByteLength, "page"); long beforeHeader = out.getPos(); if (currentChunkFirstDataPage < 0) { diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SingleModuleDataPageV2.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SingleModuleDataPageV2.java new file mode 100644 index 0000000000..825418829f --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/SingleModuleDataPageV2.java @@ -0,0 +1,155 @@ +/* + * 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 org.apache.parquet.hadoop; + +import java.io.IOException; +import org.apache.parquet.SemanticVersion; +import org.apache.parquet.Strings; +import org.apache.parquet.VersionParser; +import org.apache.parquet.VersionParser.ParsedVersion; +import org.apache.parquet.VersionParser.VersionParseException; +import org.apache.parquet.bytes.ByteBufferInputStream; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.page.DataPageV2; +import org.apache.parquet.column.statistics.Statistics; + +/** + * An encrypted DataPageV2 whose levels and values are one encrypted module. Until {@code + * ColumnChunkPageReader} decrypts the module and splits it using the level lengths, {@link + * #getData()} holds the whole module and the levels are empty. + */ +final class SingleModuleDataPageV2 extends DataPageV2 { + + /** How the bytes after the header of an encrypted DataPageV2 page that has levels are laid out. */ + enum Layout { + /** {@code [module of levels and values]} */ + SINGLE_MODULE, + /** {@code [plaintext levels][module of values]} */ + PLAINTEXT_LEVELS + } + + private static final SemanticVersion SINGLE_MODULE_SINCE = new SemanticVersion(1, 19, 0); + + private final int repetitionLevelsByteLength; + private final int definitionLevelsByteLength; + + /** + * @param encryptedModule {@code length (4 bytes, LE) | nonce | ciphertext [| tag]} + * @param uncompressedSize uncompressed page size, levels included + */ + SingleModuleDataPageV2( + int rowCount, + int nullCount, + int valueCount, + Encoding dataEncoding, + BytesInput encryptedModule, + int repetitionLevelsByteLength, + int definitionLevelsByteLength, + int uncompressedSize, + Statistics statistics, + boolean isCompressed) { + super( + rowCount, + nullCount, + valueCount, + BytesInput.empty(), + BytesInput.empty(), + dataEncoding, + encryptedModule, + uncompressedSize, + statistics, + isCompressed); + this.repetitionLevelsByteLength = repetitionLevelsByteLength; + this.definitionLevelsByteLength = definitionLevelsByteLength; + } + + BytesInput getEncryptedModule() { + return getData(); + } + + int getRepetitionLevelsByteLength() { + return repetitionLevelsByteLength; + } + + int getDefinitionLevelsByteLength() { + return definitionLevelsByteLength; + } + + /** + * Detects the layout of the bytes after the header of a page that has levels. A module starts with the length of the + * rest of the module as a 4 byte LE int, so: + * + *

+ * + * If both hold, the layout is the one written by {@code createdBy}. If neither holds, the page is + * corrupt and {@link Layout#SINGLE_MODULE} is returned, so that decryption fails. + * + * @param pageBytes the bytes after the page header, {@code compressed_page_size} of them + * @param levelsByteLength the length of the repetition and definition levels, above zero + * @param createdBy the {@code created_by} of the file + */ + static Layout detectLayout(BytesInput pageBytes, int levelsByteLength, String createdBy) throws IOException { + long size = pageBytes.size(); + boolean singleModule = size >= Integer.BYTES && readLength(pageBytes, 0) == size - Integer.BYTES; + boolean plaintextLevels = size - levelsByteLength >= Integer.BYTES + && readLength(pageBytes, levelsByteLength) == size - levelsByteLength - Integer.BYTES; + if (plaintextLevels && singleModule) { + return layoutWrittenBy(createdBy); + } + return plaintextLevels ? Layout.PLAINTEXT_LEVELS : Layout.SINGLE_MODULE; + } + + /** + * Returns the layout written by the application in {@code created_by}: {@link + * Layout#PLAINTEXT_LEVELS} for parquet-java before 1.19.0, pre-releases of 1.19.0 included, and + * {@link Layout#SINGLE_MODULE} otherwise, including if {@code created_by} is missing or cannot be + * parsed. + */ + static Layout layoutWrittenBy(String createdBy) { + if (Strings.isNullOrEmpty(createdBy)) { + return Layout.SINGLE_MODULE; + } + ParsedVersion writer; + try { + writer = VersionParser.parse(createdBy); + } catch (VersionParseException | RuntimeException e) { + return Layout.SINGLE_MODULE; + } + if ("parquet-mr".equals(writer.application) + && writer.hasSemanticVersion() + && writer.getSemanticVersion().compareTo(SINGLE_MODULE_SINCE) < 0) { + return Layout.PLAINTEXT_LEVELS; + } + return Layout.SINGLE_MODULE; + } + + private static int readLength(BytesInput pageBytes, int offset) throws IOException { + try (ByteBufferInputStream in = pageBytes.toInputStream()) { + in.skipFully(offset); + return BytesUtils.readIntLittleEndian(in); + } + } +} diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java index fce8866850..851b84646c 100644 --- a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/rewrite/ParquetRewriter.java @@ -738,6 +738,7 @@ private void processChunk( BytesInput dlLevels = readBlockAllocate(dlLength, reader); int payLoadLength = pageHeader.getCompressed_page_size() - rlLength - dlLength; int rawDataLength = pageHeader.getUncompressed_page_size() - rlLength - dlLength; + // Not encrypted here: the writer encrypts the levels and the payload together pageLoad = processPageLoad( reader, headerV2.is_compressed, @@ -745,9 +746,9 @@ private void processChunk( decompressor, payLoadLength, rawDataLength, - encryptColumn, - dataEncryptor, - dataPageAAD); + false, + null, + null); statistics = convertStatistics( originalCreatedBy, normalizeNameInType(chunk.getPrimitiveType()), @@ -777,7 +778,10 @@ private void processChunk( rawDataLength, statistics, metaEncryptor, - dataPageHeaderAAD); + dataPageHeaderAAD, + null, + dataEncryptor, + dataPageAAD); pageOrdinal++; break; default: diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestDataPageV2EncryptionInterop.java b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestDataPageV2EncryptionInterop.java new file mode 100644 index 0000000000..e9a3527851 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestDataPageV2EncryptionInterop.java @@ -0,0 +1,747 @@ +/* + * 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 org.apache.parquet.crypto; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.ByteArrayInputStream; +import java.io.File; +import java.io.IOException; +import java.io.RandomAccessFile; +import java.net.URISyntaxException; +import java.nio.charset.StandardCharsets; +import java.nio.file.Files; +import java.util.ArrayList; +import java.util.Arrays; +import java.util.HashMap; +import java.util.List; +import java.util.Map; +import java.util.UUID; +import java.util.stream.Stream; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.HadoopReadOptions; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.bytes.DirectByteBufferAllocator; +import org.apache.parquet.bytes.HeapByteBufferAllocator; +import org.apache.parquet.bytes.TrackingByteBufferAllocator; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +import org.apache.parquet.column.page.PageReadStore; +import org.apache.parquet.column.statistics.Statistics; +import org.apache.parquet.crypto.ModuleCipherFactory.ModuleType; +import org.apache.parquet.crypto.keytools.KeyToolkit; +import org.apache.parquet.crypto.keytools.PropertiesDrivenCryptoFactory; +import org.apache.parquet.crypto.keytools.mocks.Base64KmsClient; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.example.data.simple.convert.GroupRecordConverter; +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.filter2.predicate.FilterApi; +import org.apache.parquet.format.BlockCipher; +import org.apache.parquet.format.FileCryptoMetaData; +import org.apache.parquet.format.PageHeader; +import org.apache.parquet.format.PageType; +import org.apache.parquet.format.Util; +import org.apache.parquet.hadoop.ParquetFileReader; +import org.apache.parquet.hadoop.ParquetFileWriter; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.hadoop.metadata.BlockMetaData; +import org.apache.parquet.hadoop.metadata.ColumnChunkMetaData; +import org.apache.parquet.hadoop.metadata.ColumnPath; +import org.apache.parquet.hadoop.metadata.CompressionCodecName; +import org.apache.parquet.hadoop.metadata.FileMetaData; +import org.apache.parquet.hadoop.metadata.ParquetMetadata; +import org.apache.parquet.hadoop.rewrite.ParquetRewriter; +import org.apache.parquet.hadoop.rewrite.RewriteOptions; +import org.apache.parquet.hadoop.util.HadoopInputFile; +import org.apache.parquet.hadoop.util.HadoopOutputFile; +import org.apache.parquet.io.ColumnIOFactory; +import org.apache.parquet.io.InputFile; +import org.apache.parquet.io.MessageColumnIO; +import org.apache.parquet.io.ParquetDecodingException; +import org.apache.parquet.io.RecordReader; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.io.TempDir; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.Arguments; +import org.junit.jupiter.params.provider.MethodSource; +import org.junit.jupiter.params.provider.ValueSource; + +/** + * Encryption of DataPageV2 pages with repetition or definition levels (apache/parquet-java#3934). + * + *

Fixtures in {@code src/test/resources/encryption/datapagev2}: + * + *

+ */ +public class TestDataPageV2EncryptionInterop { + + private static final String FIXTURES = "/encryption/datapagev2/"; + + private static final int NUM_ROWS = 500; + + /** Footer key, also the key of all columns with uniform encryption. */ + private static final byte[] FOOTER_KEY = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); + + private static final byte[] FOOTER_KEY_METADATA = "footkey".getBytes(StandardCharsets.UTF_8); + + private static final byte[] COLUMN_KEY = "fedcba9876543210".getBytes(StandardCharsets.UTF_8); + + private static final byte[] COLUMN_KEY_METADATA = "colkey".getBytes(StandardCharsets.UTF_8); + + /** + * {@code seq} is required (no levels), {@code id} and {@code name} are optional (definition + * levels), {@code tags} is a list (repetition and definition levels). + */ + private static final MessageType SCHEMA = MessageTypeParser.parseMessageType("message schema {" + + " required int64 seq;" + + " optional int32 id;" + + " optional binary name (STRING);" + + " optional group tags (LIST) {" + + " repeated group list {" + + " optional binary element (STRING);" + + " }" + + " }" + + "}"); + + /** {@code seq >= 250} lets the first two pages (rows 0-199) be skipped. */ + private static final int FIRST_ROW_OF_PAGE_WITH_SEQ_250 = 200; + + private static final FilterCompat.Filter SEQ_AT_LEAST_250 = + FilterCompat.get(FilterApi.gtEq(FilterApi.longColumn("seq"), 250L)); + + @TempDir + private java.nio.file.Path tempDir; + + @BeforeEach + @AfterEach + public void clearKmsClients() { + // KMS clients are cached per JVM by instance id, and the Arrow fixtures use the default id + KeyToolkit.removeCacheEntriesForAllTokens(); + } + + static Stream fixturesAndMemoryModes(String prefix) { + List arguments = new ArrayList<>(); + for (String cipher : new String[] {"aes-gcm-v1", "aes-gcm-ctr-v1"}) { + for (String codec : new String[] {"uncompressed", "snappy"}) { + for (boolean directMemory : new boolean[] {false, true}) { + arguments.add(Arguments.of(prefix + cipher + "-" + codec + ".parquet", directMemory)); + } + } + } + return arguments.stream(); + } + + static Stream arrowFixtures() { + return fixturesAndMemoryModes("arrow-v2-"); + } + + static Stream legacyFixtures() { + return fixturesAndMemoryModes("java-legacy-v2-"); + } + + static Stream legacyFixtureNames() { + return Stream.of("aes-gcm-v1", "aes-gcm-ctr-v1") + .flatMap(cipher -> Stream.of("uncompressed", "snappy") + .map(codec -> "java-legacy-v2-" + cipher + "-" + codec + ".parquet")); + } + + static Stream ambiguousFixtures() { + return Stream.of("aes-gcm-v1", "aes-gcm-ctr-v1") + .flatMap(cipher -> Stream.of(false, true) + .map(directMemory -> Arguments.of("legacy-v2-ambiguous-" + cipher + ".parquet", directMemory))); + } + + static Stream ciphersAndCodecs() { + List arguments = new ArrayList<>(); + for (ParquetCipher cipher : ParquetCipher.values()) { + for (CompressionCodecName codec : + new CompressionCodecName[] {CompressionCodecName.UNCOMPRESSED, CompressionCodecName.SNAPPY}) { + arguments.add(Arguments.of(cipher, codec)); + } + } + return arguments.stream(); + } + + @ParameterizedTest(name = "{0}, directMemory={1}") + @MethodSource("arrowFixtures") + public void readsFilesWrittenByArrow(String fixture, boolean directMemory) throws Exception { + Configuration conf = new Configuration(); + conf.set( + EncryptionPropertiesFactory.CRYPTO_FACTORY_CLASS_PROPERTY_NAME, + PropertiesDrivenCryptoFactory.class.getName()); + conf.set(KeyToolkit.KMS_CLIENT_CLASS_PROPERTY_NAME, Base64KmsClient.class.getName()); + + assertThat(readRows(fixture(fixture), conf, null, directMemory)).containsExactlyElementsOf(expectedRows()); + } + + @ParameterizedTest(name = "{0}, directMemory={1}") + @MethodSource("legacyFixtures") + public void readsFilesWrittenByEarlierParquetJavaVersions(String fixture, boolean directMemory) throws Exception { + assertThat(readRows(fixture(fixture), new Configuration(), decryptionProperties(), directMemory)) + .containsExactlyElementsOf(expectedRows()); + } + + @ParameterizedTest(name = "{0}") + @MethodSource("legacyFixtureNames") + public void legacyFixturesHaveThePlaintextLevelsLayout(String fixture) throws Exception { + List pages = decodeDataPagesV2(fixture(fixture), FOOTER_KEY, false); + + assertThat(pages).isNotEmpty(); + assertThat(pages).allSatisfy(page -> assertThat(page.singleModule).isFalse()); + } + + @ParameterizedTest(name = "{0}, {1}") + @MethodSource("ciphersAndCodecs") + public void writesTheLevelsInTheEncryptedPageModule(ParquetCipher cipher, CompressionCodecName codec) + throws Exception { + Path file = writeWithParquetJava(cipher, codec); + + List pages = decodeDataPagesV2(file, FOOTER_KEY, false); + + assertThat(pages).isNotEmpty(); + assertThat(pages).extracting(page -> page.column).containsOnly("id", "name", "tags.list.element"); + assertThat(pages) + .anySatisfy(page -> assertThat(page.repetitionLevelsLength).isPositive()); + for (DecodedPage page : pages) { + assertThat(page.singleModule) + .as("the levels of %s must be inside the encrypted page module", page) + .isTrue(); + if (codec == CompressionCodecName.UNCOMPRESSED) { + assertThat(page.decryptedLength).isEqualTo(page.uncompressedPageSize); + } + } + } + + @Test + public void detectsTamperingWithTheLevels() throws Exception { + Path file = writeFileWithTamperedLevels(); + // Otherwise the page checksum fails first + Configuration withoutChecksums = new Configuration(); + withoutChecksums.setBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, false); + + assertThatThrownBy(() -> readRows(file, withoutChecksums, decryptionProperties(), false)) + .isInstanceOf(ParquetDecodingException.class) + .hasMessageContaining("Can not read value at 0 in block -1") + .cause() + .isInstanceOf(TagVerificationException.class) + .hasMessage("GCM tag check failed"); + } + + @Test + public void pageChecksumDetectsTamperingWithTheLevels() throws Exception { + Path file = writeFileWithTamperedLevels(); + + assertThatThrownBy(() -> readRows(file, new Configuration(), decryptionProperties(), false)) + .isInstanceOf(ParquetDecodingException.class) + .hasMessageContaining("Can not read value at 0 in block -1") + .cause() + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("could not verify page integrity, CRC checksum verification failed"); + } + + /** Writes a file with one bit of the levels of the first DataPageV2 page flipped. */ + private Path writeFileWithTamperedLevels() throws Exception { + Path file = writeWithParquetJava(ParquetCipher.AES_GCM_V1, CompressionCodecName.UNCOMPRESSED); + DecodedPage page = decodeDataPagesV2(file, FOOTER_KEY, false).get(0); + + // First ciphertext byte, which is the first byte of the levels + File source = new File(file.toUri()); + try (RandomAccessFile raf = new RandomAccessFile(source, "rw")) { + long position = page.pageBytesOffset + Integer.BYTES + AesCipher.NONCE_LENGTH; + raf.seek(position); + int original = raf.read(); + raf.seek(position); + raf.write(original ^ 0x01); + } + // Otherwise the local filesystem rejects the file + Files.deleteIfExists(new File(source.getParent(), "." + source.getName() + ".crc").toPath()); + return file; + } + + @ParameterizedTest(name = "{0}, {1}") + @MethodSource("ciphersAndCodecs") + public void rewriterWritesTheLevelsInTheEncryptedPageModule(ParquetCipher cipher, CompressionCodecName codec) + throws Exception { + Path plaintext = writeWithParquetJava(null, codec); + Path rewritten = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + + String[] encryptColumns = {"id", "name", "tags.list.element"}; + Map columns = new HashMap<>(); + for (String column : encryptColumns) { + ColumnPath path = ColumnPath.fromDotString(column); + columns.put( + path, + ColumnEncryptionProperties.builder(path) + .withKey(COLUMN_KEY) + .withKeyMetaData(COLUMN_KEY_METADATA) + .build()); + } + FileEncryptionProperties encryption = FileEncryptionProperties.builder(FOOTER_KEY) + .withFooterKeyMetadata(FOOTER_KEY_METADATA) + .withAlgorithm(cipher) + .withEncryptedColumns(columns) + .build(); + RewriteOptions options = new RewriteOptions.Builder(new Configuration(), plaintext, rewritten) + .encrypt(Arrays.asList(encryptColumns)) + .encryptionProperties(encryption) + .build(); + try (ParquetRewriter rewriter = new ParquetRewriter(options)) { + rewriter.processBlocks(); + } + + List pages = decodeDataPagesV2(rewritten, COLUMN_KEY, true); + assertThat(pages).isNotEmpty(); + assertThat(pages).extracting(page -> page.column).containsOnly("id", "name", "tags.list.element"); + assertThat(pages) + .anySatisfy(page -> assertThat(page.repetitionLevelsLength).isPositive()); + assertThat(pages).allSatisfy(page -> assertThat(page.singleModule).isTrue()); + assertThat(readRows(rewritten, new Configuration(), decryptionProperties(), false)) + .containsExactlyElementsOf(expectedRows()); + } + + @ParameterizedTest(name = "{0}, {1}") + @MethodSource("ciphersAndCodecs") + public void readsWhatItWritesInSeveralRowGroups(ParquetCipher cipher, CompressionCodecName codec) throws Exception { + Path file = writeWithParquetJava(cipher, codec, 4096); + + try (ParquetFileReader reader = ParquetFileReader.open( + HadoopInputFile.fromPath(file, new Configuration()), + ParquetReadOptions.builder() + .withDecryption(decryptionProperties()) + .build())) { + assertThat(reader.getFooter().getBlocks()).hasSizeGreaterThan(1); + } + assertThat(readRows(file, new Configuration(), decryptionProperties(), false)) + .containsExactlyElementsOf(expectedRows()); + assertThat(readRows(file, new Configuration(), decryptionProperties(), true)) + .containsExactlyElementsOf(expectedRows()); + } + + @ParameterizedTest(name = "{0}, {1}") + @MethodSource("ciphersAndCodecs") + public void skipsPagesOfWhatItWrites(ParquetCipher cipher, CompressionCodecName codec) throws Exception { + Path file = writeWithParquetJava(cipher, codec); + + assertThat(readRows(file, new Configuration(), decryptionProperties(), false, SEQ_AT_LEAST_250)) + .containsExactlyElementsOf(expectedRows().subList(FIRST_ROW_OF_PAGE_WITH_SEQ_250, NUM_ROWS)); + assertThat(readRows(file, new Configuration(), decryptionProperties(), true, SEQ_AT_LEAST_250)) + .containsExactlyElementsOf(expectedRows().subList(FIRST_ROW_OF_PAGE_WITH_SEQ_250, NUM_ROWS)); + } + + @ParameterizedTest(name = "{0}, directMemory={1}") + @MethodSource("legacyFixtures") + public void skipsPagesOfFilesWrittenByEarlierParquetJavaVersions(String fixture, boolean directMemory) + throws Exception { + assertThat(readRows( + fixture(fixture), new Configuration(), decryptionProperties(), directMemory, SEQ_AT_LEAST_250)) + .containsExactlyElementsOf(expectedRows().subList(FIRST_ROW_OF_PAGE_WITH_SEQ_250, NUM_ROWS)); + } + + /** + * {@code legacy-v2-ambiguous-*}: one legacy page of 8 optional int32 values, only the first (12345) + * defined. Its 4 bytes of definition levels, {@code 03 01 00 00}, are a bit-packed run of 8 levels + * plus 2 unread bytes, and read as an LE int they equal {@code compressed_page_size - 4}, so the + * page fits both layouts. The {@code created_by} of the fixture, {@code parquet-mr version + * 1.19.0-SNAPSHOT}, resolves it to plaintext levels. + */ + @ParameterizedTest(name = "{0}, directMemory={1}") + @MethodSource("ambiguousFixtures") + public void readsALegacyPageThatLooksLikeASingleModule(String fixture, boolean directMemory) throws Exception { + List rows = readRows(fixture(fixture), new Configuration(), decryptionProperties(), directMemory); + + assertThat(rows).hasSize(8); + assertThat(rows.get(0)).contains("v: 12345"); + assertThat(rows.subList(1, 8)).allSatisfy(row -> assertThat(row).doesNotContain("v:")); + } + + @ParameterizedTest(name = "directMemory={0}") + @ValueSource(booleans = {false, true}) + public void readsAnAmbiguousPageAsASingleModuleIfTheFileIsFromAnotherWriter(boolean directMemory) { + // The fixture has plaintext levels, so reading it as a single module fails the GCM tag check + assertThatThrownBy(() -> readRowsWrittenBy( + fixture("legacy-v2-ambiguous-aes-gcm-v1.parquet"), + "parquet-cpp-arrow version 24.0.0", + directMemory)) + .isInstanceOf(TagVerificationException.class) + .hasMessage("GCM tag check failed"); + } + + @Test + public void writerRejectsAnEncryptedPageHeaderWithoutAnEncryptedPage() throws Exception { + Path file = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + try (ParquetFileWriter writer = new ParquetFileWriter( + HadoopOutputFile.fromPath(file, new Configuration()), + SCHEMA, + ParquetFileWriter.Mode.CREATE, + ParquetWriter.DEFAULT_BLOCK_SIZE, + ParquetWriter.MAX_PADDING_SIZE_DEFAULT, + null, + ParquetProperties.builder().build())) { + BlockCipher.Encryptor headerEncryptor = ModuleCipherFactory.getEncryptor(AesMode.GCM, FOOTER_KEY); + + assertThatThrownBy(() -> writer.writeDataPageV2( + 1, + 0, + 1, + BytesInput.empty(), + BytesInput.from(new byte[] {0x02, 0x01}), + Encoding.PLAIN, + BytesInput.from(new byte[4]), + false, + 4, + Statistics.createStats(SCHEMA.getType("id").asPrimitiveType()), + headerEncryptor, + new byte[0], + null, + null, + null)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessage("Both the page header and the page of an encrypted column must be encrypted"); + } + } + + /** Writes 100 rows per page; encrypts all columns with the footer key if {@code cipher} is set. */ + private Path writeWithParquetJava(ParquetCipher cipher, CompressionCodecName codec) throws IOException { + return writeWithParquetJava(cipher, codec, ParquetWriter.DEFAULT_BLOCK_SIZE); + } + + private Path writeWithParquetJava(ParquetCipher cipher, CompressionCodecName codec, long rowGroupSize) + throws IOException { + Path file = new Path(tempDir.resolve(UUID.randomUUID().toString()).toUri()); + ExampleParquetWriter.Builder builder = ExampleParquetWriter.builder(file) + .withType(SCHEMA) + .withWriterVersion(WriterVersion.PARQUET_2_0) + .withCompressionCodec(codec) + .withRowGroupSize(rowGroupSize) + .withMinRowCountForPageSizeCheck(10) + .withPageRowCountLimit(100); + if (cipher != null) { + builder.withEncryption(FileEncryptionProperties.builder(FOOTER_KEY) + .withFooterKeyMetadata(FOOTER_KEY_METADATA) + .withAlgorithm(cipher) + .build()); + } + try (ParquetWriter writer = builder.build()) { + for (int i = 0; i < NUM_ROWS; i++) { + writer.write(row(i)); + } + } + return file; + } + + /** Builds row {@code i}, as stored in the {@code arrow-v2-*} fixtures. */ + private static Group row(int i) { + Group group = new SimpleGroupFactory(SCHEMA).newGroup(); + group.add("seq", (long) i); + if (i % 7 != 0) { + group.add("id", i); + } + if (i % 5 != 0) { + group.add("name", "name-" + i); + } + if (i % 4 != 0) { + Group tags = group.addGroup("tags"); + int count = i % 4; + for (int j = 0; j < count; j++) { + Group element = tags.addGroup("list"); + if ((i + j) % 3 != 0) { + element.add("element", "tag-" + i + "-" + j); + } + } + } + return group; + } + + private static List expectedRows() { + List rows = new ArrayList<>(NUM_ROWS); + for (int i = 0; i < NUM_ROWS; i++) { + rows.add(row(i).toString()); + } + return rows; + } + + private static FileDecryptionProperties decryptionProperties() { + return FileDecryptionProperties.builder() + .withKeyRetriever(new DecryptionKeyRetrieverMock() + .putKey("footkey", FOOTER_KEY) + .putKey("colkey", COLUMN_KEY)) + .build(); + } + + /** + * Reads all rows as strings. Page checksums are verified unless {@code conf} disables it. + * + * @param conf not modified + * @param decryption the decryption properties, or null to use the ones configured in {@code conf} + * @param directMemory whether to read into direct buffers and decrypt off heap + */ + private static List readRows( + Path file, Configuration conf, FileDecryptionProperties decryption, boolean directMemory) + throws IOException { + return readRows(file, conf, decryption, directMemory, FilterCompat.NOOP); + } + + /** Like {@link #readRows(Path, Configuration, FileDecryptionProperties, boolean)}, filtered by page. */ + private static List readRows( + Path file, + Configuration conf, + FileDecryptionProperties decryption, + boolean directMemory, + FilterCompat.Filter filter) + throws IOException { + Configuration readConf = new Configuration(conf); + readConf.setBoolean(ParquetInputFormat.OFF_HEAP_DECRYPT_BUFFER_ENABLED, directMemory); + readConf.setBoolean( + ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, + conf.getBoolean(ParquetInputFormat.PAGE_VERIFY_CHECKSUM_ENABLED, true)); + if (filter != FilterCompat.NOOP) { + // Return every row of the pages that are read, so the test can see which pages were skipped + readConf.setBoolean(ParquetInputFormat.RECORD_FILTERING_ENABLED, false); + } + List rows = new ArrayList<>(); + // close() fails on leaked buffers + try (TrackingByteBufferAllocator allocator = TrackingByteBufferAllocator.wrap( + directMemory ? new DirectByteBufferAllocator() : new HeapByteBufferAllocator())) { + ParquetReader.Builder builder = ParquetReader.builder(new GroupReadSupport(), file) + .withConf(readConf) + .withFilter(filter) + .withAllocator(allocator); + if (decryption != null) { + builder.withDecryption(decryption); + } + try (ParquetReader reader = builder.build()) { + Group group; + while ((group = reader.read()) != null) { + rows.add(group.toString()); + } + } + } + return rows; + } + + /** Reads all rows as strings, with {@code createdBy} in place of the {@code created_by} of the file. */ + private static List readRowsWrittenBy(Path file, String createdBy, boolean directMemory) + throws IOException { + Configuration conf = new Configuration(); + InputFile inputFile = HadoopInputFile.fromPath(file, conf); + List rows = new ArrayList<>(); + // close() fails on leaked buffers + try (TrackingByteBufferAllocator allocator = TrackingByteBufferAllocator.wrap( + directMemory ? new DirectByteBufferAllocator() : new HeapByteBufferAllocator())) { + ParquetReadOptions options = HadoopReadOptions.builder(conf, file) + .withDecryption(decryptionProperties()) + .withAllocator(allocator) + .useOffHeapDecryptBuffer(directMemory) + .build(); + ParquetMetadata footer; + try (ParquetFileReader reader = ParquetFileReader.open(inputFile, options)) { + footer = reader.getFooter(); + } + FileMetaData fileMetaData = footer.getFileMetaData(); + ParquetMetadata writtenBy = new ParquetMetadata( + new FileMetaData( + fileMetaData.getSchema(), + fileMetaData.getKeyValueMetaData(), + createdBy, + fileMetaData.getEncryptionType(), + fileMetaData.getFileDecryptor()), + footer.getBlocks()); + MessageType schema = fileMetaData.getSchema(); + MessageColumnIO columnIO = new ColumnIOFactory().getColumnIO(schema); + try (ParquetFileReader reader = + new ParquetFileReader(inputFile, writtenBy, options, inputFile.newStream())) { + PageReadStore rowGroup; + while ((rowGroup = reader.readNextRowGroup()) != null) { + RecordReader records = columnIO.getRecordReader(rowGroup, new GroupRecordConverter(schema)); + for (long i = 0; i < rowGroup.getRowCount(); i++) { + rows.add(records.read().toString()); + } + rowGroup.close(); + } + } + } + return rows; + } + + private static Path fixture(String name) throws URISyntaxException { + return new Path(TestDataPageV2EncryptionInterop.class + .getResource(FIXTURES + name) + .toURI()); + } + + /** A DataPageV2 with levels, as found by {@link #decodeDataPagesV2}. */ + private static final class DecodedPage { + private final String column; + private final int repetitionLevelsLength; + private final int definitionLevelsLength; + private final long pageBytesOffset; + private final boolean singleModule; + private final int decryptedLength; + private final int uncompressedPageSize; + + private DecodedPage( + String column, + int repetitionLevelsLength, + int definitionLevelsLength, + long pageBytesOffset, + boolean singleModule, + int decryptedLength, + int uncompressedPageSize) { + this.column = column; + this.repetitionLevelsLength = repetitionLevelsLength; + this.definitionLevelsLength = definitionLevelsLength; + this.pageBytesOffset = pageBytesOffset; + this.singleModule = singleModule; + this.decryptedLength = decryptedLength; + this.uncompressedPageSize = uncompressedPageSize; + } + + private int levelsLength() { + return repetitionLevelsLength + definitionLevelsLength; + } + + @Override + public String toString() { + return "DataPageV2 of column " + column + " at offset " + pageBytesOffset + " with " + levelsLength() + + " bytes of levels"; + } + } + + /** + * Decodes the DataPageV2 pages of a file without the parquet-java page reader: parses and decrypts + * the page headers, then decrypts the bytes after each header as one module if their first 4 bytes + * are {@code size - 4}, otherwise the bytes after the levels. + * + * @param file file with an encrypted footer, whose encrypted columns all use {@code columnKey} + * @param onlyEncryptedColumns whether to skip columns that are not encrypted + * @return the pages that have levels, in file order + */ + private static List decodeDataPagesV2(Path file, byte[] columnKey, boolean onlyEncryptedColumns) + throws IOException { + byte[] bytes = Files.readAllBytes(new File(file.toUri()).toPath()); + int footerLength = BytesUtils.readIntLittleEndian(bytes, bytes.length - 8); + FileCryptoMetaData cryptoMetaData = Util.readFileCryptoMetaData( + new ByteArrayInputStream(bytes, bytes.length - 8 - footerLength, footerLength)); + org.apache.parquet.format.EncryptionAlgorithm algorithm = cryptoMetaData.getEncryption_algorithm(); + boolean ctr = algorithm.isSetAES_GCM_CTR_V1(); + byte[] fileAAD = (ctr + ? algorithm.getAES_GCM_CTR_V1().getAad_file_unique() + : algorithm.getAES_GCM_V1().getAad_file_unique()); + + BlockCipher.Decryptor headerDecryptor = ModuleCipherFactory.getDecryptor(AesMode.GCM, columnKey); + BlockCipher.Decryptor pageDecryptor = + ModuleCipherFactory.getDecryptor(ctr ? AesMode.CTR : AesMode.GCM, columnKey); + + List decoded = new ArrayList<>(); + try (ParquetFileReader reader = ParquetFileReader.open( + HadoopInputFile.fromPath(file, new Configuration()), + ParquetReadOptions.builder() + .withDecryption(decryptionProperties()) + .build())) { + for (BlockMetaData block : reader.getFooter().getBlocks()) { + for (int columnOrdinal = 0; columnOrdinal < block.getColumns().size(); columnOrdinal++) { + ColumnChunkMetaData chunk = block.getColumns().get(columnOrdinal); + if (onlyEncryptedColumns && !chunk.isEncrypted()) { + continue; + } + ByteArrayInputStream in = new ByteArrayInputStream(bytes); + in.skip(chunk.getStartingPos()); + boolean dictionaryPageRead = !chunk.hasDictionaryPage(); + int pageOrdinal = 0; + long valuesRead = 0; + while (valuesRead < chunk.getValueCount()) { + byte[] headerAAD = dictionaryPageRead + ? AesCipher.createModuleAAD( + fileAAD, + ModuleType.DataPageHeader, + block.getOrdinal(), + columnOrdinal, + pageOrdinal) + : AesCipher.createModuleAAD( + fileAAD, + ModuleType.DictionaryPageHeader, + block.getOrdinal(), + columnOrdinal, + -1); + PageHeader header = Util.readPageHeader(in, headerDecryptor, headerAAD); + long pageBytesOffset = bytes.length - in.available(); + byte[] pageBytes = in.readNBytes(header.getCompressed_page_size()); + assertThat(pageBytes).hasSize(header.getCompressed_page_size()); + if (!dictionaryPageRead) { + dictionaryPageRead = true; + continue; + } + if (header.getType() == PageType.DATA_PAGE_V2) { + int repetitionLevels = + header.getData_page_header_v2().getRepetition_levels_byte_length(); + int definitionLevels = + header.getData_page_header_v2().getDefinition_levels_byte_length(); + int levels = repetitionLevels + definitionLevels; + boolean singleModule = + BytesUtils.readIntLittleEndian(pageBytes, 0) == pageBytes.length - Integer.BYTES; + byte[] module = + singleModule ? pageBytes : Arrays.copyOfRange(pageBytes, levels, pageBytes.length); + assertThat(BytesUtils.readIntLittleEndian(module, 0)) + .isEqualTo(module.length - Integer.BYTES); + byte[] pageAAD = AesCipher.createModuleAAD( + fileAAD, ModuleType.DataPage, block.getOrdinal(), columnOrdinal, pageOrdinal); + byte[] decrypted = pageDecryptor.decrypt(module, pageAAD); + if (levels > 0) { + decoded.add(new DecodedPage( + chunk.getPath().toDotString(), + repetitionLevels, + definitionLevels, + pageBytesOffset, + singleModule, + decrypted.length, + header.getUncompressed_page_size())); + } + valuesRead += header.getData_page_header_v2().getNum_values(); + } + pageOrdinal++; + } + } + } + } + return decoded; + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/crypto/keytools/mocks/Base64KmsClient.java b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/keytools/mocks/Base64KmsClient.java new file mode 100644 index 0000000000..7b0cb0974a --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/keytools/mocks/Base64KmsClient.java @@ -0,0 +1,43 @@ +/* + * 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 org.apache.parquet.crypto.keytools.mocks; + +import java.util.Base64; +import org.apache.hadoop.conf.Configuration; +import org.apache.parquet.crypto.KeyAccessDeniedException; +import org.apache.parquet.crypto.keytools.KmsClient; + +/** Wraps keys by base64 encoding them, like the KMS client that wrote the {@code arrow-v2-*} fixtures. */ +public class Base64KmsClient implements KmsClient { + + @Override + public void initialize(Configuration configuration, String kmsInstanceID, String kmsInstanceURL, String accessToken) + throws KeyAccessDeniedException {} + + @Override + public String wrapKey(byte[] keyBytes, String masterKeyIdentifier) throws KeyAccessDeniedException { + return Base64.getEncoder().encodeToString(keyBytes); + } + + @Override + public byte[] unwrapKey(String wrappedKey, String masterKeyIdentifier) throws KeyAccessDeniedException { + return Base64.getDecoder().decode(wrappedKey); + } +} diff --git a/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSingleModuleDataPageV2.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSingleModuleDataPageV2.java new file mode 100644 index 0000000000..cad6f3a661 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestSingleModuleDataPageV2.java @@ -0,0 +1,236 @@ +/* + * 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 org.apache.parquet.hadoop; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +import java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import java.nio.charset.StandardCharsets; +import java.util.Collections; +import org.apache.parquet.ParquetReadOptions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.page.DataPage; +import org.apache.parquet.column.page.DataPageV2; +import org.apache.parquet.crypto.AesCipher; +import org.apache.parquet.crypto.AesMode; +import org.apache.parquet.crypto.ModuleCipherFactory; +import org.apache.parquet.crypto.ModuleCipherFactory.ModuleType; +import org.apache.parquet.hadoop.ColumnChunkPageReadStore.ColumnChunkPageReader; +import org.apache.parquet.hadoop.SingleModuleDataPageV2.Layout; +import org.apache.parquet.io.ParquetDecodingException; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.params.ParameterizedTest; +import org.junit.jupiter.params.provider.CsvSource; + +public class TestSingleModuleDataPageV2 { + + private static final String PARQUET_JAVA_1_18 = "parquet-mr version 1.18.0 (build abc)"; + private static final String ARROW = "parquet-cpp-arrow version 24.0.0"; + + /** A bit-packed run of 8 levels, then an RLE run of 8 levels. */ + private static final byte[] LEVELS = {0x03, (byte) 0xff, 0x10, 0x01}; + + /** {@code contentLength} as a 4 byte LE int, followed by that many bytes. */ + private static byte[] module(int contentLength) { + ByteBuffer module = ByteBuffer.allocate(Integer.BYTES + contentLength).order(ByteOrder.LITTLE_ENDIAN); + module.putInt(contentLength); + for (int i = 0; i < contentLength; i++) { + module.put((byte) (0x80 + i)); + } + return module.array(); + } + + private static byte[] concat(byte[] first, byte[] second) { + byte[] result = new byte[first.length + second.length]; + System.arraycopy(first, 0, result, 0, first.length); + System.arraycopy(second, 0, result, first.length, second.length); + return result; + } + + @Test + public void pageThatIsOneModuleIsASingleModule() throws IOException { + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.from(module(50)), LEVELS.length, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void pageWithPlaintextLevelsInFrontOfAModuleHasPlaintextLevels() throws IOException { + byte[] pageBytes = concat(LEVELS, module(50)); + + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.from(pageBytes), LEVELS.length, ARROW)) + .isEqualTo(Layout.PLAINTEXT_LEVELS); + } + + @Test + public void pageThatIsOneModuleIsRecognizedWhenItIsNotInOneBuffer() throws IOException { + byte[] pageBytes = module(300); + ByteBuffer first = ByteBuffer.wrap(pageBytes, 0, 2).slice(); + ByteBuffer second = ByteBuffer.wrap(pageBytes, 2, pageBytes.length - 2).slice(); + + assertThat(SingleModuleDataPageV2.detectLayout( + BytesInput.from(first, second), LEVELS.length, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void pageWithPlaintextLevelsIsRecognizedWhenTheModuleLengthIsNotInOneBuffer() throws IOException { + byte[] pageBytes = concat(LEVELS, module(300)); + // Split inside the module length + ByteBuffer first = ByteBuffer.wrap(pageBytes, 0, LEVELS.length + 2).slice(); + ByteBuffer second = ByteBuffer.wrap(pageBytes, LEVELS.length + 2, pageBytes.length - LEVELS.length - 2) + .slice(); + + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.from(first, second), LEVELS.length, ARROW)) + .isEqualTo(Layout.PLAINTEXT_LEVELS); + } + + /** Levels whose first 4 bytes are {@code pageSize - 4}, followed by a module. */ + private static byte[] pageThatFitsBothLayouts() { + byte[] module = module(50); + int pageSize = LEVELS.length + module.length; + byte[] levels = ByteBuffer.allocate(LEVELS.length) + .order(ByteOrder.LITTLE_ENDIAN) + .putInt(pageSize - Integer.BYTES) + .array(); + return concat(levels, module); + } + + @Test + public void pageThatFitsBothLayoutsHasPlaintextLevelsIfWrittenByParquetJava118() throws IOException { + assertThat(SingleModuleDataPageV2.detectLayout( + BytesInput.from(pageThatFitsBothLayouts()), LEVELS.length, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.PLAINTEXT_LEVELS); + } + + @Test + public void pageThatFitsBothLayoutsIsASingleModuleIfWrittenByAnotherWriter() throws IOException { + assertThat(SingleModuleDataPageV2.detectLayout( + BytesInput.from(pageThatFitsBothLayouts()), LEVELS.length, ARROW)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void pageThatFitsNeitherLayoutIsTreatedAsASingleModuleSoThatDecryptionReportsIt() throws IOException { + byte[] pageBytes = concat(LEVELS, module(50)); + pageBytes[LEVELS.length] ^= 0x01; + + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.from(pageBytes), LEVELS.length, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void pageTooShortForAModuleLengthIsASingleModule() throws IOException { + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.empty(), 1, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.SINGLE_MODULE); + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.from(new byte[] {1, 2, 3}), 1, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.SINGLE_MODULE); + // Page is only levels + assertThat(SingleModuleDataPageV2.detectLayout(BytesInput.from(LEVELS), LEVELS.length, PARQUET_JAVA_1_18)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @ParameterizedTest + @CsvSource({ + "'parquet-mr version 1.18.0 (build abc)', PLAINTEXT_LEVELS", + "'parquet-mr version 1.19.0-SNAPSHOT (build abc)', PLAINTEXT_LEVELS", + "'parquet-mr version 1.19.0 (build abc)', SINGLE_MODULE", + "'parquet-cpp-arrow version 24.0.0', SINGLE_MODULE", + "'parquet-mr version not-a-version', SINGLE_MODULE", + "'not a created_by', SINGLE_MODULE", + "'', SINGLE_MODULE", + }) + public void layoutWrittenByFollowsTheWriterVersion(String createdBy, Layout expected) { + assertThat(SingleModuleDataPageV2.layoutWrittenBy(createdBy)).isEqualTo(expected); + } + + @Test + public void layoutWrittenByAnUnknownWriterIsASingleModule() { + assertThat(SingleModuleDataPageV2.layoutWrittenBy(null)).isEqualTo(Layout.SINGLE_MODULE); + } + + private static DataPage readUncompressedPage(int uncompressedPageSize) { + return readUncompressedPage(LEVELS.length, uncompressedPageSize); + } + + /** + * Reads one uncompressed page whose decrypted module is {@code LEVELS} and 8 value bytes, and whose + * header gives {@code definitionLevelsByteLength} bytes of definition levels. + */ + private static DataPage readUncompressedPage(int definitionLevelsByteLength, int uncompressedPageSize) { + byte[] key = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); + byte[] fileAAD = "file".getBytes(StandardCharsets.UTF_8); + byte[] plainPage = concat(LEVELS, new byte[8]); + byte[] dataPageAAD = AesCipher.createModuleAAD(fileAAD, ModuleType.DataPage, 0, 0, 0); + byte[] encryptedModule = + ModuleCipherFactory.getEncryptor(AesMode.GCM, key).encrypt(plainPage, dataPageAAD); + SingleModuleDataPageV2 page = new SingleModuleDataPageV2( + 8, + 0, + 8, + Encoding.PLAIN, + BytesInput.from(encryptedModule), + 0, + definitionLevelsByteLength, + uncompressedPageSize, + null, + false); + ColumnChunkPageReader reader = new ColumnChunkPageReader( + null, + Collections.singletonList(page), + null, + null, + 8, + ModuleCipherFactory.getDecryptor(AesMode.GCM, key), + fileAAD, + 0, + 0, + ParquetReadOptions.builder().build()); + return reader.readPage(); + } + + @Test + public void uncompressedPageWithTheSizeOfItsDecryptedModuleIsRead() { + DataPageV2 page = (DataPageV2) readUncompressedPage(LEVELS.length + 8); + + assertThat(page.getDefinitionLevels().size()).isEqualTo(LEVELS.length); + assertThat(page.getData().size()).isEqualTo(8); + } + + @Test + public void uncompressedPageWhoseHeaderDisagreesWithItsDecryptedModuleIsRejected() { + assertThatThrownBy(() -> readUncompressedPage(LEVELS.length + 9)) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage( + "Decrypted DataPageV2 page has 12 bytes, but the page header gives an uncompressed page size" + + " of 13 bytes"); + } + + @Test + public void pageWhoseDecryptedModuleIsShorterThanItsLevelsIsRejected() { + assertThatThrownBy(() -> readUncompressedPage(LEVELS.length + 9, LEVELS.length + 9)) + .isInstanceOf(ParquetDecodingException.class) + .hasMessage("Decrypted DataPageV2 page has 12 bytes, which is less than the 13 bytes of repetition" + + " and definition levels in the page header"); + } +} diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-ctr-v1-snappy.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-ctr-v1-snappy.parquet new file mode 100644 index 0000000000..0e56362f2e Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-ctr-v1-snappy.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-ctr-v1-uncompressed.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-ctr-v1-uncompressed.parquet new file mode 100644 index 0000000000..43ecb0d14e Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-ctr-v1-uncompressed.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-v1-snappy.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-v1-snappy.parquet new file mode 100644 index 0000000000..c7f7d983df Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-v1-snappy.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-v1-uncompressed.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-v1-uncompressed.parquet new file mode 100644 index 0000000000..696840fdad Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/arrow-v2-aes-gcm-v1-uncompressed.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-ctr-v1-snappy.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-ctr-v1-snappy.parquet new file mode 100644 index 0000000000..d99a49f92c Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-ctr-v1-snappy.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-ctr-v1-uncompressed.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-ctr-v1-uncompressed.parquet new file mode 100644 index 0000000000..805361aca0 Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-ctr-v1-uncompressed.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-v1-snappy.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-v1-snappy.parquet new file mode 100644 index 0000000000..e0d18ce38a Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-v1-snappy.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-v1-uncompressed.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-v1-uncompressed.parquet new file mode 100644 index 0000000000..754253eb1d Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/java-legacy-v2-aes-gcm-v1-uncompressed.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/legacy-v2-ambiguous-aes-gcm-ctr-v1.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/legacy-v2-ambiguous-aes-gcm-ctr-v1.parquet new file mode 100644 index 0000000000..3a7363dc88 Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/legacy-v2-ambiguous-aes-gcm-ctr-v1.parquet differ diff --git a/parquet-hadoop/src/test/resources/encryption/datapagev2/legacy-v2-ambiguous-aes-gcm-v1.parquet b/parquet-hadoop/src/test/resources/encryption/datapagev2/legacy-v2-ambiguous-aes-gcm-v1.parquet new file mode 100644 index 0000000000..39b7292b83 Binary files /dev/null and b/parquet-hadoop/src/test/resources/encryption/datapagev2/legacy-v2-ambiguous-aes-gcm-v1.parquet differ