API: KMS client interface

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

@@ -0,0 +1,103 @@
+/*
+ * 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.nio.ByteBuffer;
+import java.util.Map;
+
+/**
+ * A minimum client interface to connect to a key management service (KMS).
+ */
+public interface KmsClient extends Serializable {
+
+  /**
+   * Wrap a secret key, using a wrapping/master key which is stored in KMS and referenced by an ID.
+   * Wrapping means encryption of the secret key with the master key, and adding optional KMS-specific metadata
+   * that allows the KMS to decrypt the secret key in an unwrapping call.
+   *
+   * @param key a secret key being wrapped
+   * @param wrappingKeyId a key ID that represents a wrapping key stored in KMS
+   * @return wrapped key material
+   */
+  String wrapKey(ByteBuffer key, String wrappingKeyId);
+
+  /**
+   * Some KMS systems support generation of secret keys inside the KMS server.
+   *
+   * @return true if KMS server supports key generation and KmsClient implementation
+   * is interested to leverage this capability. Otherwise, return false - Iceberg will
+   * then generate secret keys locally (using the SecureRandom mechanism) and call
+   * {@link #wrapKey(ByteBuffer, String)} to wrap them in KMS.
+   */
+  default boolean supportsKeyGeneration() {
+    return false;
+  }
+
+  /**
+   * Generate a new secret key in the KMS server, and wrap it using a wrapping/master key
+   * which is stored in KMS and referenced by an ID. This method will be called only if
+   * supportsKeyGeneration returns true.
+   *
+   * @param wrappingKeyId a key ID that represents a wrapping key stored in KMS
+   * @return key in two forms: raw, and wrapped with the given wrappingKeyId
+   */
+  default KeyGenerationResult generateKey(String wrappingKeyId) {
+    throw new UnsupportedOperationException("Key generation is not supported in this KmsClient");
+  }
+
+  /**
+   * Unwrap a secret key, using a wrapping/master key which is stored in KMS and referenced by an ID.
+   *
+   * @param wrappedKey wrapped key material (encrypted key and optional KMS metadata, returned by the wrapKey method)
+   * @param wrappingKeyId a key ID that represents a wrapping key stored in KMS
+   * @return raw key bytes
+   */
+  ByteBuffer unwrapKey(String wrappedKey, String wrappingKeyId);
+
+  /**
+   * Initialize the KMS client with given properties
+   *
+   * @param properties kms client properties
+   */
+  void initialize(Map<String, String> properties);
+
+  /**
+   * For KMS systems that support key generation, this class keeps the key generation result -
+   * the raw secret key, and its wrap.
+   */
+  class KeyGenerationResult {
+    private final ByteBuffer key;
+    private final String wrappedKey;
+
+    public KeyGenerationResult(ByteBuffer key, String wrappedKey) {
+      this.key = key;
+      this.wrappedKey = wrappedKey;
+    }
+
+    public ByteBuffer key() {
+      return key;
+    }
+
+    public String wrappedKey() {
+      return wrappedKey;
+    }
+  }
+}

core/src/test/java/org/apache/iceberg/encryption/kms/KeyStoreKmsClient.java

@@ -0,0 +1,118 @@
+/*
+ * 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.kms;
+
+import java.io.FileInputStream;
+import java.io.FileNotFoundException;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.KeyStoreException;
+import java.security.NoSuchAlgorithmException;
+import java.security.UnrecoverableKeyException;
+import java.security.cert.CertificateException;
+import java.util.Enumeration;
+import java.util.Map;
+import javax.crypto.SecretKey;
+import org.apache.iceberg.relocated.com.google.common.base.Preconditions;
+import org.apache.iceberg.relocated.com.google.common.collect.Maps;
+
+/**
+ * KMS client demo class, based on the Java KeyStore API that reads keys from standard PKCS12 keystore files.
+ * Not for use in production.
+ */
+public class KeyStoreKmsClient extends MemoryMockKMS {
+
+  // Path to keystore file. Preferably kept in volatile storage, such as ramdisk. Don't store with data.
+  public static final String KEYSTORE_FILE_PATH_PROP = "keystore.kms.client.file.path";
+
+  // Credentials (such as keystore password) must never be kept in a persistent storage.
+  // In this class, the password is passed as a system environment variable.
+  public static final String KEYSTORE_PASSWORD_ENV_VAR = "KEYSTORE_PASSWORD";
+
+  @Override
+  public String wrapKey(ByteBuffer key, String wrappingKeyId) {
+    // keytool keeps key names in lower case
+    return super.wrapKey(key, wrappingKeyId.toLowerCase());
+  }
+
+  @Override
+  public ByteBuffer unwrapKey(String wrappedKey, String wrappingKeyId) {
+    // keytool keeps key names in lower case
+    return super.unwrapKey(wrappedKey, wrappingKeyId.toLowerCase());
+  }
+
+  @Override
+  public void initialize(Map<String, String> properties) {
+    String keystorePath = properties.get(KEYSTORE_FILE_PATH_PROP);
+    Preconditions.checkNotNull(keystorePath, KEYSTORE_FILE_PATH_PROP + " must be set in hadoop or table " +
+        "properties");
+
+    String keystorePassword = System.getenv(KEYSTORE_PASSWORD_ENV_VAR);
+    Preconditions.checkNotNull(keystorePassword, KEYSTORE_PASSWORD_ENV_VAR + " environment variable " +
+        "must be set");
+
+    KeyStore keyStore;
+    try {
+      keyStore = KeyStore.getInstance("PKCS12");
+    } catch (KeyStoreException e) {
+      throw new RuntimeException("Failed to init keystore", e);
+    }
+
+    char[] pwdArray = keystorePassword.toCharArray();
+
+    FileInputStream fis;
+    try {
+      fis = new FileInputStream(keystorePath);
+    } catch (FileNotFoundException e) {
+      throw new RuntimeException("Failed to find keystore file " + keystorePath, e);
+    }
+
+    try {
+      keyStore.load(fis, pwdArray);
+    } catch (IOException | NoSuchAlgorithmException | CertificateException e) {
+      throw new RuntimeException("Failed to load keystore file " + keystorePath, e);
+    }
+
+    Enumeration<String> keyAliases;
+    try {
+      keyAliases =  keyStore.aliases();
+    } catch (KeyStoreException e) {
+      throw new RuntimeException("Failed to get key aliases in keystore file " + keystorePath, e);
+    }
+
+    masterKeys = Maps.newHashMap();
+    while (keyAliases.hasMoreElements()) {
+      String keyAlias = keyAliases.nextElement();
+      SecretKey secretKey;
+      try {
+        secretKey = (SecretKey) keyStore.getKey(keyAlias, pwdArray);
+      } catch (KeyStoreException | NoSuchAlgorithmException | UnrecoverableKeyException e) {
+        throw new RuntimeException("Failed to get key " + keyAlias, e);
+      }
+
+      masterKeys.put(keyAlias, secretKey.getEncoded());
+    }
+
+    if (masterKeys.isEmpty()) {
+      throw new RuntimeException("No keys found in " + keystorePath);
+    }
+  }
+}

core/src/test/java/org/apache/iceberg/encryption/kms/MemoryMockKMS.java

@@ -0,0 +1,57 @@
+/*
+ * 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.kms;
+
+import java.nio.ByteBuffer;
+import java.util.Base64;
+import java.util.Map;
+import org.apache.iceberg.encryption.Ciphers;
+import org.apache.iceberg.encryption.KmsClient;
+
+/**
+ * For testing and demonstrations; not for use in production.
+ */
+public abstract class MemoryMockKMS implements KmsClient {
+
+  protected Map<String, byte[]> masterKeys;
+
+  @Override
+  public String wrapKey(ByteBuffer key, String wrappingKeyId) {
+    byte[] wrappingKey = masterKeys.get(wrappingKeyId);
+    if (null == wrappingKey) {
+      throw new RuntimeException("Cannot wrap, because wrapping key " + wrappingKeyId + " is not found");
+    }
+    Ciphers.AesGcmEncryptor keyEncryptor = new Ciphers.AesGcmEncryptor(wrappingKey);
+    byte[] encryptedKey = keyEncryptor.encrypt(key.array(), null);
+    return Base64.getEncoder().encodeToString(encryptedKey);
+  }
+
+  @Override
+  public ByteBuffer unwrapKey(String wrappedKey, String wrappingKeyId) {
+    byte[] encryptedKey = Base64.getDecoder().decode(wrappedKey);
+    byte[] wrappingKey = masterKeys.get(wrappingKeyId);
+    if (null == wrappingKey) {
+      throw new RuntimeException("Cannot unwrap, because wrapping key " + wrappingKeyId + " is not found");
+    }
+    Ciphers.AesGcmDecryptor keyDecryptor = new Ciphers.AesGcmDecryptor(wrappingKey);
+    byte[] key = keyDecryptor.decrypt(encryptedKey, null);
+    return ByteBuffer.wrap(key);
+  }
+}