diff --git a/parquet-hadoop/README.md b/parquet-hadoop/README.md index 2025243143..11ea7b6b55 100644 --- a/parquet-hadoop/README.md +++ b/parquet-hadoop/README.md @@ -451,6 +451,11 @@ 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, the repetition and definition levels of the DataPageV2 pages of an encrypted column are encrypted +together with the values, as the Parquet encryption specification has it and as Arrow does. Earlier versions left +those levels in plaintext. 1.19.0 reads files of both kinds, but 1.18 and earlier cannot read the DataPageV2 pages +that 1.19.0 writes for encrypted columns that have levels: with `AES_GCM_V1` they fail to decrypt the page, with +`AES_GCM_CTR_V1` they decode garbage levels. **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..fda3e9a3ad 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 @@ -42,7 +42,9 @@ import org.apache.parquet.column.page.PageReader; import org.apache.parquet.compression.CompressionCodecFactory.BytesInputDecompressor; import org.apache.parquet.crypto.AesCipher; +import org.apache.parquet.crypto.AesCtrDecryptor; import org.apache.parquet.crypto.ModuleCipherFactory.ModuleType; +import org.apache.parquet.crypto.TagVerificationException; import org.apache.parquet.filter2.columnindex.RowRanges; import org.apache.parquet.format.BlockCipher; import org.apache.parquet.internal.column.columnindex.OffsetIndex; @@ -218,85 +220,184 @@ public DataPage visit(DataPageV1 dataPageV1) { @Override public DataPage visit(DataPageV2 dataPageV2) { - if (!dataPageV2.isCompressed() && offsetIndex == null && null == blockDecryptor) { - return dataPageV2; + if (dataPageV2 instanceof FullBodyEncryptedDataPageV2) { + return readPageV2WithEncryptedLevels( + (FullBodyEncryptedDataPageV2) 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. If the layout is ambiguous, the page + * is decrypted as one module first, and read with plaintext levels if the GCM tag check fails. CTR + * has no tag, so ambiguous CTR pages are read with plaintext levels. + */ + private DataPageV2 readPageV2WithEncryptedLevels(FullBodyEncryptedDataPageV2 page, int currentPageIndex) { + DataPageV2 plaintextLevelsPage = page.getPlaintextLevelsLayoutPage(); + if (plaintextLevelsPage == null) { + return decryptPageV2WithEncryptedLevels(page, currentPageIndex, decryptsOffHeap()); + } + if (blockDecryptor instanceof AesCtrDecryptor) { + return readPageV2WithPlaintextLevels(plaintextLevelsPage, currentPageIndex); + } + try { + // Not off heap: decrypting in place would overwrite the bytes the fallback reads + return decryptPageV2WithEncryptedLevels(page, currentPageIndex, false); + } catch (TagVerificationException notASingleModule) { + LOG.debug( + "DataPageV2 body is not a single encrypted module, reading it with plaintext levels", + notASingleModule); + try { + return readPageV2WithPlaintextLevels(plaintextLevelsPage, currentPageIndex); + } catch (TagVerificationException notPlaintextLevelsEither) { + notPlaintextLevelsEither.addSuppressed(notASingleModule); + throw notPlaintextLevelsEither; + } + } + } + + /** Decrypts the body, splits it into levels and values, and decompresses the values. */ + private DataPageV2 decryptPageV2WithEncryptedLevels( + FullBodyEncryptedDataPageV2 page, int currentPageIndex, boolean offHeap) { + int repetitionLevelsByteLength = page.getRepetitionLevelsByteLength(); + int definitionLevelsByteLength = page.getDefinitionLevelsByteLength(); + int levelsByteLength = repetitionLevelsByteLength + definitionLevelsByteLength; + try { + ByteBuffer body = decryptDataPageModule(page.getData(), offHeap); + checkDecryptedPageSize(body.limit(), levelsByteLength); + BytesInput repetitionLevels = BytesInput.from(body.slice(0, repetitionLevelsByteLength)); + BytesInput definitionLevels = + BytesInput.from(body.slice(repetitionLevelsByteLength, definitionLevelsByteLength)); + BytesInput values = BytesInput.from(body.slice(levelsByteLength, body.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(int decryptedSize, int levelsByteLength) { + if (decryptedSize < levelsByteLength) { + throw new ParquetDecodingException("Decrypted DataPageV2 body has " + decryptedSize + + " bytes, which is less than the " + levelsByteLength + + " bytes of repetition and definition levels in the page header"); + } + } + + 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..32f2ceb804 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 @@ -317,12 +317,23 @@ public void writePageV2( compressedData = compressor.compress(data); compressed = true; } + // When encrypted, the levels and the values are encrypted together as one module + List pageBody = new ArrayList<>(3); if (null != pageBlockEncryptor) { AesCipher.quickUpdatePageAAD(dataPageAAD, pageOrdinal); - compressedData = BytesInput.from(pageBlockEncryptor.encrypt(compressedData.toByteArray(), dataPageAAD)); + byte[] plainPageBody = BytesInput.concat(repetitionLevels, definitionLevels, compressedData) + .toByteArray(); + pageBody.add(BytesInput.from(pageBlockEncryptor.encrypt(plainPageBody, dataPageAAD))); + } else { + pageBody.add(repetitionLevels); + pageBody.add(definitionLevels); + pageBody.add(compressedData); + } + long pageBodySize = 0; + for (BytesInput part : pageBody) { + pageBodySize += part.size(); } - int compressedSize = - toIntWithCheck(compressedData.size() + repetitionLevels.size() + definitionLevels.size()); + int compressedSize = toIntWithCheck(pageBodySize); tempOutputStream.reset(); if (null != headerBlockEncryptor) { AesCipher.quickUpdatePageAAD(dataPageHeaderAAD, pageOrdinal); @@ -330,14 +341,10 @@ public void writePageV2( if (pageWriteChecksumEnabled) { crc.reset(); try (ByteBufferReleaser pageReleaser = new ByteBufferReleaser(allocator)) { - if (repetitionLevels.size() > 0) { - crc.update(repetitionLevels.toByteBuffer(pageReleaser)); - } - if (definitionLevels.size() > 0) { - crc.update(definitionLevels.toByteBuffer(pageReleaser)); - } - if (compressedData.size() > 0) { - crc.update(compressedData.toByteBuffer(pageReleaser)); + for (BytesInput part : pageBody) { + if (part.size() > 0) { + crc.update(part.toByteBuffer(pageReleaser)); + } } } parquetMetadataConverter.writeDataPageV2Header( @@ -382,8 +389,10 @@ public void writePageV2( // by concatenating before collecting instead of collecting twice, // we only allocate one buffer to copy into instead of multiple. - buf.collect(BytesInput.concat( - BytesInput.from(tempOutputStream), repetitionLevels, definitionLevels, compressedData)); + List pageBytes = new ArrayList<>(pageBody.size() + 1); + pageBytes.add(BytesInput.from(tempOutputStream)); + pageBytes.addAll(pageBody); + buf.collect(BytesInput.concat(pageBytes)); dataEncodings.add(dataEncoding); } diff --git a/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/FullBodyEncryptedDataPageV2.java b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/FullBodyEncryptedDataPageV2.java new file mode 100644 index 0000000000..181e5dc92b --- /dev/null +++ b/parquet-hadoop/src/main/java/org/apache/parquet/hadoop/FullBodyEncryptedDataPageV2.java @@ -0,0 +1,140 @@ +/* + * 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.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 body, levels included, is one encrypted module. The body is held as + * {@link #getData()} and the levels are empty until {@code ColumnChunkPageReader} decrypts the page + * and splits the body using the level lengths. + */ +final class FullBodyEncryptedDataPageV2 extends DataPageV2 { + + /** How the body of an encrypted DataPageV2 page that has levels is laid out. */ + enum Layout { + /** {@code [module of levels and values]} */ + SINGLE_MODULE, + /** {@code [plaintext levels][module of values]} */ + PLAINTEXT_LEVELS, + /** Both length checks match; only decryption can tell. */ + AMBIGUOUS + } + + private final int repetitionLevelsByteLength; + private final int definitionLevelsByteLength; + private final DataPageV2 plaintextLevelsLayoutPage; + + /** + * @param encryptedBody {@code length (4 bytes, LE) | nonce | ciphertext [| tag]} + * @param uncompressedSize uncompressed page size, levels included + * @param plaintextLevelsLayoutPage the page split as {@link Layout#PLAINTEXT_LEVELS}, if the layout is + * {@link Layout#AMBIGUOUS}; otherwise null + */ + FullBodyEncryptedDataPageV2( + int rowCount, + int nullCount, + int valueCount, + Encoding dataEncoding, + BytesInput encryptedBody, + int repetitionLevelsByteLength, + int definitionLevelsByteLength, + int uncompressedSize, + Statistics statistics, + boolean isCompressed, + DataPageV2 plaintextLevelsLayoutPage) { + super( + rowCount, + nullCount, + valueCount, + BytesInput.empty(), + BytesInput.empty(), + dataEncoding, + encryptedBody, + uncompressedSize, + statistics, + isCompressed); + this.repetitionLevelsByteLength = repetitionLevelsByteLength; + this.definitionLevelsByteLength = definitionLevelsByteLength; + this.plaintextLevelsLayoutPage = plaintextLevelsLayoutPage; + } + + int getRepetitionLevelsByteLength() { + return repetitionLevelsByteLength; + } + + int getDefinitionLevelsByteLength() { + return definitionLevelsByteLength; + } + + /** @return the page split as {@link Layout#PLAINTEXT_LEVELS}, or null if the layout is not ambiguous */ + DataPageV2 getPlaintextLevelsLayoutPage() { + return plaintextLevelsLayoutPage; + } + + @Override + public void setCrc(int crc) { + super.setCrc(crc); + if (plaintextLevelsLayoutPage != null) { + plaintextLevelsLayoutPage.setCrc(crc); + } + } + + /** + * Detects the layout of the body 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 neither holds, the page is corrupt and {@link Layout#SINGLE_MODULE} is returned, so that + * decryption fails. + * + * @param pageBody the page body, of {@code compressed_page_size} bytes + * @param levelsByteLength the length of the repetition and definition levels, above zero + */ + static Layout detectLayout(BytesInput pageBody, int levelsByteLength) throws IOException { + long size = pageBody.size(); + boolean singleModule = size >= Integer.BYTES && readLength(pageBody, 0) == size - Integer.BYTES; + boolean plaintextLevels = size - levelsByteLength >= Integer.BYTES + && readLength(pageBody, levelsByteLength) == size - levelsByteLength - Integer.BYTES; + if (plaintextLevels && singleModule) { + return Layout.AMBIGUOUS; + } + return plaintextLevels ? Layout.PLAINTEXT_LEVELS : Layout.SINGLE_MODULE; + } + + private static int readLength(BytesInput pageBody, int offset) throws IOException { + try (ByteBufferInputStream in = pageBody.toInputStream()) { + in.skipFully(offset); + return BytesUtils.readIntLittleEndian(in); + } + } +} 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..c33bb800de 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,129 @@ private int getPageOrdinal(int dataPageCountReadSoFar) { return offsetIndex.getPageOrdinal(dataPageCountReadSoFar); } + /** + * Reads a DataPageV2 body. For an encrypted column with levels, returns a {@link + * FullBodyEncryptedDataPageV2} if the levels may be 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 pageBody = this.readAsBytesInput(compressedPageSize); + if (options.usePageChecksumVerification() && pageHeader.isSetCrc()) { + verifyCrc( + pageHeader.getCrc(), + "could not verify page integrity, CRC checksum verification failed", + pageBody); + } + switch (FullBodyEncryptedDataPageV2.detectLayout(pageBody, levelsByteLength)) { + case PLAINTEXT_LEVELS: + return pageWithPlaintextLevels(pageHeader, pageBody, encoding, statistics); + case SINGLE_MODULE: + return fullBodyEncryptedPage(pageHeader, pageBody, encoding, statistics, null); + case AMBIGUOUS: + default: + return fullBodyEncryptedPage( + pageHeader, + pageBody, + encoding, + statistics, + pageWithPlaintextLevels(pageHeader, pageBody, 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 levels off the front of a page body whose levels are not encrypted. */ + private DataPageV2 pageWithPlaintextLevels( + PageHeader pageHeader, + BytesInput pageBody, + org.apache.parquet.column.Encoding encoding, + org.apache.parquet.column.statistics.Statistics statistics) + throws IOException { + DataPageHeaderV2 header = pageHeader.getData_page_header_v2(); + ByteBufferInputStream pageBodyStream = pageBody.toInputStream(); + return new DataPageV2( + header.getNum_rows(), + header.getNum_nulls(), + header.getNum_values(), + BytesInput.from(pageBodyStream.sliceBuffers(header.getRepetition_levels_byte_length())), + BytesInput.from(pageBodyStream.sliceBuffers(header.getDefinition_levels_byte_length())), + encoding, + BytesInput.from(pageBodyStream.remainingBuffers()), + pageHeader.getUncompressed_page_size(), + statistics, + header.isIs_compressed()); + } + + /** + * @param plaintextLevelsPage the same page read as {@link #pageWithPlaintextLevels}, if the layout is + * ambiguous; null otherwise + */ + private FullBodyEncryptedDataPageV2 fullBodyEncryptedPage( + PageHeader pageHeader, + BytesInput pageBody, + org.apache.parquet.column.Encoding encoding, + org.apache.parquet.column.statistics.Statistics statistics, + DataPageV2 plaintextLevelsPage) { + DataPageHeaderV2 header = pageHeader.getData_page_header_v2(); + return new FullBodyEncryptedDataPageV2( + header.getNum_rows(), + header.getNum_nulls(), + header.getNum_values(), + encoding, + pageBody, + header.getRepetition_levels_byte_length(), + header.getDefinition_levels_byte_length(), + pageHeader.getUncompressed_page_size(), + statistics, + header.isIs_compressed(), + plaintextLevelsPage); + } + /** * @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..f800e0c5b3 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 @@ -28,6 +28,8 @@ import java.io.IOException; import java.nio.charset.StandardCharsets; import java.util.ArrayList; +import java.util.Arrays; +import java.util.Collections; import java.util.HashMap; import java.util.HashSet; import java.util.LinkedHashSet; @@ -1163,11 +1165,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 +1211,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 +1233,10 @@ 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,12 @@ 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 +1325,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 +1339,11 @@ 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 +1379,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 +1394,13 @@ 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 +1416,114 @@ public void writeDataPageV2( byte[] pageHeaderAAD, SizeStatistics sizeStatistics) throws IOException { + writeDataPageV2Body( + rowCount, + nullCount, + valueCount, + toIntWithCheck(repetitionLevels.size(), "page repetition levels"), + toIntWithCheck(definitionLevels.size(), "page definition levels"), + dataEncoding, + Arrays.asList(repetitionLevels, definitionLevels, 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 page body, 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 body of an encrypted column must be encrypted"); + List pageBody; + if (pageBlockEncryptor == null) { + pageBody = Arrays.asList(repetitionLevels, definitionLevels, bytes); + } else { + Preconditions.checkArgument(dataPageAAD != null, "dataPageAAD is required to encrypt the page body"); + pageBody = Collections.singletonList(BytesInput.from(pageBlockEncryptor.encrypt( + BytesInput.concat(repetitionLevels, definitionLevels, bytes).toByteArray(), dataPageAAD))); + } + writeDataPageV2Body( + rowCount, + nullCount, + valueCount, + toIntWithCheck(repetitionLevels.size(), "page repetition levels"), + toIntWithCheck(definitionLevels.size(), "page definition levels"), + dataEncoding, + pageBody, + compressed, + uncompressedDataSize, + statistics, + metadataBlockEncryptor, + pageHeaderAAD, + sizeStatistics); + } + + /** @param pageBody the bytes written after the page header */ + private void writeDataPageV2Body( + int rowCount, + int nullCount, + int valueCount, + int rlByteLength, + int dlByteLength, + Encoding dataEncoding, + List pageBody, + 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"); + long pageBodySize = 0; + for (BytesInput part : pageBody) { + pageBodySize += part.size(); + } + int compressedSize = toIntWithCheck(pageBodySize, "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) { @@ -1403,14 +1532,10 @@ public void writeDataPageV2( if (pageWriteChecksumEnabled) { crc.reset(); - if (repetitionLevels.size() > 0) { - crcUpdate(repetitionLevels); - } - if (definitionLevels.size() > 0) { - crcUpdate(definitionLevels); - } - if (bytes.size() > 0) { - crcUpdate(bytes); + for (BytesInput part : pageBody) { + if (part.size() > 0) { + crcUpdate(part); + } } metadataConverter.writeDataPageV2Header( uncompressedSize, @@ -1451,7 +1576,7 @@ public void writeDataPageV2( currentEncodings.add(dataEncoding); encodingStatsBuilder.addDataEncoding(dataEncoding); - BytesInput.concat(repetitionLevels, definitionLevels, bytes).writeAllTo(out); + BytesInput.concat(pageBody).writeAllTo(out); offsetIndexBuilder.add( toIntWithCheck(out.getPos() - beforeHeader, "page"), 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/DataPageV2EncryptionTestData.java b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/DataPageV2EncryptionTestData.java new file mode 100644 index 0000000000..7c7d1deaba --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/DataPageV2EncryptionTestData.java @@ -0,0 +1,169 @@ +/* + * 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 java.io.IOException; +import java.nio.charset.StandardCharsets; +import java.util.ArrayList; +import java.util.List; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.Path; +import org.apache.parquet.bytes.DirectByteBufferAllocator; +import org.apache.parquet.bytes.HeapByteBufferAllocator; +import org.apache.parquet.bytes.TrackingByteBufferAllocator; +import org.apache.parquet.example.data.Group; +import org.apache.parquet.example.data.simple.SimpleGroupFactory; +import org.apache.parquet.filter2.compat.FilterCompat; +import org.apache.parquet.hadoop.ParquetInputFormat; +import org.apache.parquet.hadoop.ParquetReader; +import org.apache.parquet.hadoop.example.GroupReadSupport; +import org.apache.parquet.schema.MessageType; +import org.apache.parquet.schema.MessageTypeParser; + +/** + * Schema, rows and keys of the DataPageV2 encryption tests and fixtures. {@code seq} is required (no + * levels), {@code id} and {@code name} are optional (definition levels), {@code tags} is a list + * (repetition and definition levels). + */ +final class DataPageV2EncryptionTestData { + + public static final int NUM_ROWS = 500; + + /** Footer key, also the key of all columns with uniform encryption. */ + public static final byte[] FOOTER_KEY = "0123456789abcdef".getBytes(StandardCharsets.UTF_8); + + public static final byte[] FOOTER_KEY_METADATA = "footkey".getBytes(StandardCharsets.UTF_8); + + public 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);" + + " }" + + " }" + + "}"); + + private DataPageV2EncryptionTestData() {} + + /** Builds row {@code i}, as stored in the {@code arrow-v2-*} fixtures. */ + public 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; + } + + public static List expectedRows() { + List rows = new ArrayList<>(NUM_ROWS); + for (int i = 0; i < NUM_ROWS; i++) { + rows.add(row(i).toString()); + } + return rows; + } + + public static final byte[] COLUMN_KEY = "fedcba9876543210".getBytes(StandardCharsets.UTF_8); + + public static final byte[] COLUMN_KEY_METADATA = "colkey".getBytes(StandardCharsets.UTF_8); + + /** Resolves the key metadata of {@link #FOOTER_KEY} and {@link #COLUMN_KEY}. */ + public static FileDecryptionProperties decryptionProperties() { + return FileDecryptionProperties.builder() + .withKeyRetriever(keyMetadata -> { + String id = new String(keyMetadata, StandardCharsets.UTF_8); + switch (id) { + case "footkey": + return FOOTER_KEY; + case "colkey": + return COLUMN_KEY; + default: + throw new KeyAccessDeniedException("Unknown key " + id); + } + }) + .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 + */ + public 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. */ + public 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; + } +} 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..2226e7543f --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/crypto/TestDataPageV2EncryptionInterop.java @@ -0,0 +1,580 @@ +/* + * 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.apache.parquet.crypto.DataPageV2EncryptionTestData.COLUMN_KEY; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.COLUMN_KEY_METADATA; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.FOOTER_KEY; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.FOOTER_KEY_METADATA; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.NUM_ROWS; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.SCHEMA; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.decryptionProperties; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.expectedRows; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.readRows; +import static org.apache.parquet.crypto.DataPageV2EncryptionTestData.row; +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.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.ParquetReadOptions; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.bytes.BytesUtils; +import org.apache.parquet.column.Encoding; +import org.apache.parquet.column.ParquetProperties; +import org.apache.parquet.column.ParquetProperties.WriterVersion; +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.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.ParquetWriter; +import org.apache.parquet.hadoop.example.ExampleParquetWriter; +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.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.ParquetDecodingException; +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; + +/** + * 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/"; + + /** {@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 writesTheLevelsInTheEncryptedPageBody(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 body", page) + .isTrue(); + assertThat(page.decryptedLength).isGreaterThanOrEqualTo(page.levelsLength()); + if (codec == CompressionCodecName.UNCOMPRESSED) { + assertThat(page.decryptedLength).isEqualTo(page.uncompressedPageSize); + } + } + } + + @ParameterizedTest(name = "{0}, {1}") + @MethodSource("ciphersAndCodecs") + public void readsWhatItWrites(ParquetCipher cipher, CompressionCodecName codec) throws Exception { + Path file = writeWithParquetJava(cipher, codec); + + assertThat(readRows(file, new Configuration(), decryptionProperties(), false)) + .containsExactlyElementsOf(expectedRows()); + assertThat(readRows(file, new Configuration(), decryptionProperties(), true)) + .containsExactlyElementsOf(expectedRows()); + } + + @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.bodyOffset + 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 rewriterWritesTheLevelsInTheEncryptedPageBody(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()); + assertThat(readRows(rewritten, new Configuration(), decryptionProperties(), true)) + .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(decodeDataPagesV2(file, FOOTER_KEY, false)) + .allSatisfy(page -> assertThat(page.singleModule).isTrue()); + 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 bodySize - 4}, so the layout is {@code + * AMBIGUOUS}. + */ + @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:")); + } + + @Test + public void writerRejectsAnEncryptedPageHeaderWithoutAnEncryptedPageBody() 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 body 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; + } + + 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 bodyOffset; + private final boolean singleModule; + private final int decryptedLength; + private final int uncompressedPageSize; + + private DecodedPage( + String column, + int repetitionLevelsLength, + int definitionLevelsLength, + long bodyOffset, + boolean singleModule, + int decryptedLength, + int uncompressedPageSize) { + this.column = column; + this.repetitionLevelsLength = repetitionLevelsLength; + this.definitionLevelsLength = definitionLevelsLength; + this.bodyOffset = bodyOffset; + 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 " + bodyOffset + " 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 page body as one module if its 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 bodyOffset = bytes.length - in.available(); + byte[] body = in.readNBytes(header.getCompressed_page_size()); + assertThat(body).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(body, 0) == body.length - Integer.BYTES; + byte[] module = singleModule ? body : Arrays.copyOfRange(body, levels, body.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, + bodyOffset, + 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/TestFullBodyEncryptedDataPageV2.java b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFullBodyEncryptedDataPageV2.java new file mode 100644 index 0000000000..1cbe9c0870 --- /dev/null +++ b/parquet-hadoop/src/test/java/org/apache/parquet/hadoop/TestFullBodyEncryptedDataPageV2.java @@ -0,0 +1,123 @@ +/* + * 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 java.io.IOException; +import java.nio.ByteBuffer; +import java.nio.ByteOrder; +import org.apache.parquet.bytes.BytesInput; +import org.apache.parquet.hadoop.FullBodyEncryptedDataPageV2.Layout; +import org.junit.jupiter.api.Test; + +public class TestFullBodyEncryptedDataPageV2 { + + /** 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 bodyThatIsOneModuleIsASingleModule() throws IOException { + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(module(50)), LEVELS.length)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void bodyWithPlaintextLevelsInFrontOfAModuleHasPlaintextLevels() throws IOException { + byte[] body = concat(LEVELS, module(50)); + + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(body), LEVELS.length)) + .isEqualTo(Layout.PLAINTEXT_LEVELS); + } + + @Test + public void bodyThatIsOneModuleIsRecognizedWhenItIsNotInOneBuffer() throws IOException { + byte[] body = module(300); + ByteBuffer first = ByteBuffer.wrap(body, 0, 2).slice(); + ByteBuffer second = ByteBuffer.wrap(body, 2, body.length - 2).slice(); + + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(first, second), LEVELS.length)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void bodyWithPlaintextLevelsIsRecognizedWhenTheModuleLengthIsNotInOneBuffer() throws IOException { + byte[] body = concat(LEVELS, module(300)); + // Split inside the module length + ByteBuffer first = ByteBuffer.wrap(body, 0, LEVELS.length + 2).slice(); + ByteBuffer second = ByteBuffer.wrap(body, LEVELS.length + 2, body.length - LEVELS.length - 2) + .slice(); + + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(first, second), LEVELS.length)) + .isEqualTo(Layout.PLAINTEXT_LEVELS); + } + + @Test + public void bodyThatFitsBothLayoutsIsAmbiguous() throws IOException { + // Levels whose first 4 bytes are bodySize - 4 + byte[] module = module(50); + int bodySize = LEVELS.length + module.length; + byte[] levels = ByteBuffer.allocate(LEVELS.length) + .order(ByteOrder.LITTLE_ENDIAN) + .putInt(bodySize - Integer.BYTES) + .array(); + byte[] body = concat(levels, module); + + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(body), levels.length)) + .isEqualTo(Layout.AMBIGUOUS); + } + + @Test + public void bodyThatFitsNeitherLayoutIsTreatedAsASingleModuleSoThatDecryptionReportsIt() throws IOException { + byte[] body = concat(LEVELS, module(50)); + body[LEVELS.length] ^= 0x01; + + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(body), LEVELS.length)) + .isEqualTo(Layout.SINGLE_MODULE); + } + + @Test + public void bodyThatIsTooShortToHoldAModuleLengthIsNotAnythingElseThanASingleModule() throws IOException { + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.empty(), 1)) + .isEqualTo(Layout.SINGLE_MODULE); + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(new byte[] {1, 2, 3}), 1)) + .isEqualTo(Layout.SINGLE_MODULE); + // Body is only levels + assertThat(FullBodyEncryptedDataPageV2.detectLayout(BytesInput.from(LEVELS), LEVELS.length)) + .isEqualTo(Layout.SINGLE_MODULE); + } +} 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