HBASE-29368 [Feature] Key management for encryption at rest (MVP changes)

hbase-client/src/main/java/org/apache/hadoop/hbase/client/Admin.java

@@ -2705,4 +2705,28 @@ List<LogEntry> getLogEntries(Set<ServerName> serverNames, String logType, Server
 
   @InterfaceAudience.Private
   void restoreBackupSystemTable(String snapshotName) throws IOException;
+
+  /**
+   * Refresh the system key cache on all specified region servers.
+   * @param regionServers the list of region servers to refresh the system key cache on
+   */
+  void refreshSystemKeyCacheOnServers(List<ServerName> regionServers) throws IOException;
+
+  /**
+   * Eject a specific managed key entry from the managed key data cache on all specified region
+   * servers.
+   * @param regionServers the list of region servers to eject the managed key entry from
+   * @param keyCustodian  the key custodian
+   * @param keyNamespace  the key namespace
+   * @param keyMetadata   the key metadata
+   */
+  void ejectManagedKeyDataCacheEntryOnServers(List<ServerName> regionServers, byte[] keyCustodian,
+    String keyNamespace, String keyMetadata) throws IOException;
+
+  /**
+   * Clear all entries in the managed key data cache on all specified region servers without having
+   * to restart the process.
+   * @param regionServers the list of region servers to clear the managed key data cache on
+   */
+  void clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) throws IOException;
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AdminOverAsyncAdmin.java

@@ -1157,4 +1157,21 @@ public List<String> getCachedFilesList(ServerName serverName) throws IOException
   public void restoreBackupSystemTable(String snapshotName) throws IOException {
     get(admin.restoreBackupSystemTable(snapshotName));
   }
+
+  @Override
+  public void refreshSystemKeyCacheOnServers(List<ServerName> regionServers) throws IOException {
+    get(admin.refreshSystemKeyCacheOnServers(regionServers));
+  }
+
+  @Override
+  public void ejectManagedKeyDataCacheEntryOnServers(List<ServerName> regionServers,
+    byte[] keyCustodian, String keyNamespace, String keyMetadata) throws IOException {
+    get(admin.ejectManagedKeyDataCacheEntryOnServers(regionServers, keyCustodian, keyNamespace,
+      keyMetadata));
+  }
+
+  @Override
+  public void clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) throws IOException {
+    get(admin.clearManagedKeyDataCacheOnServers(regionServers));
+  }
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncAdmin.java

@@ -1892,4 +1892,27 @@ CompletableFuture<List<LogEntry>> getLogEntries(Set<ServerName> serverNames, Str
 
   @InterfaceAudience.Private
   CompletableFuture<Void> restoreBackupSystemTable(String snapshotName);
+
+  /**
+   * Refresh the system key cache on all specified region servers.
+   * @param regionServers the list of region servers to refresh the system key cache on
+   */
+  CompletableFuture<Void> refreshSystemKeyCacheOnServers(List<ServerName> regionServers);
+
+  /**
+   * Eject a specific managed key entry from the managed key data cache on all specified region
+   * servers.
+   * @param regionServers the list of region servers to eject the managed key entry from
+   * @param keyCustodian  the key custodian
+   * @param keyNamespace  the key namespace
+   * @param keyMetadata   the key metadata
+   */
+  CompletableFuture<Void> ejectManagedKeyDataCacheEntryOnServers(List<ServerName> regionServers,
+    byte[] keyCustodian, String keyNamespace, String keyMetadata);
+
+  /**
+   * Clear all entries in the managed key data cache on all specified region servers.
+   * @param regionServers the list of region servers to clear the managed key data cache on
+   */
+  CompletableFuture<Void> clearManagedKeyDataCacheOnServers(List<ServerName> regionServers);
 }

hbase-client/src/main/java/org/apache/hadoop/hbase/client/AsyncHBaseAdmin.java

@@ -696,6 +696,23 @@ public CompletableFuture<Void> updateConfiguration(String groupName) {
     return wrap(rawAdmin.updateConfiguration(groupName));
   }
 
+  @Override
+  public CompletableFuture<Void> refreshSystemKeyCacheOnServers(List<ServerName> regionServers) {
+    return wrap(rawAdmin.refreshSystemKeyCacheOnServers(regionServers));
+  }
+
+  @Override
+  public CompletableFuture<Void> ejectManagedKeyDataCacheEntryOnServers(
+    List<ServerName> regionServers, byte[] keyCustodian, String keyNamespace, String keyMetadata) {
+    return wrap(rawAdmin.ejectManagedKeyDataCacheEntryOnServers(regionServers, keyCustodian,
+      keyNamespace, keyMetadata));
+  }
+
+  @Override
+  public CompletableFuture<Void> clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) {
+    return wrap(rawAdmin.clearManagedKeyDataCacheOnServers(regionServers));
+  }
+
   @Override
   public CompletableFuture<Void> rollWALWriter(ServerName serverName) {
     return wrap(rawAdmin.rollWALWriter(serverName));

hbase-client/src/main/java/org/apache/hadoop/hbase/client/RawAsyncHBaseAdmin.java

@@ -76,6 +76,7 @@
 import org.apache.hadoop.hbase.client.replication.TableCFs;
 import org.apache.hadoop.hbase.client.security.SecurityCapability;
 import org.apache.hadoop.hbase.exceptions.DeserializationException;
+import org.apache.hadoop.hbase.io.crypto.ManagedKeyData;
 import org.apache.hadoop.hbase.ipc.HBaseRpcController;
 import org.apache.hadoop.hbase.net.Address;
 import org.apache.hadoop.hbase.quotas.QuotaFilter;
@@ -150,7 +151,10 @@
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.AdminProtos.UpdateConfigurationResponse;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.EmptyMsg;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.LastHighestWalFilenum;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ManagedKeyEntryRequest;
+import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ManagedKeyRequest;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.NameStringPair;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.ProcedureDescription;
 import org.apache.hadoop.hbase.shaded.protobuf.generated.HBaseProtos.RegionSpecifier.RegionSpecifierType;
@@ -4697,4 +4701,86 @@ MasterProtos.RestoreBackupSystemTableResponse> procedureCall(request,
         MasterProtos.RestoreBackupSystemTableResponse::getProcId,
         new RestoreBackupSystemTableProcedureBiConsumer());
   }
+
+  @Override
+  public CompletableFuture<Void> refreshSystemKeyCacheOnServers(List<ServerName> regionServers) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(this::refreshSystemKeyCache).collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> refreshSystemKeyCache(ServerName serverName) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<EmptyMsg, EmptyMsg, Void> adminCall(controller, stub,
+        EmptyMsg.getDefaultInstance(),
+        (s, c, req, done) -> s.refreshSystemKeyCache(controller, req, done), resp -> null))
+      .serverName(serverName).call();
+  }
+
+  @Override
+  public CompletableFuture<Void> ejectManagedKeyDataCacheEntryOnServers(
+    List<ServerName> regionServers, byte[] keyCustodian, String keyNamespace, String keyMetadata) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    // Create the request once instead of repeatedly for each server
+    byte[] keyMetadataHash = ManagedKeyData.constructMetadataHash(keyMetadata);
+    ManagedKeyEntryRequest request = ManagedKeyEntryRequest.newBuilder()
+      .setKeyCustNs(ManagedKeyRequest.newBuilder().setKeyCust(ByteString.copyFrom(keyCustodian))
+        .setKeyNamespace(keyNamespace).build())
+      .setKeyMetadataHash(ByteString.copyFrom(keyMetadataHash)).build();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(serverName -> ejectManagedKeyDataCacheEntry(serverName, request))
+        .collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> ejectManagedKeyDataCacheEntry(ServerName serverName,
+    ManagedKeyEntryRequest request) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<ManagedKeyEntryRequest, HBaseProtos.BooleanMsg,
+        Void> adminCall(controller, stub, request,
+          (s, c, req, done) -> s.ejectManagedKeyDataCacheEntry(controller, req, done),
+          resp -> null))
+      .serverName(serverName).call();
+  }
+
+  @Override
+  public CompletableFuture<Void> clearManagedKeyDataCacheOnServers(List<ServerName> regionServers) {
+    CompletableFuture<Void> future = new CompletableFuture<>();
+    List<CompletableFuture<Void>> futures =
+      regionServers.stream().map(this::clearManagedKeyDataCache).collect(Collectors.toList());
+    addListener(CompletableFuture.allOf(futures.toArray(new CompletableFuture<?>[0])),
+      (result, err) -> {
+        if (err != null) {
+          future.completeExceptionally(err);
+        } else {
+          future.complete(result);
+        }
+      });
+    return future;
+  }
+
+  private CompletableFuture<Void> clearManagedKeyDataCache(ServerName serverName) {
+    return this.<Void> newAdminCaller()
+      .action((controller, stub) -> this.<EmptyMsg, EmptyMsg, Void> adminCall(controller, stub,
+        EmptyMsg.getDefaultInstance(),
+        (s, c, req, done) -> s.clearManagedKeyDataCache(controller, req, done), resp -> null))
+      .serverName(serverName).call();
+  }
 }

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

@@ -343,6 +343,7 @@ public enum OperationStatusCode {
 
   /** Parameter name for HBase instance root directory */
   public static final String HBASE_DIR = "hbase.rootdir";
+  public static final String HBASE_ORIGINAL_DIR = "hbase.originalRootdir";
 
   /** Parameter name for HBase client IPC pool type */
   public static final String HBASE_CLIENT_IPC_POOL_TYPE = "hbase.client.ipc.pool.type";
@@ -1193,6 +1194,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
    */
@@ -1282,6 +1288,14 @@ public enum OperationStatusCode {
   public static final String CRYPTO_KEYPROVIDER_PARAMETERS_KEY =
     "hbase.crypto.keyprovider.parameters";
 
+  /** Configuration key for the managed crypto key provider, a class name */
+  public static final String CRYPTO_MANAGED_KEYPROVIDER_CONF_KEY =
+    "hbase.crypto.managed.keyprovider";
+
+  /** Configuration key for the managed crypto key provider parameters */
+  public static final String CRYPTO_MANAGED_KEYPROVIDER_PARAMETERS_KEY =
+    "hbase.crypto.managed.keyprovider.parameters";
+
   /** Configuration key for the name of the master key for the cluster, a string */
   public static final String CRYPTO_MASTERKEY_NAME_CONF_KEY = "hbase.crypto.master.key.name";
 
@@ -1305,6 +1319,43 @@ 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;
+
+  /**
+   * 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;
+
+  /** Enables or disables local key generation per file. */
+  public static final String CRYPTO_MANAGED_KEYS_LOCAL_KEY_GEN_PER_FILE_ENABLED_CONF_KEY =
+    "hbase.crypto.managed_keys.local_key_gen_per_file.enabled";
+  public static final boolean CRYPTO_MANAGED_KEYS_LOCAL_KEY_GEN_PER_FILE_DEFAULT_ENABLED = false;
+
   /** 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

@@ -29,12 +29,14 @@
 import java.util.Arrays;
 import java.util.Map;
 import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.BiFunction;
 import javax.crypto.SecretKeyFactory;
 import javax.crypto.spec.PBEKeySpec;
 import javax.crypto.spec.SecretKeySpec;
 import org.apache.commons.io.IOUtils;
 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;
@@ -554,29 +556,50 @@ public static CipherProvider getCipherProvider(Configuration conf) {
     }
   }
 
-  static final Map<Pair<String, String>, KeyProvider> keyProviderCache = new ConcurrentHashMap<>();
+  static final Map<Pair<String, String>, Object> keyProviderCache = new ConcurrentHashMap<>();
 
-  public static KeyProvider getKeyProvider(Configuration conf) {
-    String providerClassName =
-      conf.get(HConstants.CRYPTO_KEYPROVIDER_CONF_KEY, KeyStoreKeyProvider.class.getName());
-    String providerParameters = conf.get(HConstants.CRYPTO_KEYPROVIDER_PARAMETERS_KEY, "");
-    try {
-      Pair<String, String> providerCacheKey = new Pair<>(providerClassName, providerParameters);
-      KeyProvider provider = keyProviderCache.get(providerCacheKey);
-      if (provider != null) {
-        return provider;
-      }
-      provider = (KeyProvider) ReflectionUtils
-        .newInstance(getClassLoaderForClass(KeyProvider.class).loadClass(providerClassName), conf);
-      provider.init(providerParameters);
-      if (LOG.isDebugEnabled()) {
-        LOG.debug("Installed " + providerClassName + " into key provider cache");
+  private static Object createProvider(final Configuration conf, String classNameKey,
+    String parametersKey, Class<?> defaultProviderClass, ClassLoader classLoaderForClass,
+    BiFunction<Object, String, Void> initFunction) {
+    String providerClassName = conf.get(classNameKey, defaultProviderClass.getName());
+    String providerParameters = conf.get(parametersKey, "");
+    Pair<String, String> providerCacheKey = new Pair<>(providerClassName, providerParameters);
+    Object provider = keyProviderCache.get(providerCacheKey);
+    if (provider == null) {
+      try {
+        provider =
+          ReflectionUtils.newInstance(classLoaderForClass.loadClass(providerClassName), conf);
+        initFunction.apply(provider, providerParameters);
+      } catch (Exception e) {
+        throw new RuntimeException(e);
       }
       keyProviderCache.put(providerCacheKey, provider);
-      return provider;
-    } catch (Exception e) {
-      throw new RuntimeException(e);
+      LOG.debug("Installed {} into key provider cache", providerClassName);
     }
+    return provider;
+  }
+
+  public static KeyProvider getKeyProvider(final Configuration conf) {
+    return (KeyProvider) createProvider(conf, HConstants.CRYPTO_KEYPROVIDER_CONF_KEY,
+      HConstants.CRYPTO_KEYPROVIDER_PARAMETERS_KEY, KeyStoreKeyProvider.class,
+      getClassLoaderForClass(KeyProvider.class), (provider, providerParameters) -> {
+        ((KeyProvider) provider).init(providerParameters);
+        return null;
+      });
+  }
+
+  public static ManagedKeyProvider getManagedKeyProvider(final Configuration conf) {
+    return (ManagedKeyProvider) createProvider(conf, HConstants.CRYPTO_MANAGED_KEYPROVIDER_CONF_KEY,
+      HConstants.CRYPTO_MANAGED_KEYPROVIDER_PARAMETERS_KEY, ManagedKeyProvider.class,
+      getClassLoaderForClass(ManagedKeyProvider.class), (provider, providerParameters) -> {
+        ((ManagedKeyProvider) provider).initConfig(conf, providerParameters);
+        return null;
+      });
+  }
+
+  @InterfaceAudience.LimitedPrivate(HBaseInterfaceAudience.UNITTEST)
+  public static void clearKeyProviderCache() {
+    keyProviderCache.clear();
   }
 
   public static void incrementIv(byte[] iv) {

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

@@ -78,6 +78,8 @@
 @InterfaceStability.Evolving
 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
@@ -174,9 +176,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);