flyrain commented on a change in pull request #3471: URL: https://github.com/apache/iceberg/pull/3471#discussion_r806104745
########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeConfig.java ########## @@ -0,0 +1,105 @@ +/* + * 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.iceberg.encryption; + +import java.io.Serializable; +import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Configuration for an envelope encryption. + */ +public class EnvelopeConfig implements Serializable { + + private final String kekId; + private final EncryptionAlgorithm algorithm; + + public EnvelopeConfig( + String kekId, + EncryptionAlgorithm algorithm) { + this.kekId = Preconditions.checkNotNull(kekId, + "Cannot construct envelope config because KEK ID is not specified"); + this.algorithm = Preconditions.checkNotNull(algorithm, + "Cannot construct envelope config because encryption algorithm is not specified"); + } + + public String kekId() { + return kekId; + } + + public EncryptionAlgorithm algorithm() { + return algorithm; + } + + @Override + public boolean equals(Object o) { + if (this == o) { + return true; + } + + if (o == null || getClass() != o.getClass()) { + return false; + } + + EnvelopeConfig config = (EnvelopeConfig) o; + return Objects.equals(kekId, config.kekId) && + algorithm == config.algorithm; + } + + @Override + public int hashCode() { + return Objects.hash(kekId, algorithm); + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("kekId", kekId) + .add("algorithm", algorithm) + .toString(); + } + + public static Builder builder() { + return new Builder(); + } + + /** + * A builder used to create valid {@link EnvelopeConfig}. + */ + public static class Builder { + private String kekId; + private EncryptionAlgorithm algorithm; + + public Builder singleWrap(String keyId) { + this.kekId = keyId; + return this; + } + + public Builder useAlgorithm(EncryptionAlgorithm encryptionAlgorithm) { + this.algorithm = encryptionAlgorithm; + return this; + } + + public EnvelopeConfig build() { + return new EnvelopeConfig(kekId, algorithm); + } + } +} Review comment: Sure, we can keep the builder if it is going to be used, and make the constructor private. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) Review comment: I'd prefer to check the class member `pushdown` even though non-pushdown is not supported at this moment. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) Review comment: I'd prefer to check the class member `pushdown` even though non-pushdown is not supported at this moment. The same to the method `decrypt`. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) Review comment: ``` if(pushdown) { return nativeEncrypt(output); } else { throw new Exception(); } ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { Review comment: The precondition check would looks simpler than `try... catch` ``` Preconditions.checkArgument(rawOutput instanceof NativelyEncryptedFile, ""); ``` ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { + ((NativelyEncryptedFile) rawOutput).setNativeCryptoParameters(nativeEncryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively encrypt " + rawOutput.location() + " because the class " + + rawOutput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return new BaseEncryptedOutputFile(rawOutput, metadata); + } + + @Override + public InputFile decrypt(EncryptedInputFile encrypted) { + EnvelopeMetadata metadata = EnvelopeMetadataParser.fromJson(encrypted.keyMetadata().buffer()); + ByteBuffer fileDek = kmsClient.unwrapKey(metadata.wrappedDek(), metadata.kekId()); + + // Pushdown to data formats with native decryption support (such as Parquet) + NativeFileCryptoParameters nativeDecryptParams = NativeFileCryptoParameters.create(fileDek).build(); + InputFile rawInput = encrypted.encryptedInputFile(); + + try { + ((NativelyEncryptedFile) rawInput).setNativeCryptoParameters(nativeDecryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively decrypt " + rawInput.location() + " because the class " + + rawInput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return rawInput; + } + + private EnvelopeMetadata generateEnvelopeMetadata(EnvelopeConfig config) { + if (null == workerRNG) { + workerRNG = new SecureRandom(); + } + + String tableKekID = config.kekId(); + byte[] fileDek; + String wrappedFileDEK; + + if (kmsClient.supportsKeyGeneration()) { + KmsClient.KeyGenerationResult generatedDek = kmsClient.generateKey(tableKekID); + fileDek = generatedDek.key().array(); + wrappedFileDEK = generatedDek.wrappedKey(); + } else { + fileDek = new byte[dataKeyLength]; + workerRNG.nextBytes(fileDek); + wrappedFileDEK = kmsClient.wrapKey(ByteBuffer.wrap(fileDek), tableKekID); + } + + EnvelopeMetadata fileEnvelopeMetadata = new EnvelopeMetadata(tableKekID, wrappedFileDEK, null, + config.algorithm()); Review comment: Nit: these two lines can be in one line. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { + ((NativelyEncryptedFile) rawOutput).setNativeCryptoParameters(nativeEncryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively encrypt " + rawOutput.location() + " because the class " + + rawOutput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return new BaseEncryptedOutputFile(rawOutput, metadata); + } + + @Override + public InputFile decrypt(EncryptedInputFile encrypted) { + EnvelopeMetadata metadata = EnvelopeMetadataParser.fromJson(encrypted.keyMetadata().buffer()); + ByteBuffer fileDek = kmsClient.unwrapKey(metadata.wrappedDek(), metadata.kekId()); + + // Pushdown to data formats with native decryption support (such as Parquet) + NativeFileCryptoParameters nativeDecryptParams = NativeFileCryptoParameters.create(fileDek).build(); + InputFile rawInput = encrypted.encryptedInputFile(); + + try { + ((NativelyEncryptedFile) rawInput).setNativeCryptoParameters(nativeDecryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively decrypt " + rawInput.location() + " because the class " + + rawInput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return rawInput; + } + + private EnvelopeMetadata generateEnvelopeMetadata(EnvelopeConfig config) { + if (null == workerRNG) { + workerRNG = new SecureRandom(); + } + + String tableKekID = config.kekId(); + byte[] fileDek; + String wrappedFileDEK; + + if (kmsClient.supportsKeyGeneration()) { + KmsClient.KeyGenerationResult generatedDek = kmsClient.generateKey(tableKekID); + fileDek = generatedDek.key().array(); + wrappedFileDEK = generatedDek.wrappedKey(); + } else { + fileDek = new byte[dataKeyLength]; + workerRNG.nextBytes(fileDek); + wrappedFileDEK = kmsClient.wrapKey(ByteBuffer.wrap(fileDek), tableKekID); + } + + EnvelopeMetadata fileEnvelopeMetadata = new EnvelopeMetadata(tableKekID, wrappedFileDEK, null, + config.algorithm()); + fileEnvelopeMetadata.setDek(ByteBuffer.wrap(fileDek)); + + return fileEnvelopeMetadata; + } + + public static KmsClient loadKmsClient(String classPath, Map<String, String> properties) { + Preconditions.checkNotNull(classPath, + "Cannot initialize KmsClient, class name is null"); Review comment: Nit: can be in one line. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { + ((NativelyEncryptedFile) rawOutput).setNativeCryptoParameters(nativeEncryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively encrypt " + rawOutput.location() + " because the class " + + rawOutput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return new BaseEncryptedOutputFile(rawOutput, metadata); + } + + @Override + public InputFile decrypt(EncryptedInputFile encrypted) { + EnvelopeMetadata metadata = EnvelopeMetadataParser.fromJson(encrypted.keyMetadata().buffer()); + ByteBuffer fileDek = kmsClient.unwrapKey(metadata.wrappedDek(), metadata.kekId()); + + // Pushdown to data formats with native decryption support (such as Parquet) + NativeFileCryptoParameters nativeDecryptParams = NativeFileCryptoParameters.create(fileDek).build(); + InputFile rawInput = encrypted.encryptedInputFile(); + + try { + ((NativelyEncryptedFile) rawInput).setNativeCryptoParameters(nativeDecryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively decrypt " + rawInput.location() + " because the class " + + rawInput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return rawInput; + } + + private EnvelopeMetadata generateEnvelopeMetadata(EnvelopeConfig config) { + if (null == workerRNG) { + workerRNG = new SecureRandom(); + } + + String tableKekID = config.kekId(); + byte[] fileDek; + String wrappedFileDEK; + + if (kmsClient.supportsKeyGeneration()) { + KmsClient.KeyGenerationResult generatedDek = kmsClient.generateKey(tableKekID); + fileDek = generatedDek.key().array(); + wrappedFileDEK = generatedDek.wrappedKey(); + } else { Review comment: Suggest to move line 98 and 99 here. The SecureRandom is not needed in case of Kms client key generation. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeMetadata.java ########## @@ -0,0 +1,142 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Envelope encryption metadata used to encrypt and decrypt data. + * Information except plaintext KEK and DEK can be serialized and stored. + */ +public class EnvelopeMetadata implements EncryptionKeyMetadata { + + private final String kekId; + private final String wrappedDek; + private final byte[] iv; + private final EncryptionAlgorithm algorithm; + + // fields that are expected to be updated later + private byte[] kek; + private byte[] dek; + + public EnvelopeMetadata( Review comment: Nit: they can be in the same line. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeMetadata.java ########## @@ -0,0 +1,142 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Envelope encryption metadata used to encrypt and decrypt data. + * Information except plaintext KEK and DEK can be serialized and stored. + */ +public class EnvelopeMetadata implements EncryptionKeyMetadata { + + private final String kekId; + private final String wrappedDek; + private final byte[] iv; + private final EncryptionAlgorithm algorithm; + + // fields that are expected to be updated later + private byte[] kek; + private byte[] dek; + + public EnvelopeMetadata( + String kekId, + String wrappedDek, + ByteBuffer iv, + EncryptionAlgorithm algorithm) { + this.kekId = Preconditions.checkNotNull(kekId, + "Cannot construct envelope metadata because KEK ID is not specified"); + this.wrappedDek = wrappedDek; + this.iv = iv == null ? null : iv.array(); + this.algorithm = Preconditions.checkNotNull(algorithm, + "Cannot construct envelope metadata because encryption algorithm is not specified"); + } + + public String kekId() { + return kekId; + } + + public ByteBuffer kek() { + return kek == null ? null : ByteBuffer.wrap(kek); + } + + public ByteBuffer dek() { + return dek == null ? null : ByteBuffer.wrap(dek); + } + + public String wrappedDek() { + return wrappedDek; + } + + public ByteBuffer iv() { + return iv == null ? null : ByteBuffer.wrap(iv); + } + + public EncryptionAlgorithm algorithm() { + return algorithm; + } + + public void setKek(ByteBuffer kekBuffer) { + this.kek = kekBuffer == null ? null : kekBuffer.array(); + } + + public void setDek(ByteBuffer dekBuffer) { + this.dek = dekBuffer == null ? null : dekBuffer.array(); + } + + @Override + public ByteBuffer buffer() { + return EnvelopeMetadataParser.toJson(this); + } + + @Override + public EncryptionKeyMetadata copy() { + EnvelopeMetadata metadata = new EnvelopeMetadata(kekId(), wrappedDek(), + iv(), algorithm()); Review comment: Nit: they can be in the same line. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeMetadata.java ########## @@ -0,0 +1,142 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Envelope encryption metadata used to encrypt and decrypt data. + * Information except plaintext KEK and DEK can be serialized and stored. + */ +public class EnvelopeMetadata implements EncryptionKeyMetadata { + + private final String kekId; + private final String wrappedDek; + private final byte[] iv; + private final EncryptionAlgorithm algorithm; + + // fields that are expected to be updated later + private byte[] kek; + private byte[] dek; + + public EnvelopeMetadata( + String kekId, + String wrappedDek, + ByteBuffer iv, + EncryptionAlgorithm algorithm) { + this.kekId = Preconditions.checkNotNull(kekId, + "Cannot construct envelope metadata because KEK ID is not specified"); + this.wrappedDek = wrappedDek; + this.iv = iv == null ? null : iv.array(); + this.algorithm = Preconditions.checkNotNull(algorithm, + "Cannot construct envelope metadata because encryption algorithm is not specified"); + } + + public String kekId() { + return kekId; + } + + public ByteBuffer kek() { + return kek == null ? null : ByteBuffer.wrap(kek); + } + + public ByteBuffer dek() { + return dek == null ? null : ByteBuffer.wrap(dek); + } + + public String wrappedDek() { + return wrappedDek; + } + + public ByteBuffer iv() { + return iv == null ? null : ByteBuffer.wrap(iv); + } + + public EncryptionAlgorithm algorithm() { + return algorithm; + } + + public void setKek(ByteBuffer kekBuffer) { + this.kek = kekBuffer == null ? null : kekBuffer.array(); + } + + public void setDek(ByteBuffer dekBuffer) { + this.dek = dekBuffer == null ? null : dekBuffer.array(); + } + + @Override + public ByteBuffer buffer() { + return EnvelopeMetadataParser.toJson(this); + } + + @Override + public EncryptionKeyMetadata copy() { + EnvelopeMetadata metadata = new EnvelopeMetadata(kekId(), wrappedDek(), + iv(), algorithm()); + metadata.setDek(dek()); + metadata.setKek(kek()); + return metadata; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + + if (other == null || getClass() != other.getClass()) { + return false; + } + + EnvelopeMetadata metadata = (EnvelopeMetadata) other; + return Objects.equals(kekId, metadata.kekId) && + Objects.equals(wrappedDek, metadata.wrappedDek) && + Arrays.equals(iv, metadata.iv) && + algorithm == metadata.algorithm && + Arrays.equals(kek, metadata.kek) && + Arrays.equals(dek, metadata.dek); + } + + @Override + public int hashCode() { + int result = Objects.hash(kekId, algorithm); + result = 31 * result + Objects.hashCode(wrappedDek); + result = 31 * result + Arrays.hashCode(iv); + result = 31 * result + Arrays.hashCode(kek); + result = 31 * result + Arrays.hashCode(dek); + return result; + } + + @Override + public String toString() { + return MoreObjects.toStringHelper(this) + .add("kekId", kekId) + .add("wrappedDek", wrappedDek) + .add("iv", iv) + .add("algorithm", algorithm) + .add("kek", kek == null ? "null" : "(redacted)") + .add("dek", dek == null ? "null" : "(redacted)") + .toString(); Review comment: Nit: indentation is off. We use 4 chars as the line wrap indentation. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeMetadata.java ########## @@ -0,0 +1,142 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Envelope encryption metadata used to encrypt and decrypt data. + * Information except plaintext KEK and DEK can be serialized and stored. + */ +public class EnvelopeMetadata implements EncryptionKeyMetadata { + + private final String kekId; + private final String wrappedDek; + private final byte[] iv; + private final EncryptionAlgorithm algorithm; + + // fields that are expected to be updated later + private byte[] kek; + private byte[] dek; + + public EnvelopeMetadata( + String kekId, + String wrappedDek, + ByteBuffer iv, + EncryptionAlgorithm algorithm) { + this.kekId = Preconditions.checkNotNull(kekId, + "Cannot construct envelope metadata because KEK ID is not specified"); + this.wrappedDek = wrappedDek; + this.iv = iv == null ? null : iv.array(); + this.algorithm = Preconditions.checkNotNull(algorithm, + "Cannot construct envelope metadata because encryption algorithm is not specified"); + } + + public String kekId() { + return kekId; + } + + public ByteBuffer kek() { + return kek == null ? null : ByteBuffer.wrap(kek); + } + + public ByteBuffer dek() { + return dek == null ? null : ByteBuffer.wrap(dek); + } + + public String wrappedDek() { + return wrappedDek; + } + + public ByteBuffer iv() { + return iv == null ? null : ByteBuffer.wrap(iv); + } + + public EncryptionAlgorithm algorithm() { + return algorithm; + } + + public void setKek(ByteBuffer kekBuffer) { + this.kek = kekBuffer == null ? null : kekBuffer.array(); + } + + public void setDek(ByteBuffer dekBuffer) { + this.dek = dekBuffer == null ? null : dekBuffer.array(); + } + + @Override + public ByteBuffer buffer() { + return EnvelopeMetadataParser.toJson(this); + } + + @Override + public EncryptionKeyMetadata copy() { + EnvelopeMetadata metadata = new EnvelopeMetadata(kekId(), wrappedDek(), + iv(), algorithm()); + metadata.setDek(dek()); + metadata.setKek(kek()); + return metadata; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + + if (other == null || getClass() != other.getClass()) { + return false; + } + + EnvelopeMetadata metadata = (EnvelopeMetadata) other; + return Objects.equals(kekId, metadata.kekId) && + Objects.equals(wrappedDek, metadata.wrappedDek) && + Arrays.equals(iv, metadata.iv) && + algorithm == metadata.algorithm && + Arrays.equals(kek, metadata.kek) && + Arrays.equals(dek, metadata.dek); Review comment: Nit: indentation is off. We use 4 chars as the line wrap indentation. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeMetadata.java ########## @@ -0,0 +1,142 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.util.Arrays; +import java.util.Objects; +import org.apache.iceberg.relocated.com.google.common.base.MoreObjects; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Envelope encryption metadata used to encrypt and decrypt data. + * Information except plaintext KEK and DEK can be serialized and stored. + */ +public class EnvelopeMetadata implements EncryptionKeyMetadata { + + private final String kekId; + private final String wrappedDek; + private final byte[] iv; + private final EncryptionAlgorithm algorithm; + + // fields that are expected to be updated later + private byte[] kek; + private byte[] dek; + + public EnvelopeMetadata( + String kekId, + String wrappedDek, + ByteBuffer iv, + EncryptionAlgorithm algorithm) { + this.kekId = Preconditions.checkNotNull(kekId, + "Cannot construct envelope metadata because KEK ID is not specified"); + this.wrappedDek = wrappedDek; + this.iv = iv == null ? null : iv.array(); + this.algorithm = Preconditions.checkNotNull(algorithm, + "Cannot construct envelope metadata because encryption algorithm is not specified"); + } + + public String kekId() { + return kekId; + } + + public ByteBuffer kek() { + return kek == null ? null : ByteBuffer.wrap(kek); + } + + public ByteBuffer dek() { + return dek == null ? null : ByteBuffer.wrap(dek); + } + + public String wrappedDek() { + return wrappedDek; + } + + public ByteBuffer iv() { + return iv == null ? null : ByteBuffer.wrap(iv); + } + + public EncryptionAlgorithm algorithm() { + return algorithm; + } + + public void setKek(ByteBuffer kekBuffer) { + this.kek = kekBuffer == null ? null : kekBuffer.array(); + } + + public void setDek(ByteBuffer dekBuffer) { + this.dek = dekBuffer == null ? null : dekBuffer.array(); + } + + @Override + public ByteBuffer buffer() { + return EnvelopeMetadataParser.toJson(this); + } + + @Override + public EncryptionKeyMetadata copy() { + EnvelopeMetadata metadata = new EnvelopeMetadata(kekId(), wrappedDek(), + iv(), algorithm()); + metadata.setDek(dek()); + metadata.setKek(kek()); + return metadata; + } + + @Override + public boolean equals(Object other) { + if (this == other) { + return true; + } + + if (other == null || getClass() != other.getClass()) { + return false; + } + + EnvelopeMetadata metadata = (EnvelopeMetadata) other; + return Objects.equals(kekId, metadata.kekId) && Review comment: Nit: extra space after `return` ########## File path: core/src/main/java/org/apache/iceberg/util/JsonUtil.java ########## @@ -160,6 +167,20 @@ public static void writeLongFieldIf(boolean condition, String key, Long value, J } } + public static void writeStringIf(boolean condition, String key, String value, JsonGenerator generator) + throws IOException { + if (condition) { + generator.writeStringField(key, value); + } + } + + public static void writeBinaryIf(boolean condition, String key, ByteBuffer value, JsonGenerator generator) + throws IOException { + if (condition) { + generator.writeStringField(key, StandardCharsets.UTF_8.decode(value).toString()); Review comment: Base64 makes more sense. The UTF may not human readable. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { + ((NativelyEncryptedFile) rawOutput).setNativeCryptoParameters(nativeEncryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively encrypt " + rawOutput.location() + " because the class " + + rawOutput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return new BaseEncryptedOutputFile(rawOutput, metadata); Review comment: Wondering how or when the raw output file is encrypted. Didn't see an encryption action here. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { + ((NativelyEncryptedFile) rawOutput).setNativeCryptoParameters(nativeEncryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively encrypt " + rawOutput.location() + " because the class " + + rawOutput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return new BaseEncryptedOutputFile(rawOutput, metadata); + } + + @Override + public InputFile decrypt(EncryptedInputFile encrypted) { + EnvelopeMetadata metadata = EnvelopeMetadataParser.fromJson(encrypted.keyMetadata().buffer()); + ByteBuffer fileDek = kmsClient.unwrapKey(metadata.wrappedDek(), metadata.kekId()); + + // Pushdown to data formats with native decryption support (such as Parquet) + NativeFileCryptoParameters nativeDecryptParams = NativeFileCryptoParameters.create(fileDek).build(); + InputFile rawInput = encrypted.encryptedInputFile(); + + try { + ((NativelyEncryptedFile) rawInput).setNativeCryptoParameters(nativeDecryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively decrypt " + rawInput.location() + " because the class " + + rawInput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return rawInput; + } + + private EnvelopeMetadata generateEnvelopeMetadata(EnvelopeConfig config) { + if (null == workerRNG) { + workerRNG = new SecureRandom(); + } + + String tableKekID = config.kekId(); + byte[] fileDek; + String wrappedFileDEK; + + if (kmsClient.supportsKeyGeneration()) { + KmsClient.KeyGenerationResult generatedDek = kmsClient.generateKey(tableKekID); + fileDek = generatedDek.key().array(); + wrappedFileDEK = generatedDek.wrappedKey(); + } else { + fileDek = new byte[dataKeyLength]; + workerRNG.nextBytes(fileDek); + wrappedFileDEK = kmsClient.wrapKey(ByteBuffer.wrap(fileDek), tableKekID); + } + + EnvelopeMetadata fileEnvelopeMetadata = new EnvelopeMetadata(tableKekID, wrappedFileDEK, null, + config.algorithm()); + fileEnvelopeMetadata.setDek(ByteBuffer.wrap(fileDek)); + + return fileEnvelopeMetadata; + } + + public static KmsClient loadKmsClient(String classPath, Map<String, String> properties) { Review comment: Are we going to have a dedicated Kms utility class in the future? This method should fit better there. I'm OK with the current solution though, if EnvelopeEncryptionManager is the only place to use KMS client. ########## File path: core/src/main/java/org/apache/iceberg/encryption/EnvelopeEncryptionManager.java ########## @@ -0,0 +1,146 @@ +/* + * 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.iceberg.encryption; + +import java.nio.ByteBuffer; +import java.security.SecureRandom; +import java.util.Map; +import org.apache.iceberg.common.DynConstructors; +import org.apache.iceberg.io.InputFile; +import org.apache.iceberg.io.OutputFile; +import org.apache.iceberg.relocated.com.google.common.base.Preconditions; + +/** + * Encryption manager that performs envelope encryption. + * <p> + */ +public class EnvelopeEncryptionManager implements EncryptionManager { + private final EnvelopeConfig dataEncryptionConfig; + private final boolean pushdown; + private final KmsClient kmsClient; + private final int dataKeyLength; + + private transient volatile SecureRandom workerRNG = null; + + public EnvelopeEncryptionManager( + boolean pushdown, EnvelopeConfig dataEncryptionConfig, + KmsClient kmsClient, int dataKeyLength) { + if (!pushdown) { + throw new UnsupportedOperationException("EnvelopeEncryptionManager currently supports only pushdown " + + "data encryption / decryption"); + } + this.dataEncryptionConfig = Preconditions.checkNotNull(dataEncryptionConfig, + "Cannot create EnvelopeEncryptionManager because data encryption config is not passed"); + Preconditions.checkNotNull(dataEncryptionConfig.kekId(), + "Cannot create EnvelopeEncryptionManager because table key encryption key ID is not specified"); + this.kmsClient = Preconditions.checkNotNull(kmsClient, + "Cannot create EnvelopeEncryptionManager because KmsClient is null"); + this.pushdown = pushdown; + this.dataKeyLength = dataKeyLength; + } + + @Override + public EncryptedOutputFile encrypt(OutputFile rawOutput) { + EnvelopeMetadata metadata = generateEnvelopeMetadata(dataEncryptionConfig); + + // Pushdown to data formats with native encryption support (such as Parquet) + NativeFileCryptoParameters nativeEncryptParams = NativeFileCryptoParameters.create(metadata.dek()) + .encryptionAlgorithm(metadata.algorithm().toString()) + .build(); + + try { + ((NativelyEncryptedFile) rawOutput).setNativeCryptoParameters(nativeEncryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively encrypt " + rawOutput.location() + " because the class " + + rawOutput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return new BaseEncryptedOutputFile(rawOutput, metadata); + } + + @Override + public InputFile decrypt(EncryptedInputFile encrypted) { + EnvelopeMetadata metadata = EnvelopeMetadataParser.fromJson(encrypted.keyMetadata().buffer()); + ByteBuffer fileDek = kmsClient.unwrapKey(metadata.wrappedDek(), metadata.kekId()); + + // Pushdown to data formats with native decryption support (such as Parquet) + NativeFileCryptoParameters nativeDecryptParams = NativeFileCryptoParameters.create(fileDek).build(); + InputFile rawInput = encrypted.encryptedInputFile(); + + try { + ((NativelyEncryptedFile) rawInput).setNativeCryptoParameters(nativeDecryptParams); + } catch (ClassCastException e) { + throw new RuntimeException("Can't natively decrypt " + rawInput.location() + " because the class " + + rawInput.getClass() + " doesn't implement NativelyEncryptedFile interface", e); + } + + return rawInput; + } + + private EnvelopeMetadata generateEnvelopeMetadata(EnvelopeConfig config) { + if (null == workerRNG) { + workerRNG = new SecureRandom(); + } + + String tableKekID = config.kekId(); + byte[] fileDek; + String wrappedFileDEK; + + if (kmsClient.supportsKeyGeneration()) { + KmsClient.KeyGenerationResult generatedDek = kmsClient.generateKey(tableKekID); + fileDek = generatedDek.key().array(); + wrappedFileDEK = generatedDek.wrappedKey(); + } else { + fileDek = new byte[dataKeyLength]; + workerRNG.nextBytes(fileDek); + wrappedFileDEK = kmsClient.wrapKey(ByteBuffer.wrap(fileDek), tableKekID); + } + + EnvelopeMetadata fileEnvelopeMetadata = new EnvelopeMetadata(tableKekID, wrappedFileDEK, null, + config.algorithm()); + fileEnvelopeMetadata.setDek(ByteBuffer.wrap(fileDek)); + + return fileEnvelopeMetadata; + } + + public static KmsClient loadKmsClient(String classPath, Map<String, String> properties) { Review comment: Maybe we can start to create one KMS utility. -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
