diff --git a/api/src/main/java/org/apache/iceberg/encryption/EncryptedOutputFile.java b/api/src/main/java/org/apache/iceberg/encryption/EncryptedOutputFile.java index 54dce5d3263e..300c88d18862 100644 --- a/api/src/main/java/org/apache/iceberg/encryption/EncryptedOutputFile.java +++ b/api/src/main/java/org/apache/iceberg/encryption/EncryptedOutputFile.java @@ -39,7 +39,7 @@ public interface EncryptedOutputFile { EncryptionKeyMetadata keyMetadata(); /** Underlying output file for native encryption. */ - default OutputFile rawOutputFile() { + default OutputFile plainOutputFile() { throw new UnsupportedOperationException("Not implemented"); - }; + } } diff --git a/core/src/main/java/org/apache/iceberg/CatalogProperties.java b/core/src/main/java/org/apache/iceberg/CatalogProperties.java index 9d350525cbf8..b6fd990f0ac6 100644 --- a/core/src/main/java/org/apache/iceberg/CatalogProperties.java +++ b/core/src/main/java/org/apache/iceberg/CatalogProperties.java @@ -156,7 +156,7 @@ private CatalogProperties() {} public static final String AUTH_SESSION_TIMEOUT_MS = "auth.session-timeout-ms"; public static final long AUTH_SESSION_TIMEOUT_MS_DEFAULT = TimeUnit.HOURS.toMillis(1); + public static final String ENCRYPTION_KMS_TYPE = "encryption.kms-type"; - public static final String ENCRYPTION_KMS_CUSTOM_TYPE = "custom"; - public static final String ENCRYPTION_KMS_CLIENT_IMPL = "encryption.kms.client-impl"; + public static final String ENCRYPTION_KMS_IMPL = "encryption.kms-impl"; } diff --git a/core/src/main/java/org/apache/iceberg/encryption/EncryptedFiles.java b/core/src/main/java/org/apache/iceberg/encryption/EncryptedFiles.java index 912104a5305d..c0fc41ca1385 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/EncryptedFiles.java +++ b/core/src/main/java/org/apache/iceberg/encryption/EncryptedFiles.java @@ -57,9 +57,5 @@ public static EncryptedOutputFile encryptedOutput( encryptedOutputFile, BaseEncryptionKeyMetadata.fromByteArray(keyMetadata)); } - public static EncryptedOutputFile plainAsEncryptedOutput(OutputFile encryptingOutputFile) { - return new BaseEncryptedOutputFile(encryptingOutputFile, EncryptionKeyMetadata.empty()); - } - private EncryptedFiles() {} } diff --git a/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java b/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java index ec8156b139ca..ad1deecb8da1 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java +++ b/core/src/main/java/org/apache/iceberg/encryption/EncryptionUtil.java @@ -18,72 +18,40 @@ */ package org.apache.iceberg.encryption; -import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT; -import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT; -import static org.apache.iceberg.TableProperties.ENCRYPTION_DEK_LENGTH; -import static org.apache.iceberg.TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT; -import static org.apache.iceberg.TableProperties.ENCRYPTION_TABLE_KEY; - -import java.nio.ByteBuffer; import java.util.Map; import org.apache.iceberg.CatalogProperties; import org.apache.iceberg.FileFormat; +import org.apache.iceberg.TableProperties; import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.util.PropertyUtil; public class EncryptionUtil { private EncryptionUtil() {} - public static KeyMetadata parseKeyMetadata(ByteBuffer metadataBuffer) { - return KeyMetadata.parse(metadataBuffer); - } - - public static EncryptionKeyMetadata createKeyMetadata(ByteBuffer key, ByteBuffer aadPrefix) { - return new KeyMetadata(key, aadPrefix); - } - - public static long gcmEncryptionLength(long plainFileLength) { - int numberOfFullBlocks = Math.toIntExact(plainFileLength / Ciphers.PLAIN_BLOCK_SIZE); - int plainBytesInLastBlock = - Math.toIntExact(plainFileLength - numberOfFullBlocks * Ciphers.PLAIN_BLOCK_SIZE); - boolean fullBlocksOnly = (0 == plainBytesInLastBlock); - int cipherBytesInLastBlock = - fullBlocksOnly ? 0 : plainBytesInLastBlock + Ciphers.NONCE_LENGTH + Ciphers.GCM_TAG_LENGTH; - int cipherBlockSize = Ciphers.PLAIN_BLOCK_SIZE + Ciphers.NONCE_LENGTH + Ciphers.GCM_TAG_LENGTH; - return (long) Ciphers.GCM_STREAM_HEADER_LENGTH - + numberOfFullBlocks * cipherBlockSize - + cipherBytesInLastBlock; - } - public static KeyManagementClient createKmsClient(Map catalogProperties) { String kmsType = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_TYPE); + String kmsImpl = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_IMPL); - if (kmsType == null) { - throw new IllegalStateException( - "Cannot create StandardEncryptionManagerFactory without KMS type"); - } + Preconditions.checkArgument( + kmsType == null || kmsImpl == null, + "Cannot set both KMS type (%s) and KMS impl (%s)", + kmsType, + kmsImpl); - if (!kmsType.equals(CatalogProperties.ENCRYPTION_KMS_CUSTOM_TYPE)) { - // Currently support only custom types - throw new UnsupportedOperationException("Undefined KMS type " + kmsType); - } - - String kmsClientImpl = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_CLIENT_IMPL); - - if (kmsClientImpl == null) { - throw new IllegalStateException("Custom KMS client class is not defined"); - } + // TODO: Add KMS implementations + Preconditions.checkArgument(kmsType == null, "Unsupported KMS type: %s", kmsType); KeyManagementClient kmsClient; DynConstructors.Ctor ctor; try { - ctor = DynConstructors.builder(KeyManagementClient.class).impl(kmsClientImpl).buildChecked(); + ctor = DynConstructors.builder(KeyManagementClient.class).impl(kmsImpl).buildChecked(); } catch (NoSuchMethodException e) { throw new IllegalArgumentException( String.format( "Cannot initialize KeyManagementClient, missing no-arg constructor for class %s", - kmsClientImpl), + kmsImpl), e); } @@ -93,7 +61,7 @@ public static KeyManagementClient createKmsClient(Map catalogPro throw new IllegalArgumentException( String.format( "Cannot initialize kms client, %s does not implement KeyManagementClient interface", - kmsClientImpl), + kmsImpl), e); } @@ -104,20 +72,19 @@ public static KeyManagementClient createKmsClient(Map catalogPro public static EncryptionManager createEncryptionManager( Map tableProperties, KeyManagementClient kmsClient) { - String tableKeyId = tableProperties.get(ENCRYPTION_TABLE_KEY); + Preconditions.checkArgument(kmsClient != null, "Invalid KMS client: null"); + String tableKeyId = tableProperties.get(TableProperties.ENCRYPTION_TABLE_KEY); if (null == tableKeyId) { // Unencrypted table return PlaintextEncryptionManager.instance(); } - if (kmsClient == null) { - throw new IllegalStateException("Encrypted table. No KMS client is configured in catalog"); - } - String fileFormat = PropertyUtil.propertyAsString( - tableProperties, DEFAULT_FILE_FORMAT, DEFAULT_FILE_FORMAT_DEFAULT); + tableProperties, + TableProperties.DEFAULT_FILE_FORMAT, + TableProperties.DEFAULT_FILE_FORMAT_DEFAULT); if (FileFormat.fromString(fileFormat) != FileFormat.PARQUET) { throw new UnsupportedOperationException( @@ -126,12 +93,15 @@ public static EncryptionManager createEncryptionManager( int dataKeyLength = PropertyUtil.propertyAsInt( - tableProperties, ENCRYPTION_DEK_LENGTH, ENCRYPTION_DEK_LENGTH_DEFAULT); + tableProperties, + TableProperties.ENCRYPTION_DEK_LENGTH, + TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT); - return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient); - } + Preconditions.checkState( + dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32, + "Invalid data key length: %s (must be 16, 24, or 32)", + dataKeyLength); - public static boolean useNativeEncryption(EncryptionKeyMetadata keyMetadata) { - return keyMetadata != null && keyMetadata instanceof KeyMetadata; + return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient); } } diff --git a/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataDecoder.java b/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataDecoder.java index 674685c30164..7e57163d73ea 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataDecoder.java +++ b/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataDecoder.java @@ -28,9 +28,9 @@ import org.apache.iceberg.data.avro.RawDecoder; import org.apache.iceberg.relocated.com.google.common.collect.MapMaker; -class KeyMetadataDecoder extends MessageDecoder.BaseDecoder { +class KeyMetadataDecoder extends MessageDecoder.BaseDecoder { private final org.apache.iceberg.Schema readSchema; - private final Map> decoders = new MapMaker().makeMap(); + private final Map> decoders = new MapMaker().makeMap(); /** * Creates a new decoder that constructs key metadata instances described by schema version. @@ -39,11 +39,11 @@ class KeyMetadataDecoder extends MessageDecoder.BaseDecoder { * instances created by this class will are described by the expected schema. */ KeyMetadataDecoder(byte readSchemaVersion) { - this.readSchema = KeyMetadata.supportedSchemaVersions().get(readSchemaVersion); + this.readSchema = StandardKeyMetadata.supportedSchemaVersions().get(readSchemaVersion); } @Override - public KeyMetadata decode(InputStream stream, KeyMetadata reuse) { + public StandardKeyMetadata decode(InputStream stream, StandardKeyMetadata reuse) { byte writeSchemaVersion; try { @@ -56,14 +56,14 @@ public KeyMetadata decode(InputStream stream, KeyMetadata reuse) { throw new RuntimeException("Version byte - end of stream reached"); } - Schema writeSchema = KeyMetadata.supportedAvroSchemaVersions().get(writeSchemaVersion); + Schema writeSchema = StandardKeyMetadata.supportedAvroSchemaVersions().get(writeSchemaVersion); if (writeSchema == null) { throw new UnsupportedOperationException( "Cannot resolve schema for version: " + writeSchemaVersion); } - RawDecoder decoder = decoders.get(writeSchemaVersion); + RawDecoder decoder = decoders.get(writeSchemaVersion); if (decoder == null) { decoder = new RawDecoder<>(readSchema, GenericAvroReader::create, writeSchema); diff --git a/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataEncoder.java b/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataEncoder.java index faab6a47c814..6f9d1769f07d 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataEncoder.java +++ b/core/src/main/java/org/apache/iceberg/encryption/KeyMetadataEncoder.java @@ -29,18 +29,18 @@ import org.apache.avro.message.MessageEncoder; import org.apache.iceberg.avro.GenericAvroWriter; -class KeyMetadataEncoder implements MessageEncoder { +class KeyMetadataEncoder implements MessageEncoder { private static final ThreadLocal TEMP = ThreadLocal.withInitial(BufferOutputStream::new); private static final ThreadLocal ENCODER = new ThreadLocal<>(); private final byte schemaVersion; private final boolean copyOutputBytes; - private final DatumWriter writer; + private final DatumWriter writer; /** - * Creates a new {@link MessageEncoder} that will deconstruct {@link KeyMetadata} instances - * described by the schema version. + * Creates a new {@link MessageEncoder} that will deconstruct {@link StandardKeyMetadata} + * instances described by the schema version. * *

Buffers returned by {@code encode} are copied and will not be modified by future calls to * {@code encode}. @@ -50,8 +50,8 @@ class KeyMetadataEncoder implements MessageEncoder { } /** - * Creates a new {@link MessageEncoder} that will deconstruct {@link KeyMetadata} instances - * described by the schema version. + * Creates a new {@link MessageEncoder} that will deconstruct {@link StandardKeyMetadata} + * instances described by the schema version. * *

If {@code shouldCopy} is true, then buffers returned by {@code encode} are copied and will * not be modified by future calls to {@code encode}. @@ -62,7 +62,7 @@ class KeyMetadataEncoder implements MessageEncoder { * next call to {@code encode}. */ KeyMetadataEncoder(byte schemaVersion, boolean shouldCopy) { - Schema writeSchema = KeyMetadata.supportedAvroSchemaVersions().get(schemaVersion); + Schema writeSchema = StandardKeyMetadata.supportedAvroSchemaVersions().get(schemaVersion); if (writeSchema == null) { throw new UnsupportedOperationException( @@ -75,7 +75,7 @@ class KeyMetadataEncoder implements MessageEncoder { } @Override - public ByteBuffer encode(KeyMetadata datum) throws IOException { + public ByteBuffer encode(StandardKeyMetadata datum) throws IOException { BufferOutputStream temp = TEMP.get(); temp.reset(); temp.write(schemaVersion); @@ -89,7 +89,7 @@ public ByteBuffer encode(KeyMetadata datum) throws IOException { } @Override - public void encode(KeyMetadata datum, OutputStream stream) throws IOException { + public void encode(StandardKeyMetadata datum, OutputStream stream) throws IOException { BinaryEncoder encoder = EncoderFactory.get().directBinaryEncoder(stream, ENCODER.get()); ENCODER.set(encoder); writer.write(datum, encoder); diff --git a/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java b/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java index 7f95efd9ea60..63f89e7661b3 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java +++ b/core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java @@ -23,6 +23,7 @@ import org.apache.iceberg.TableProperties; import org.apache.iceberg.io.InputFile; import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.io.SeekableInputStream; import org.apache.iceberg.relocated.com.google.common.base.Preconditions; import org.apache.iceberg.relocated.com.google.common.collect.Iterables; import org.apache.iceberg.util.ByteBuffers; @@ -34,38 +35,6 @@ public class StandardEncryptionManager implements EncryptionManager { private transient volatile SecureRandom lazyRNG = null; - class StandardEncryptedOutputFile implements EncryptedOutputFile { - - private final OutputFile encryptingOutputFile; - - private final EncryptionKeyMetadata keyMetadata; - private final OutputFile rawOutputFile; - - StandardEncryptedOutputFile( - OutputFile encryptingOutputFile, - EncryptionKeyMetadata keyMetadata, - OutputFile rawOutputFile) { - this.encryptingOutputFile = encryptingOutputFile; - this.keyMetadata = keyMetadata; - this.rawOutputFile = rawOutputFile; - } - - @Override - public OutputFile encryptingOutputFile() { - return encryptingOutputFile; - } - - @Override - public EncryptionKeyMetadata keyMetadata() { - return keyMetadata; - } - - @Override - public OutputFile rawOutputFile() { - return rawOutputFile; - } - } - /** * @param tableKeyId table encryption key id * @param dataKeyLength length of data encryption key (16/24/32 bytes) @@ -74,6 +43,10 @@ public OutputFile rawOutputFile() { public StandardEncryptionManager( String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) { Preconditions.checkNotNull(tableKeyId, "Invalid encryption key ID: null"); + Preconditions.checkArgument( + dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32, + "Invalid data key length: %s (must be 16, 24, or 32)", + dataKeyLength); Preconditions.checkNotNull(kmsClient, "Invalid KMS client: null"); this.tableKeyId = tableKeyId; this.kmsClient = kmsClient; @@ -81,39 +54,20 @@ public StandardEncryptionManager( } @Override - public EncryptedOutputFile encrypt(OutputFile rawOutput) { - ByteBuffer fileDek = ByteBuffer.allocate(dataKeyLength); - workerRNG().nextBytes(fileDek.array()); - - ByteBuffer aadPrefix = ByteBuffer.allocate(TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT); - workerRNG().nextBytes(aadPrefix.array()); - - KeyMetadata encryptionMetadata = new KeyMetadata(fileDek, aadPrefix); - - return new StandardEncryptedOutputFile( - new AesGcmOutputFile(rawOutput, fileDek.array(), aadPrefix.array()), - encryptionMetadata, - rawOutput); + public EncryptedOutputFile encrypt(OutputFile plainOutput) { + return new StandardEncryptedOutputFile(plainOutput, dataKeyLength); } @Override public InputFile decrypt(EncryptedInputFile encrypted) { - KeyMetadata keyMetadata = KeyMetadata.castOrParse(encrypted.keyMetadata()); - - byte[] fileDek = ByteBuffers.toByteArray(keyMetadata.encryptionKey()); - byte[] aadPrefix = ByteBuffers.toByteArray(keyMetadata.aadPrefix()); - - return new AesGcmInputFile(encrypted.encryptedInputFile(), fileDek, aadPrefix); + // this input file will lazily parse key metadata in case the file is not an AES GCM stream. + return new StandardDecryptedInputFile(encrypted); } @Override public Iterable decrypt(Iterable encrypted) { // Bulk decrypt is only applied to data files. Returning source input files for parquet. - return Iterables.transform(encrypted, this::getSourceFile); - } - - private InputFile getSourceFile(EncryptedInputFile encryptedFile) { - return encryptedFile.encryptedInputFile(); + return Iterables.transform(encrypted, this::decrypt); } private SecureRandom workerRNG() { @@ -126,7 +80,8 @@ private SecureRandom workerRNG() { public ByteBuffer wrapKey(ByteBuffer secretKey) { if (kmsClient == null) { - throw new IllegalStateException("Null KmsClient. WrapKey can't be called from workers"); + throw new IllegalStateException( + "Cannot wrap key after called after serialization (missing KMS client)"); } return kmsClient.wrapKey(secretKey, tableKeyId); @@ -134,9 +89,105 @@ public ByteBuffer wrapKey(ByteBuffer secretKey) { public ByteBuffer unwrapKey(ByteBuffer wrappedSecretKey) { if (kmsClient == null) { - throw new IllegalStateException("Null KmsClient. UnwrapKey can't be called from workers"); + throw new IllegalStateException( + "Cannot wrap key after called after serialization (missing KMS client)"); } return kmsClient.unwrapKey(wrappedSecretKey, tableKeyId); } + + private class StandardEncryptedOutputFile implements EncryptedOutputFile { + private final OutputFile plainOutputFile; + private final int dataKeyLength; + private StandardKeyMetadata lazyKeyMetadata = null; + private OutputFile lazyEncryptingOutputFile = null; + + StandardEncryptedOutputFile(OutputFile plainOutputFile, int dataKeyLength) { + this.plainOutputFile = plainOutputFile; + this.dataKeyLength = dataKeyLength; + } + + @Override + public StandardKeyMetadata keyMetadata() { + if (null == lazyKeyMetadata) { + byte[] fileDek = new byte[dataKeyLength]; + workerRNG().nextBytes(fileDek); + + byte[] aadPrefix = new byte[TableProperties.ENCRYPTION_AAD_LENGTH_DEFAULT]; + workerRNG().nextBytes(aadPrefix); + + this.lazyKeyMetadata = new StandardKeyMetadata(fileDek, aadPrefix); + } + + return lazyKeyMetadata; + } + + @Override + public OutputFile encryptingOutputFile() { + if (null == lazyEncryptingOutputFile) { + this.lazyEncryptingOutputFile = + new AesGcmOutputFile( + plainOutputFile, + ByteBuffers.toByteArray(keyMetadata().encryptionKey()), + ByteBuffers.toByteArray(keyMetadata().aadPrefix())); + } + + return lazyEncryptingOutputFile; + } + + @Override + public OutputFile plainOutputFile() { + return plainOutputFile; + } + } + + private static class StandardDecryptedInputFile implements InputFile { + private final EncryptedInputFile encryptedInputFile; + private StandardKeyMetadata lazyKeyMetadata = null; + private AesGcmInputFile lazyDecryptedInputFile = null; + + private StandardDecryptedInputFile(EncryptedInputFile encryptedInputFile) { + this.encryptedInputFile = encryptedInputFile; + } + + private StandardKeyMetadata keyMetadata() { + if (null == lazyKeyMetadata) { + this.lazyKeyMetadata = StandardKeyMetadata.castOrParse(encryptedInputFile.keyMetadata()); + } + + return lazyKeyMetadata; + } + + private AesGcmInputFile decrypted() { + if (null == lazyDecryptedInputFile) { + this.lazyDecryptedInputFile = + new AesGcmInputFile( + encryptedInputFile.encryptedInputFile(), + ByteBuffers.toByteArray(keyMetadata().encryptionKey()), + ByteBuffers.toByteArray(keyMetadata().aadPrefix())); + } + + return lazyDecryptedInputFile; + } + + @Override + public long getLength() { + return decrypted().getLength(); + } + + @Override + public SeekableInputStream newStream() { + return decrypted().newStream(); + } + + @Override + public String location() { + return decrypted().location(); + } + + @Override + public boolean exists() { + return decrypted().exists(); + } + } } diff --git a/core/src/main/java/org/apache/iceberg/encryption/KeyMetadata.java b/core/src/main/java/org/apache/iceberg/encryption/StandardKeyMetadata.java similarity index 84% rename from core/src/main/java/org/apache/iceberg/encryption/KeyMetadata.java rename to core/src/main/java/org/apache/iceberg/encryption/StandardKeyMetadata.java index 02873eb764d6..2ac70aebc316 100644 --- a/core/src/main/java/org/apache/iceberg/encryption/KeyMetadata.java +++ b/core/src/main/java/org/apache/iceberg/encryption/StandardKeyMetadata.java @@ -31,14 +31,14 @@ import org.apache.iceberg.relocated.com.google.common.collect.ImmutableMap; import org.apache.iceberg.types.Types; -class KeyMetadata implements EncryptionKeyMetadata, IndexedRecord { +class StandardKeyMetadata implements EncryptionKeyMetadata, IndexedRecord { private static final byte V1 = 1; private static final Schema SCHEMA_V1 = new Schema( required(0, "encryption_key", Types.BinaryType.get()), optional(1, "aad_prefix", Types.BinaryType.get())); private static final org.apache.avro.Schema AVRO_SCHEMA_V1 = - AvroSchemaUtil.convert(SCHEMA_V1, KeyMetadata.class.getCanonicalName()); + AvroSchemaUtil.convert(SCHEMA_V1, StandardKeyMetadata.class.getCanonicalName()); private static final Map schemaVersions = ImmutableMap.of(V1, SCHEMA_V1); private static final Map avroSchemaVersions = @@ -52,9 +52,14 @@ class KeyMetadata implements EncryptionKeyMetadata, IndexedRecord { private org.apache.avro.Schema avroSchema; /** Used by Avro reflection to instantiate this class * */ - KeyMetadata() {} + StandardKeyMetadata() {} - KeyMetadata(ByteBuffer encryptionKey, ByteBuffer aadPrefix) { + StandardKeyMetadata(byte[] key, byte[] aad) { + this.encryptionKey = ByteBuffer.wrap(key); + this.aadPrefix = ByteBuffer.wrap(aad); + } + + private StandardKeyMetadata(ByteBuffer encryptionKey, ByteBuffer aadPrefix) { this.encryptionKey = encryptionKey; this.aadPrefix = aadPrefix; this.avroSchema = AVRO_SCHEMA_V1; @@ -76,9 +81,9 @@ ByteBuffer aadPrefix() { return aadPrefix; } - static KeyMetadata castOrParse(EncryptionKeyMetadata keyMetadata) { - if (keyMetadata instanceof KeyMetadata) { - return (KeyMetadata) keyMetadata; + static StandardKeyMetadata castOrParse(EncryptionKeyMetadata keyMetadata) { + if (keyMetadata instanceof StandardKeyMetadata) { + return (StandardKeyMetadata) keyMetadata; } ByteBuffer kmBuffer = keyMetadata.buffer(); @@ -90,7 +95,7 @@ static KeyMetadata castOrParse(EncryptionKeyMetadata keyMetadata) { return parse(kmBuffer); } - static KeyMetadata parse(ByteBuffer buffer) { + static StandardKeyMetadata parse(ByteBuffer buffer) { try { return KEY_METADATA_DECODER.decode(buffer); } catch (IOException e) { @@ -109,8 +114,7 @@ public ByteBuffer buffer() { @Override public EncryptionKeyMetadata copy() { - KeyMetadata metadata = new KeyMetadata(encryptionKey(), aadPrefix()); - return metadata; + return new StandardKeyMetadata(encryptionKey(), aadPrefix()); } @Override diff --git a/core/src/test/java/org/apache/iceberg/encryption/TestKeyMetadataParser.java b/core/src/test/java/org/apache/iceberg/encryption/TestStandardKeyMetadataParser.java similarity index 84% rename from core/src/test/java/org/apache/iceberg/encryption/TestKeyMetadataParser.java rename to core/src/test/java/org/apache/iceberg/encryption/TestStandardKeyMetadataParser.java index 27b00b1829c6..889506cb93e4 100644 --- a/core/src/test/java/org/apache/iceberg/encryption/TestKeyMetadataParser.java +++ b/core/src/test/java/org/apache/iceberg/encryption/TestStandardKeyMetadataParser.java @@ -24,16 +24,17 @@ import org.junit.Assert; import org.junit.Test; -public class TestKeyMetadataParser { +public class TestStandardKeyMetadataParser { @Test public void testParser() { ByteBuffer encryptionKey = ByteBuffer.wrap("0123456789012345".getBytes(StandardCharsets.UTF_8)); ByteBuffer aadPrefix = ByteBuffer.wrap("1234567890123456".getBytes(StandardCharsets.UTF_8)); - KeyMetadata metadata = new KeyMetadata(encryptionKey, aadPrefix); + StandardKeyMetadata metadata = + new StandardKeyMetadata(encryptionKey.array(), aadPrefix.array()); ByteBuffer serialized = metadata.buffer(); - KeyMetadata parsedMetadata = KeyMetadata.parse(serialized); + StandardKeyMetadata parsedMetadata = StandardKeyMetadata.parse(serialized); Assert.assertEquals(parsedMetadata.encryptionKey(), encryptionKey); Assert.assertEquals(parsedMetadata.aadPrefix(), aadPrefix); } @@ -41,7 +42,7 @@ public void testParser() { @Test public void testUnsupportedVersion() { ByteBuffer badBuffer = ByteBuffer.wrap(new byte[] {0x02}); - Assertions.assertThatThrownBy(() -> KeyMetadata.parse(badBuffer)) + Assertions.assertThatThrownBy(() -> StandardKeyMetadata.parse(badBuffer)) .isInstanceOf(UnsupportedOperationException.class) .hasMessage("Cannot resolve schema for version: 2"); }