Standard key manager

core/src/main/java/org/apache/iceberg/CatalogProperties.java

@@ -156,4 +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";
 }

core/src/main/java/org/apache/iceberg/TableOperations.java

@@ -71,7 +71,7 @@ public interface TableOperations {
    * files.
    */
   default EncryptionManager encryption() {
-    return new PlaintextEncryptionManager();
+    return PlaintextEncryptionManager.instance();
   }
 
   /**

core/src/main/java/org/apache/iceberg/encryption/EncryptedFiles.java

@@ -45,6 +45,13 @@ public static EncryptedOutputFile encryptedOutput(
     return new BaseEncryptedOutputFile(encryptingOutputFile, keyMetadata);
   }
 
+  public static EncryptedOutputFile encryptedOutput(
+      OutputFile encryptingOutputFile,
+      EncryptionKeyMetadata keyMetadata,
+      OutputFile rawOutputFile) {
+    return new BaseEncryptedOutputFile(encryptingOutputFile, keyMetadata, rawOutputFile);
+  }
+
   public static EncryptedOutputFile encryptedOutput(
       OutputFile encryptingOutputFile, ByteBuffer keyMetadata) {
     return encryptedOutput(
@@ -57,5 +64,9 @@ public static EncryptedOutputFile encryptedOutput(
         encryptedOutputFile, BaseEncryptionKeyMetadata.fromByteArray(keyMetadata));
   }
 
+  public static EncryptedOutputFile plainAsEncryptedOutput(OutputFile encryptingOutputFile) {
+    return new BaseEncryptedOutputFile(encryptingOutputFile, EncryptionKeyMetadata.empty());
+  }
+
   private EncryptedFiles() {}
 }

core/src/main/java/org/apache/iceberg/encryption/EncryptionProperties.java

@@ -0,0 +1,31 @@
+/*
+ * 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;
+
+public class EncryptionProperties {
+
+  private EncryptionProperties() {}
+
+  public static final String ENCRYPTION_TABLE_KEY = "encryption.key-id";
+
+  public static final String ENCRYPTION_DEK_LENGTH = "encryption.data-key-length";
+  public static final int ENCRYPTION_DEK_LENGTH_DEFAULT = 16;
+
+  public static final int ENCRYPTION_AAD_LENGTH_DEFAULT = 16;
+}

core/src/main/java/org/apache/iceberg/encryption/KeyMetadata.java

@@ -76,6 +76,20 @@ ByteBuffer aadPrefix() {
     return aadPrefix;
   }
 
+  static KeyMetadata castOrParse(EncryptionKeyMetadata keyMetadata) {
+    if (keyMetadata instanceof KeyMetadata) {
+      return (KeyMetadata) keyMetadata;
+    }
+
+    ByteBuffer kmBuffer = keyMetadata.buffer();
+
+    if (kmBuffer == null) {
+      throw new IllegalStateException("Null key metadata buffer");
+    }
+
+    return parse(kmBuffer);
+  }
+
   static KeyMetadata parse(ByteBuffer buffer) {
     try {
       return KEY_METADATA_DECODER.decode(buffer);

core/src/main/java/org/apache/iceberg/encryption/PlaintextEncryptionManager.java

@@ -18,26 +18,32 @@
  */
 package org.apache.iceberg.encryption;
 
-import java.nio.ByteBuffer;
 import org.apache.iceberg.io.InputFile;
 import org.apache.iceberg.io.OutputFile;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 public class PlaintextEncryptionManager implements EncryptionManager {
+  private static final EncryptionManager INSTANCE = new PlaintextEncryptionManager();
   private static final Logger LOG = LoggerFactory.getLogger(PlaintextEncryptionManager.class);
 
+  @Deprecated
+  public PlaintextEncryptionManager() {}
+
+  public static EncryptionManager instance() {
+    return INSTANCE;
+  }
+
   @Override
   public InputFile decrypt(EncryptedInputFile encrypted) {
     if (encrypted.keyMetadata().buffer() != null) {
-      LOG.warn(
-          "File encryption key metadata is present, but currently using PlaintextEncryptionManager.");
+      LOG.warn("File encryption key metadata is present, but no encryption has been configured.");
     }
     return encrypted.encryptedInputFile();
   }
 
   @Override
   public EncryptedOutputFile encrypt(OutputFile rawOutput) {
-    return EncryptedFiles.encryptedOutput(rawOutput, (ByteBuffer) null);
+    return EncryptedFiles.encryptedOutput(rawOutput, EncryptionKeyMetadata.empty());
   }
 }

core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManager.java

@@ -0,0 +1,109 @@
+/*
+ * 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 org.apache.iceberg.io.InputFile;
+import org.apache.iceberg.io.OutputFile;
+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;
+
+public class StandardEncryptionManager implements EncryptionManager {
+  private final transient KeyManagementClient kmsClient;
+  private final String tableKeyId;
+  private final int dataKeyLength;
+
+  private transient volatile SecureRandom lazyRNG = null;
+
+  /**
+   * @param tableKeyId table encryption key id
+   * @param dataKeyLength length of data encryption key (16/24/32 bytes)
+   * @param kmsClient Client of KMS used to wrap/unwrap keys in envelope encryption
+   */
+  public StandardEncryptionManager(
+      String tableKeyId, int dataKeyLength, KeyManagementClient kmsClient) {
+    Preconditions.checkNotNull(tableKeyId, "Invalid encryption key ID: null");
+    Preconditions.checkNotNull(kmsClient, "Invalid KMS client: null");
+    this.tableKeyId = tableKeyId;
+    this.kmsClient = kmsClient;
+    this.dataKeyLength = dataKeyLength;
+  }
+
+  @Override
+  public EncryptedOutputFile encrypt(OutputFile rawOutput) {
+    ByteBuffer fileDek = ByteBuffer.allocate(dataKeyLength);
+    workerRNG().nextBytes(fileDek.array());
+
+    ByteBuffer aadPrefix = ByteBuffer.allocate(EncryptionProperties.ENCRYPTION_AAD_LENGTH_DEFAULT);
+    workerRNG().nextBytes(aadPrefix.array());
+
+    KeyMetadata encryptionMetadata = new KeyMetadata(fileDek, aadPrefix);
+
+    return EncryptedFiles.encryptedOutput(
+        new AesGcmOutputFile(rawOutput, fileDek.array(), aadPrefix.array()),
+        encryptionMetadata,
+        rawOutput);
+  }
+
+  @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);
+  }
+
+  @Override
+  public Iterable<InputFile> decrypt(Iterable<EncryptedInputFile> 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();
+  }
+
+  private SecureRandom workerRNG() {
+    if (this.lazyRNG == null) {
+      this.lazyRNG = new SecureRandom();
+    }
+
+    return lazyRNG;
+  }
+
+  public ByteBuffer wrapKey(ByteBuffer secretKey) {
+    if (kmsClient == null) {
+      throw new IllegalStateException("Null KmsClient. WrapKey can't be called from workers");
+    }
+
+    return kmsClient.wrapKey(secretKey, tableKeyId);
+  }
+
+  public ByteBuffer unwrapKey(ByteBuffer wrappedSecretKey) {
+    if (kmsClient == null) {
+      throw new IllegalStateException("Null KmsClient. UnwrapKey can't be called from workers");
+    }
+
+    return kmsClient.unwrapKey(wrappedSecretKey, tableKeyId);
+  }
+}

core/src/main/java/org/apache/iceberg/encryption/StandardEncryptionManagerFactory.java

@@ -0,0 +1,90 @@
+/*
+ * 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 static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT;
+import static org.apache.iceberg.TableProperties.DEFAULT_FILE_FORMAT_DEFAULT;
+
+import java.io.Closeable;
+import java.io.IOException;
+import java.util.Map;
+import org.apache.iceberg.CatalogProperties;
+import org.apache.iceberg.FileFormat;
+import org.apache.iceberg.util.PropertyUtil;
+
+public class StandardEncryptionManagerFactory implements Closeable {
+  private final KeyManagementClient kmsClient;
+
+  public StandardEncryptionManagerFactory(Map<String, String> catalogProperties) {
+    String kmsType = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_TYPE);
+
+    if (kmsType == null) {
+      throw new IllegalStateException("Cannot create StandardEncryptionManagerFactory without KMS type");
+    }
+
+    if (kmsType.equals(CatalogProperties.ENCRYPTION_KMS_CUSTOM_TYPE)) {
+      String kmsClientImpl = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_CLIENT_IMPL);
+      if (kmsClientImpl == null) {
+        throw new IllegalStateException("Custom KMS client class is not defined");
+      }
+      kmsClient = EncryptionUtil.createKmsClient(kmsClientImpl);
+      kmsClient.initialize(catalogProperties);
+    } else {
+      // Currently support only custom types
+      throw new UnsupportedOperationException("Undefined KMS type " + kmsType);
+    }
+  }
+
+  public EncryptionManager create(Map<String, String> tableProperties) {
+    String tableKeyId = tableProperties.get(EncryptionProperties.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);
+
+    if (FileFormat.fromString(fileFormat) != FileFormat.PARQUET) {
+      throw new UnsupportedOperationException(
+          "Iceberg encryption currently supports only parquet format for data files");
+    }
+
+    int dataKeyLength =
+        PropertyUtil.propertyAsInt(
+            tableProperties,
+            EncryptionProperties.ENCRYPTION_DEK_LENGTH,
+            EncryptionProperties.ENCRYPTION_DEK_LENGTH_DEFAULT);
+
+    return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
+  }
+
+  @Override
+  public synchronized void close() throws IOException {
+    if (kmsClient != null) {
+      kmsClient.close();
+    }
+  }
+}