HBASE-29402: Comprehensive key management for encryption at rest

.gitignore

@@ -25,3 +25,9 @@ linklint/
 **/*.log
 tmp
 **/.flattened-pom.xml
+.*.sw*
+ID
+filenametags
+tags
+.codegenie
+.vscode

hbase-client/src/main/java/org/apache/hadoop/hbase/keymeta/KeymetaAdminClient.java

@@ -0,0 +1,86 @@
+/*
+ * 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.hadoop.hbase.keymeta;
+
+import java.io.IOException;
+import java.security.KeyException;
+import java.util.ArrayList;
+import java.util.List;
+
+import org.apache.hadoop.hbase.client.Connection;
+import org.apache.hadoop.hbase.io.crypto.ManagedKeyData;
+import org.apache.hadoop.hbase.io.crypto.ManagedKeyState;
+import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos;
+import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos.ManagedKeysRequest;
+import org.apache.hadoop.hbase.protobuf.generated.ManagedKeysProtos.ManagedKeysResponse;
+import org.apache.yetus.audience.InterfaceAudience;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import org.apache.hbase.thirdparty.com.google.protobuf.ServiceException;
+
+import org.apache.hadoop.hbase.shaded.protobuf.ProtobufUtil;
+
+@InterfaceAudience.Public
+public class KeymetaAdminClient implements KeymetaAdmin {
+  private static final Logger LOG = LoggerFactory.getLogger(KeymetaAdminClient.class);
+  private ManagedKeysProtos.ManagedKeysService.BlockingInterface stub;
+
+  public KeymetaAdminClient(Connection conn) throws IOException {
+    this.stub = ManagedKeysProtos.ManagedKeysService.newBlockingStub(
+        conn.getAdmin().coprocessorService());
+  }
+
+  @Override
+  public List<ManagedKeyData> enableKeyManagement(String keyCust, String keyNamespace)
+      throws IOException {
+    try {
+      ManagedKeysProtos.GetManagedKeysResponse response = stub.enableKeyManagement(null,
+        ManagedKeysRequest.newBuilder().setKeyCust(keyCust).setKeyNamespace(keyNamespace).build());
+      return generateKeyDataList(response);
+    } catch (ServiceException e) {
+      throw ProtobufUtil.handleRemoteException(e);
+    }
+  }
+
+  @Override
+  public List<ManagedKeyData> getManagedKeys(String keyCust, String keyNamespace)
+    throws IOException, KeyException {
+    try {
+      ManagedKeysProtos.GetManagedKeysResponse statusResponse = stub.getManagedKeys(null,
+        ManagedKeysRequest.newBuilder().setKeyCust(keyCust).setKeyNamespace(keyNamespace).build());
+      return generateKeyDataList(statusResponse);
+    } catch (ServiceException e) {
+      throw ProtobufUtil.handleRemoteException(e);
+    }
+  }
+
+  private static List<ManagedKeyData> generateKeyDataList(
+      ManagedKeysProtos.GetManagedKeysResponse stateResponse) {
+    List<ManagedKeyData> keyStates = new ArrayList<>();
+    for (ManagedKeysResponse state: stateResponse.getStateList()) {
+      keyStates.add(new ManagedKeyData(
+        state.getKeyCustBytes().toByteArray(),
+        state.getKeyNamespace(), null,
+        ManagedKeyState.forValue((byte) state.getKeyState().getNumber()),
+        state.getKeyMetadata(),
+        state.getRefreshTimestamp(), state.getReadOpCount(), state.getWriteOpCount()));
+    }
+    return keyStates;
+  }
+}

hbase-client/src/main/java/org/apache/hadoop/hbase/security/EncryptionUtil.java

@@ -80,6 +80,21 @@ public static byte[] wrapKey(Configuration conf, byte[] key, String algorithm)
    * @return the encrypted key bytes
    */
   public static byte[] wrapKey(Configuration conf, String subject, Key key) throws IOException {
+    return wrapKey(conf, subject, key, null);
+  }
+
+  /**
+   * Protect a key by encrypting it with the secret key of the given subject or kek. The
+   * configuration must be set up correctly for key alias resolution. Only one of the
+   * {@code subject} or {@code kek} needs to be specified and the other one can be {@code null}.
+   * @param conf    configuration
+   * @param subject subject key alias
+   * @param key     the key
+   * @param kek     the key encryption key
+   * @return the encrypted key bytes
+   */
+  public static byte[] wrapKey(Configuration conf, String subject, Key key, Key kek)
+      throws IOException {
     // Wrap the key with the configured encryption algorithm.
     String algorithm = conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
     Cipher cipher = Encryption.getCipher(conf, algorithm);
@@ -100,8 +115,13 @@ public static byte[] wrapKey(Configuration conf, String subject, Key key) throws
     builder
       .setHash(UnsafeByteOperations.unsafeWrap(Encryption.computeCryptoKeyHash(conf, keyBytes)));
     ByteArrayOutputStream out = new ByteArrayOutputStream();
-    Encryption.encryptWithSubjectKey(out, new ByteArrayInputStream(keyBytes), subject, conf, cipher,
-      iv);
+    if (kek != null) {
+      Encryption.encryptWithGivenKey(kek, out, new ByteArrayInputStream(keyBytes), cipher, iv);
+    }
+    else {
+      Encryption.encryptWithSubjectKey(out, new ByteArrayInputStream(keyBytes), subject, conf,
+        cipher, iv);
+    }
     builder.setData(UnsafeByteOperations.unsafeWrap(out.toByteArray()));
     // Build and return the protobuf message
     out.reset();
@@ -118,6 +138,21 @@ public static byte[] wrapKey(Configuration conf, String subject, Key key) throws
    * @return the raw key bytes
    */
   public static Key unwrapKey(Configuration conf, String subject, byte[] value)
+    throws IOException, KeyException {
+    return unwrapKey(conf, subject, value, null);
+  }
+
+  /**
+   * Unwrap a key by decrypting it with the secret key of the given subject. The configuration must
+   * be set up correctly for key alias resolution. Only one of the {@code subject} or {@code kek}
+   * needs to be specified and the other one can be {@code null}.
+   * @param conf    configuration
+   * @param subject subject key alias
+   * @param value   the encrypted key bytes
+   * @param kek   the key encryption key
+   * @return the raw key bytes
+   */
+  public static Key unwrapKey(Configuration conf, String subject, byte[] value, Key kek)
     throws IOException, KeyException {
     EncryptionProtos.WrappedKey wrappedKey =
       EncryptionProtos.WrappedKey.parser().parseDelimitedFrom(new ByteArrayInputStream(value));
@@ -126,11 +161,12 @@ public static Key unwrapKey(Configuration conf, String subject, byte[] value)
     if (cipher == null) {
       throw new RuntimeException("Cipher '" + algorithm + "' not available");
     }
-    return getUnwrapKey(conf, subject, wrappedKey, cipher);
+    return getUnwrapKey(conf, subject, wrappedKey, cipher, kek);
   }
 
   private static Key getUnwrapKey(Configuration conf, String subject,
-    EncryptionProtos.WrappedKey wrappedKey, Cipher cipher) throws IOException, KeyException {
+      EncryptionProtos.WrappedKey wrappedKey, Cipher cipher, Key kek)
+      throws IOException, KeyException {
     String configuredHashAlgorithm = Encryption.getConfiguredHashAlgorithm(conf);
     String wrappedHashAlgorithm = wrappedKey.getHashAlgorithm().trim();
     if (!configuredHashAlgorithm.equalsIgnoreCase(wrappedHashAlgorithm)) {
@@ -143,8 +179,14 @@ private static Key getUnwrapKey(Configuration conf, String subject,
     }
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     byte[] iv = wrappedKey.hasIv() ? wrappedKey.getIv().toByteArray() : null;
-    Encryption.decryptWithSubjectKey(out, wrappedKey.getData().newInput(), wrappedKey.getLength(),
-      subject, conf, cipher, iv);
+    if (kek != null) {
+      Encryption.decryptWithGivenKey(kek, out, wrappedKey.getData().newInput(),
+          wrappedKey.getLength(), cipher, iv);
+    }
+    else {
+      Encryption.decryptWithSubjectKey(out, wrappedKey.getData().newInput(), wrappedKey.getLength(),
+        subject, conf, cipher, iv);
+    }
     byte[] keyBytes = out.toByteArray();
     if (wrappedKey.hasHash()) {
       if (
@@ -176,7 +218,7 @@ public static Key unwrapWALKey(Configuration conf, String subject, byte[] value)
     if (cipher == null) {
       throw new RuntimeException("Cipher '" + algorithm + "' not available");
     }
-    return getUnwrapKey(conf, subject, wrappedKey, cipher);
+    return getUnwrapKey(conf, subject, wrappedKey, cipher, null);
   }
 
   /**

hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java

@@ -1192,6 +1192,11 @@ public enum OperationStatusCode {
 
   /** Temporary directory used for table creation and deletion */
   public static final String HBASE_TEMP_DIRECTORY = ".tmp";
+  /**
+   * Directory used for storing master keys for the cluster
+   */
+  public static final String SYSTEM_KEYS_DIRECTORY = ".system_keys";
+  public static final String SYSTEM_KEY_FILE_PREFIX = "system_key.";
   /**
    * The period (in milliseconds) between computing region server point in time metrics
    */
@@ -1304,6 +1309,39 @@ public enum OperationStatusCode {
   /** Configuration key for enabling WAL encryption, a boolean */
   public static final String ENABLE_WAL_ENCRYPTION = "hbase.regionserver.wal.encryption";
 
+  /** Property used by ManagedKeyStoreKeyProvider class to set the alias that identifies
+   *  the current system key. */
+  public static final String CRYPTO_MANAGED_KEY_STORE_SYSTEM_KEY_NAME_CONF_KEY =
+    "hbase.crypto.managed_key_store.system.key.name";
+  public static final String CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX =
+    "hbase.crypto.managed_key_store.cust.";
+
+  /** Enables or disables the key management feature. */
+  public static final String CRYPTO_MANAGED_KEYS_ENABLED_CONF_KEY =
+    "hbase.crypto.managed_keys.enabled";
+  public static final boolean CRYPTO_MANAGED_KEYS_DEFAULT_ENABLED = false;
+
+  /** The number of keys to retrieve from Key Provider per each custodian and namespace
+   *  combination. */
+  public static final String CRYPTO_MANAGED_KEYS_PER_CUST_NAMESPACE_ACTIVE_KEY_COUNT =
+    "hbase.crypto.managed_keys.per_cust_namespace.active_count";
+  public static final int CRYPTO_MANAGED_KEYS_PER_CUST_NAMESPACE_ACTIVE_KEY_DEFAULT_COUNT = 1;
+  /** Enables or disables key lookup during data path as an alternative to static injection of keys
+   *  using control path. */
+  public static final String CRYPTO_MANAGED_KEYS_DYNAMIC_LOOKUP_ENABLED_CONF_KEY =
+    "hbase.crypto.managed_keys.dynamic_lookup.enabled";
+  public static final boolean CRYPTO_MANAGED_KEYS_DYNAMIC_LOOKUP_DEFAULT_ENABLED = true;
+
+  /** Maximum number of entries in the managed key data cache. */
+  public static final String CRYPTO_MANAGED_KEYS_L1_CACHE_MAX_ENTRIES_CONF_KEY =
+    "hbase.crypto.managed_keys.l1_cache.max_entries";
+  public static final int CRYPTO_MANAGED_KEYS_L1_CACHE_MAX_ENTRIES_DEFAULT = 1000;
+
+  /** Maximum number of entries in the managed key active keys cache. */
+  public static final String CRYPTO_MANAGED_KEYS_L1_ACTIVE_CACHE_MAX_NS_ENTRIES_CONF_KEY =
+    "hbase.crypto.managed_keys.l1_active_cache.max_ns_entries";
+  public static final int CRYPTO_MANAGED_KEYS_L1_ACTIVE_CACHE_MAX_NS_ENTRIES_DEFAULT = 100;
+
   /** Configuration key for setting RPC codec class name */
   public static final String RPC_CODEC_CONF_KEY = "hbase.client.rpc.codec";
 

hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/Encryption.java

@@ -33,8 +33,10 @@
 import javax.crypto.spec.PBEKeySpec;
 import javax.crypto.spec.SecretKeySpec;
 import org.apache.commons.io.IOUtils;
+import org.apache.hadoop.classification.InterfaceAudience.Private;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HBaseInterfaceAudience;
 import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.io.crypto.aes.AES;
 import org.apache.hadoop.hbase.util.Bytes;
@@ -468,6 +470,19 @@ public static void encryptWithSubjectKey(OutputStream out, InputStream in, Strin
     if (key == null) {
       throw new IOException("No key found for subject '" + subject + "'");
     }
+    encryptWithGivenKey(key, out, in, cipher, iv);
+  }
+
+  /**
+   * Encrypts a block of plaintext with the specified symmetric key.
+   * @param key    The symmetric key
+   * @param out    ciphertext
+   * @param in     plaintext
+   * @param cipher the encryption algorithm
+   * @param iv     the initialization vector, can be null
+   */
+  public static void encryptWithGivenKey(Key key, OutputStream out, InputStream in,
+      Cipher cipher, byte[] iv) throws IOException {
     Encryptor e = cipher.getEncryptor();
     e.setKey(key);
     e.setIv(iv); // can be null
@@ -490,36 +505,39 @@ public static void decryptWithSubjectKey(OutputStream out, InputStream in, int o
     if (key == null) {
       throw new IOException("No key found for subject '" + subject + "'");
     }
-    Decryptor d = cipher.getDecryptor();
-    d.setKey(key);
-    d.setIv(iv); // can be null
     try {
-      decrypt(out, in, outLen, d);
+      decryptWithGivenKey(key, out, in, outLen, cipher, iv);
     } catch (IOException e) {
       // If the current cipher algorithm fails to unwrap, try the alternate cipher algorithm, if one
       // is configured
       String alternateAlgorithm = conf.get(HConstants.CRYPTO_ALTERNATE_KEY_ALGORITHM_CONF_KEY);
       if (alternateAlgorithm != null) {
         if (LOG.isDebugEnabled()) {
-          LOG.debug("Unable to decrypt data with current cipher algorithm '"
-            + conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES)
+          LOG.debug("Unable to decrypt data with current cipher algorithm '" + conf.get(
+            HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES)
             + "'. Trying with the alternate cipher algorithm '" + alternateAlgorithm
             + "' configured.");
         }
         Cipher alterCipher = Encryption.getCipher(conf, alternateAlgorithm);
         if (alterCipher == null) {
           throw new RuntimeException("Cipher '" + alternateAlgorithm + "' not available");
         }
-        d = alterCipher.getDecryptor();
-        d.setKey(key);
-        d.setIv(iv); // can be null
-        decrypt(out, in, outLen, d);
-      } else {
-        throw new IOException(e);
+        decryptWithGivenKey(key, out, in, outLen, alterCipher, iv);
+      }
+      else {
+        throw e;
       }
     }
   }
 
+  public static void decryptWithGivenKey(Key key, OutputStream out, InputStream in, int outLen,
+      Cipher cipher, byte[] iv) throws IOException {
+    Decryptor d = cipher.getDecryptor();
+    d.setKey(key);
+    d.setIv(iv); // can be null
+    decrypt(out, in, outLen, d);
+  }
+
   private static ClassLoader getClassLoaderForClass(Class<?> c) {
     ClassLoader cl = Thread.currentThread().getContextClassLoader();
     if (cl == null) {
@@ -561,6 +579,9 @@ public static KeyProvider getKeyProvider(Configuration conf) {
       provider = (KeyProvider) ReflectionUtils
         .newInstance(getClassLoaderForClass(KeyProvider.class).loadClass(providerClassName), conf);
       provider.init(providerParameters);
+      if (provider instanceof ManagedKeyProvider) {
+        ((ManagedKeyProvider) provider).initConfig(conf);
+      }
       if (LOG.isDebugEnabled()) {
         LOG.debug("Installed " + providerClassName + " into key provider cache");
       }
@@ -571,6 +592,11 @@ public static KeyProvider getKeyProvider(Configuration conf) {
     }
   }
 
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public static void clearKeyProviderCache() {
+    keyProviderCache.clear();
+  }
+
   public static void incrementIv(byte[] iv) {
     incrementIv(iv, 1);
   }

hbase-common/src/main/java/org/apache/hadoop/hbase/io/crypto/KeyStoreKeyProvider.java

@@ -76,6 +76,8 @@
 @InterfaceAudience.Public
 public class KeyStoreKeyProvider implements KeyProvider {
 
+  private static final char[] NO_PASSWORD = new char[0];
+
   protected KeyStore store;
   protected char[] password; // can be null if no password
   protected Properties passwordFile; // can be null if no file provided
@@ -172,9 +174,15 @@ protected char[] getAliasPassword(String alias) {
 
   @Override
   public Key getKey(String alias) {
+    // First try with no password, as it is more common to have a password only for the store.
     try {
-      return store.getKey(alias, getAliasPassword(alias));
+      return store.getKey(alias, NO_PASSWORD);
     } catch (UnrecoverableKeyException e) {
+      try {
+        return store.getKey(alias, getAliasPassword(alias));
+      } catch (UnrecoverableKeyException|NoSuchAlgorithmException|KeyStoreException e2) {
+        // Ignore.
+      }
       throw new RuntimeException(e);
     } catch (KeyStoreException e) {
       throw new RuntimeException(e);