Core: Add StandardEncryptionManager (#9277)

Co-Authored-By: Jian Tang <jian_tang@apple.com>
Co-authored-by: Gidon Gershinsky <ggershinsky@apple.com>

api/src/main/java/org/apache/iceberg/encryption/EncryptedOutputFile.java

@@ -37,4 +37,9 @@ public interface EncryptedOutputFile {
    * #encryptingOutputFile()}.
    */
   EncryptionKeyMetadata keyMetadata();
+
+  /** Underlying output file for native encryption. */
+  default OutputFile plainOutputFile() {
+    throw new UnsupportedOperationException("Not implemented");
+  }
 }

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_IMPL = "encryption.kms-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/TableProperties.java

@@ -360,4 +360,11 @@ private TableProperties() {}
 
   public static final String UPSERT_ENABLED = "write.upsert.enabled";
   public static final boolean UPSERT_ENABLED_DEFAULT = false;
+
+  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/EncryptionUtil.java

@@ -0,0 +1,107 @@
+/*
+ * 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.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 KeyManagementClient createKmsClient(Map<String, String> catalogProperties) {
+    String kmsType = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_TYPE);
+    String kmsImpl = catalogProperties.get(CatalogProperties.ENCRYPTION_KMS_IMPL);
+
+    Preconditions.checkArgument(
+        kmsType == null || kmsImpl == null,
+        "Cannot set both KMS type (%s) and KMS impl (%s)",
+        kmsType,
+        kmsImpl);
+
+    // TODO: Add KMS implementations
+    Preconditions.checkArgument(kmsType == null, "Unsupported KMS type: %s", kmsType);
+
+    KeyManagementClient kmsClient;
+    DynConstructors.Ctor<KeyManagementClient> ctor;
+    try {
+      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",
+              kmsImpl),
+          e);
+    }
+
+    try {
+      kmsClient = ctor.newInstance();
+    } catch (ClassCastException e) {
+      throw new IllegalArgumentException(
+          String.format(
+              "Cannot initialize kms client, %s does not implement KeyManagementClient interface",
+              kmsImpl),
+          e);
+    }
+
+    kmsClient.initialize(catalogProperties);
+
+    return kmsClient;
+  }
+
+  public static EncryptionManager createEncryptionManager(
+      Map<String, String> tableProperties, KeyManagementClient kmsClient) {
+    Preconditions.checkArgument(kmsClient != null, "Invalid KMS client: null");
+    String tableKeyId = tableProperties.get(TableProperties.ENCRYPTION_TABLE_KEY);
+
+    if (null == tableKeyId) {
+      // Unencrypted table
+      return PlaintextEncryptionManager.instance();
+    }
+
+    String fileFormat =
+        PropertyUtil.propertyAsString(
+            tableProperties,
+            TableProperties.DEFAULT_FILE_FORMAT,
+            TableProperties.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,
+            TableProperties.ENCRYPTION_DEK_LENGTH,
+            TableProperties.ENCRYPTION_DEK_LENGTH_DEFAULT);
+
+    Preconditions.checkState(
+        dataKeyLength == 16 || dataKeyLength == 24 || dataKeyLength == 32,
+        "Invalid data key length: %s (must be 16, 24, or 32)",
+        dataKeyLength);
+
+    return new StandardEncryptionManager(tableKeyId, dataKeyLength, kmsClient);
+  }
+}

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<KeyMetadata> {
+class KeyMetadataDecoder extends MessageDecoder.BaseDecoder<StandardKeyMetadata> {
   private final org.apache.iceberg.Schema readSchema;
-  private final Map<Byte, RawDecoder<KeyMetadata>> decoders = new MapMaker().makeMap();
+  private final Map<Byte, RawDecoder<StandardKeyMetadata>> 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<KeyMetadata> {
    * 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<KeyMetadata> decoder = decoders.get(writeSchemaVersion);
+    RawDecoder<StandardKeyMetadata> decoder = decoders.get(writeSchemaVersion);
 
     if (decoder == null) {
       decoder = new RawDecoder<>(readSchema, GenericAvroReader::create, writeSchema);

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<KeyMetadata> {
+class KeyMetadataEncoder implements MessageEncoder<StandardKeyMetadata> {
   private static final ThreadLocal<BufferOutputStream> TEMP =
       ThreadLocal.withInitial(BufferOutputStream::new);
   private static final ThreadLocal<BinaryEncoder> ENCODER = new ThreadLocal<>();
 
   private final byte schemaVersion;
   private final boolean copyOutputBytes;
-  private final DatumWriter<KeyMetadata> writer;
+  private final DatumWriter<StandardKeyMetadata> 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.
    *
    * <p>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<KeyMetadata> {
   }
 
   /**
-   * 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.
    *
    * <p>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<KeyMetadata> {
    * 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<KeyMetadata> {
   }
 
   @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);

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

@@ -18,26 +18,33 @@
  */
 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 will be removed in 1.6.0. use {@link #instance()} instead. */
+  @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());
   }
 }