HBASE-29617: Changes to support smooth migration to key management

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

@@ -114,6 +114,9 @@ public interface ColumnFamilyDescriptor {
   /** Returns Return the raw crypto key attribute for the family, or null if not set */
   byte[] getEncryptionKey();
 
+  /** Returns the encryption key namespace for this family */
+  String getEncryptionKeyNamespace();
+
   /** Returns Return the encryption algorithm in use by this family */
   String getEncryptionType();
 

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

@@ -167,6 +167,10 @@ public class ColumnFamilyDescriptorBuilder {
   @InterfaceAudience.Private
   public static final String ENCRYPTION_KEY = "ENCRYPTION_KEY";
   private static final Bytes ENCRYPTION_KEY_BYTES = new Bytes(Bytes.toBytes(ENCRYPTION_KEY));
+  @InterfaceAudience.Private
+  public static final String ENCRYPTION_KEY_NAMESPACE = "ENCRYPTION_KEY_NAMESPACE";
+  private static final Bytes ENCRYPTION_KEY_NAMESPACE_BYTES =
+    new Bytes(Bytes.toBytes(ENCRYPTION_KEY_NAMESPACE));
 
   private static final boolean DEFAULT_MOB = false;
   @InterfaceAudience.Private
@@ -320,6 +324,7 @@ public static Map<String, String> getDefaultValues() {
     DEFAULT_VALUES.keySet().forEach(s -> RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(s))));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(ENCRYPTION)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(ENCRYPTION_KEY)));
+    RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(ENCRYPTION_KEY_NAMESPACE)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(IS_MOB)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(MOB_THRESHOLD)));
     RESERVED_KEYWORDS.add(new Bytes(Bytes.toBytes(MOB_COMPACT_PARTITION_POLICY)));
@@ -522,6 +527,11 @@ public ColumnFamilyDescriptorBuilder setEncryptionKey(final byte[] value) {
     return this;
   }
 
+  public ColumnFamilyDescriptorBuilder setEncryptionKeyNamespace(final String value) {
+    desc.setEncryptionKeyNamespace(value);
+    return this;
+  }
+
   public ColumnFamilyDescriptorBuilder setEncryptionType(String value) {
     desc.setEncryptionType(value);
     return this;
@@ -1337,6 +1347,20 @@ public ModifyableColumnFamilyDescriptor setEncryptionKey(byte[] keyBytes) {
       return setValue(ENCRYPTION_KEY_BYTES, new Bytes(keyBytes));
     }
 
+    @Override
+    public String getEncryptionKeyNamespace() {
+      return getStringOrDefault(ENCRYPTION_KEY_NAMESPACE_BYTES, Function.identity(), null);
+    }
+
+    /**
+     * Set the encryption key namespace attribute for the family
+     * @param keyNamespace the key namespace, or null to remove existing setting
+     * @return this (for chained invocation)
+     */
+    public ModifyableColumnFamilyDescriptor setEncryptionKeyNamespace(String keyNamespace) {
+      return setValue(ENCRYPTION_KEY_NAMESPACE_BYTES, keyNamespace);
+    }
+
     @Override
     public long getMobThreshold() {
       return getStringOrDefault(MOB_THRESHOLD_BYTES, Long::valueOf, DEFAULT_MOB_THRESHOLD);

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

@@ -1288,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";
 

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

@@ -29,6 +29,7 @@
 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;
@@ -556,32 +557,45 @@ 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 (provider instanceof ManagedKeyProvider) {
-        ((ManagedKeyProvider) provider).initConfig(conf);
-      }
-      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 " + providerClassName + " into key provider cache");
     }
+    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)

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

@@ -25,16 +25,16 @@
 
 /**
  * Interface for key providers of managed keys. Defines methods for generating and managing managed
- * keys, as well as handling key storage and retrieval. The interface extends the basic
- * {@link KeyProvider} interface with additional methods for working with managed keys.
+ * keys, as well as handling key storage and retrieval.
  */
 @InterfaceAudience.Public
-public interface ManagedKeyProvider extends KeyProvider {
+public interface ManagedKeyProvider {
   /**
    * Initialize the provider with the given configuration.
-   * @param conf Hadoop configuration
+   * @param conf               Hadoop configuration
+   * @param providerParameters provider parameters
    */
-  void initConfig(Configuration conf);
+  void initConfig(Configuration conf, String providerParameters);
 
   /**
    * Retrieve the system key using the given system identifier.

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

@@ -31,6 +31,7 @@
 public class ManagedKeyStoreKeyProvider extends KeyStoreKeyProvider implements ManagedKeyProvider {
   public static final String KEY_METADATA_ALIAS = "KeyAlias";
   public static final String KEY_METADATA_CUST = "KeyCustodian";
+  public static final String KEY_METADATA_NAMESPACE = "KeyNamespace";
 
   private static final java.lang.reflect.Type KEY_METADATA_TYPE =
     new TypeToken<HashMap<String, String>>() {
@@ -39,8 +40,11 @@ public class ManagedKeyStoreKeyProvider extends KeyStoreKeyProvider implements M
   private Configuration conf;
 
   @Override
-  public void initConfig(Configuration conf) {
+  public void initConfig(Configuration conf, String providerParameters) {
     this.conf = conf;
+    if (providerParameters != null) {
+      super.init(providerParameters);
+    }
   }
 
   @Override
@@ -56,8 +60,8 @@ public ManagedKeyData getSystemKey(byte[] clusterId) {
       throw new RuntimeException("Unable to find system key with alias: " + systemKeyAlias);
     }
     // Encode clusterId too for consistency with that of key custodian.
-    String keyMetadata =
-      generateKeyMetadata(systemKeyAlias, ManagedKeyProvider.encodeToStr(clusterId));
+    String keyMetadata = generateKeyMetadata(systemKeyAlias,
+      ManagedKeyProvider.encodeToStr(clusterId), ManagedKeyData.KEY_SPACE_GLOBAL);
     return new ManagedKeyData(clusterId, ManagedKeyData.KEY_SPACE_GLOBAL, key,
       ManagedKeyState.ACTIVE, keyMetadata);
   }
@@ -66,9 +70,25 @@ public ManagedKeyData getSystemKey(byte[] clusterId) {
   public ManagedKeyData getManagedKey(byte[] key_cust, String key_namespace) throws IOException {
     checkConfig();
     String encodedCust = ManagedKeyProvider.encodeToStr(key_cust);
-    String aliasConfKey =
-      HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encodedCust + "." + "alias";
-    String keyMetadata = generateKeyMetadata(conf.get(aliasConfKey, null), encodedCust);
+
+    // Handle null key_namespace by defaulting to global namespace
+    if (key_namespace == null) {
+      key_namespace = ManagedKeyData.KEY_SPACE_GLOBAL;
+    }
+
+    // Get alias configuration for the specific custodian+namespace combination
+    String aliasConfKey = buildAliasConfKey(encodedCust, key_namespace);
+    String alias = conf.get(aliasConfKey, null);
+
+    // Generate metadata with actual alias (used for both success and failure cases)
+    String keyMetadata = generateKeyMetadata(alias, encodedCust, key_namespace);
+
+    // If no alias is configured for this custodian+namespace combination, treat as key not found
+    if (alias == null) {
+      return new ManagedKeyData(key_cust, key_namespace, null, ManagedKeyState.FAILED, keyMetadata);
+    }
+
+    // Namespaces match, proceed to get the key
     return unwrapKey(keyMetadata, null);
   }
 
@@ -77,17 +97,21 @@ public ManagedKeyData unwrapKey(String keyMetadataStr, byte[] wrappedKey) throws
     Map<String, String> keyMetadata =
       GsonUtil.getDefaultInstance().fromJson(keyMetadataStr, KEY_METADATA_TYPE);
     String encodedCust = keyMetadata.get(KEY_METADATA_CUST);
-    String activeStatusConfKey =
-      HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encodedCust + ".active";
+    String namespace = keyMetadata.get(KEY_METADATA_NAMESPACE);
+    if (namespace == null) {
+      // For backwards compatibility, default to global namespace
+      namespace = ManagedKeyData.KEY_SPACE_GLOBAL;
+    }
+    String activeStatusConfKey = buildActiveStatusConfKey(encodedCust, namespace);
     boolean isActive = conf.getBoolean(activeStatusConfKey, true);
     byte[] key_cust = ManagedKeyProvider.decodeToBytes(encodedCust);
     String alias = keyMetadata.get(KEY_METADATA_ALIAS);
     Key key = alias != null ? getKey(alias) : null;
     if (key != null) {
-      return new ManagedKeyData(key_cust, ManagedKeyData.KEY_SPACE_GLOBAL, key,
+      return new ManagedKeyData(key_cust, namespace, key,
         isActive ? ManagedKeyState.ACTIVE : ManagedKeyState.INACTIVE, keyMetadataStr);
     }
-    return new ManagedKeyData(key_cust, ManagedKeyData.KEY_SPACE_GLOBAL, null,
+    return new ManagedKeyData(key_cust, namespace, null,
       isActive ? ManagedKeyState.FAILED : ManagedKeyState.DISABLED, keyMetadataStr);
   }
 
@@ -98,9 +122,24 @@ private void checkConfig() {
   }
 
   public static String generateKeyMetadata(String aliasName, String encodedCust) {
-    Map<String, String> metadata = new HashMap<>(2);
+    return generateKeyMetadata(aliasName, encodedCust, ManagedKeyData.KEY_SPACE_GLOBAL);
+  }
+
+  public static String generateKeyMetadata(String aliasName, String encodedCust, String namespace) {
+    Map<String, String> metadata = new HashMap<>(3);
     metadata.put(KEY_METADATA_ALIAS, aliasName);
     metadata.put(KEY_METADATA_CUST, encodedCust);
+    metadata.put(KEY_METADATA_NAMESPACE, namespace);
     return GsonUtil.getDefaultInstance().toJson(metadata, HashMap.class);
   }
+
+  private String buildAliasConfKey(String encodedCust, String namespace) {
+    return HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encodedCust + "." + namespace
+      + ".alias";
+  }
+
+  private String buildActiveStatusConfKey(String encodedCust, String namespace) {
+    return HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encodedCust + "." + namespace
+      + ".active";
+  }
 }

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

@@ -50,12 +50,20 @@ public Key[] getKeys(String[] aliases) {
       if (keys.containsKey(aliases[i])) {
         result[i] = keys.get(aliases[i]);
       } else {
-        result[i] = new SecretKeySpec(Encryption.hash128(aliases[i]), "AES");
+        // When not caching keys, we want to make the key generation deterministic.
+        result[i] = new SecretKeySpec(
+          Encryption.hash128(
+            cacheKeys ? aliases[i] + "-" + String.valueOf(System.currentTimeMillis()) : aliases[i]),
+          "AES");
         if (cacheKeys) {
           keys.put(aliases[i], result[i]);
         }
       }
     }
     return result;
   }
+
+  public void clearKeys() {
+    keys.clear();
+  }
 }

hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/KeymetaTestUtils.java

@@ -116,6 +116,14 @@ private KeymetaTestUtils() {
   public static void addEntry(Configuration conf, int keyLen, KeyStore store, String alias,
     String custodian, boolean withPasswordOnAlias, Map<Bytes, Bytes> cust2key,
     Map<Bytes, String> cust2alias, Properties passwordFileProps) throws Exception {
+    addEntry(conf, keyLen, store, alias, custodian, withPasswordOnAlias, cust2key, cust2alias,
+      passwordFileProps, ManagedKeyData.KEY_SPACE_GLOBAL);
+  }
+
+  public static void addEntry(Configuration conf, int keyLen, KeyStore store, String alias,
+    String custodian, boolean withPasswordOnAlias, Map<Bytes, Bytes> cust2key,
+    Map<Bytes, String> cust2alias, Properties passwordFileProps, String namespace)
+    throws Exception {
     Preconditions.checkArgument(keyLen == 256 || keyLen == 128, "Key length must be 256 or 128");
     byte[] key =
       MessageDigest.getInstance(keyLen == 256 ? "SHA-256" : "MD5").digest(Bytes.toBytes(alias));
@@ -124,8 +132,18 @@ public static void addEntry(Configuration conf, int keyLen, KeyStore store, Stri
     store.setEntry(alias, new KeyStore.SecretKeyEntry(new SecretKeySpec(key, "AES")),
       new KeyStore.PasswordProtection(withPasswordOnAlias ? PASSWORD.toCharArray() : new char[0]));
     String encCust = Base64.getEncoder().encodeToString(custodian.getBytes());
-    String confKey = HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encCust + "." + "alias";
-    conf.set(confKey, alias);
+
+    // Use new format: PREFIX.{encodedCust}.{namespace}.alias
+    // For global namespace use "*", for custom namespace use actual namespace name
+    String namespaceKey = ManagedKeyData.KEY_SPACE_GLOBAL.equals(namespace) ? "*" : namespace;
+    String aliasConfKey =
+      HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encCust + "." + namespaceKey + ".alias";
+    String activeStatusConfKey = HConstants.CRYPTO_MANAGED_KEY_STORE_CONF_KEY_PREFIX + encCust + "."
+      + namespaceKey + ".active";
+
+    conf.set(aliasConfKey, alias);
+    conf.setBoolean(activeStatusConfKey, true);
+
     if (passwordFileProps != null) {
       passwordFileProps.setProperty(alias, PASSWORD);
     }

hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/MockManagedKeyProvider.java

@@ -45,8 +45,8 @@ public class MockManagedKeyProvider extends MockAesKeyProvider implements Manage
   private String systemKeyAlias = "default_system_key_alias";
 
   @Override
-  public void initConfig(Configuration conf) {
-    // NO-OP
+  public void initConfig(Configuration conf, String providerParameters) {
+    super.init(providerParameters);
   }
 
   @Override

hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/TestKeyProvider.java

@@ -55,4 +55,18 @@ public void testTestProvider() {
       key.getEncoded().length);
   }
 
+  @Test
+  public void testManagedKeyProvider() {
+    Configuration conf = HBaseConfiguration.create();
+    conf.set(HConstants.CRYPTO_MANAGED_KEYPROVIDER_CONF_KEY,
+      MockManagedKeyProvider.class.getName());
+    ManagedKeyProvider provider = Encryption.getManagedKeyProvider(conf);
+    assertNotNull("Null returned for managed provider", provider);
+    assertTrue("Provider is not the expected type", provider instanceof MockManagedKeyProvider);
+
+    // Test that it's cached
+    ManagedKeyProvider provider2 = Encryption.getManagedKeyProvider(conf);
+    assertTrue("Provider should be cached and same instance", provider == provider2);
+  }
+
 }