HBASE-29495: Integrate key management with existing encryption

.gitignore

@@ -25,6 +25,7 @@ linklint/
 **/*.log
 tmp
 **/.flattened-pom.xml
+.sw*
 .*.sw*
 ID
 filenametags

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

@@ -27,7 +27,6 @@
 import org.apache.commons.crypto.cipher.CryptoCipherFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HConstants;
-import org.apache.hadoop.hbase.client.ColumnFamilyDescriptor;
 import org.apache.hadoop.hbase.io.crypto.Cipher;
 import org.apache.hadoop.hbase.io.crypto.Encryption;
 import org.apache.hadoop.hbase.io.crypto.aes.CryptoAES;
@@ -219,57 +218,6 @@ public static Key unwrapWALKey(Configuration conf, String subject, byte[] value)
     return getUnwrapKey(conf, subject, wrappedKey, cipher, null);
   }
 
-  /**
-   * Helper to create an encyption context.
-   * @param conf   The current configuration.
-   * @param family The current column descriptor.
-   * @return The created encryption context.
-   * @throws IOException           if an encryption key for the column cannot be unwrapped
-   * @throws IllegalStateException in case of encryption related configuration errors
-   */
-  public static Encryption.Context createEncryptionContext(Configuration conf,
-    ColumnFamilyDescriptor family) throws IOException {
-    Encryption.Context cryptoContext = Encryption.Context.NONE;
-    String cipherName = family.getEncryptionType();
-    if (cipherName != null) {
-      if (!Encryption.isEncryptionEnabled(conf)) {
-        throw new IllegalStateException("Encryption for family '" + family.getNameAsString()
-          + "' configured with type '" + cipherName + "' but the encryption feature is disabled");
-      }
-      Cipher cipher;
-      Key key;
-      byte[] keyBytes = family.getEncryptionKey();
-      if (keyBytes != null) {
-        // Family provides specific key material
-        key = unwrapKey(conf, keyBytes);
-        // Use the algorithm the key wants
-        cipher = Encryption.getCipher(conf, key.getAlgorithm());
-        if (cipher == null) {
-          throw new IllegalStateException("Cipher '" + key.getAlgorithm() + "' is not available");
-        }
-        // Fail if misconfigured
-        // We use the encryption type specified in the column schema as a sanity check on
-        // what the wrapped key is telling us
-        if (!cipher.getName().equalsIgnoreCase(cipherName)) {
-          throw new IllegalStateException(
-            "Encryption for family '" + family.getNameAsString() + "' configured with type '"
-              + cipherName + "' but key specifies algorithm '" + cipher.getName() + "'");
-        }
-      } else {
-        // Family does not provide key material, create a random key
-        cipher = Encryption.getCipher(conf, cipherName);
-        if (cipher == null) {
-          throw new IllegalStateException("Cipher '" + cipherName + "' is not available");
-        }
-        key = cipher.getRandomKey();
-      }
-      cryptoContext = Encryption.newContext(conf);
-      cryptoContext.setCipher(cipher);
-      cryptoContext.setKey(key);
-    }
-    return cryptoContext;
-  }
-
   /**
    * Helper for {@link #unwrapKey(Configuration, String, byte[])} which automatically uses the
    * configured master and alternative keys, rather than having to specify a key type to unwrap

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";
@@ -1342,6 +1343,11 @@ public enum OperationStatusCode {
     "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/Context.java

@@ -34,6 +34,8 @@ public class Context implements Configurable {
   private Configuration conf;
   private Cipher cipher;
   private Key key;
+  private ManagedKeyData kekData;
+  private String keyNamespace;
   private String keyHash;
 
   Context(Configuration conf) {
@@ -97,4 +99,22 @@ public Context setKey(Key key) {
     this.keyHash = new String(Hex.encodeHex(Encryption.computeCryptoKeyHash(conf, encoded)));
     return this;
   }
+
+  public Context setKeyNamespace(String keyNamespace) {
+    this.keyNamespace = keyNamespace;
+    return this;
+  }
+
+  public String getKeyNamespace() {
+    return keyNamespace;
+  }
+
+  public Context setKEKData(ManagedKeyData kekData) {
+    this.kekData = kekData;
+    return this;
+  }
+
+  public ManagedKeyData getKEKData() {
+    return kekData;
+  }
 }

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

@@ -117,12 +117,6 @@ public Context setCipher(Cipher cipher) {
       return this;
     }
 
-    @Override
-    public Context setKey(Key key) {
-      super.setKey(key);
-      return this;
-    }
-
     public Context setKey(byte[] key) {
       super.setKey(new SecretKeySpec(key, getCipher().getName()));
       return this;

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

@@ -54,11 +54,17 @@ public class ManagedKeyData {
    */
   public static final String KEY_SPACE_GLOBAL = "*";
 
+  /**
+   * Special value to be used for custodian to indicate that it is global, meaning it is not
+   * associated with a specific custodian.
+   */
+  public static final byte[] KEY_GLOBAL_CUSTODIAN_BYTES = KEY_SPACE_GLOBAL.getBytes();
+
   /**
    * Encoded form of global custodian.
    */
   public static final String KEY_GLOBAL_CUSTODIAN =
-    ManagedKeyProvider.encodeToStr(KEY_SPACE_GLOBAL.getBytes());
+    ManagedKeyProvider.encodeToStr(KEY_GLOBAL_CUSTODIAN_BYTES);
 
   private final byte[] keyCustodian;
   private final String keyNamespace;

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

@@ -18,6 +18,8 @@
 package org.apache.hadoop.hbase.io.crypto;
 
 import java.security.Key;
+import java.util.HashMap;
+import java.util.Map;
 import javax.crypto.spec.SecretKeySpec;
 import org.apache.yetus.audience.InterfaceAudience;
 
@@ -27,8 +29,13 @@
 @InterfaceAudience.Private
 public class MockAesKeyProvider implements KeyProvider {
 
+  private Map<String, Key> keys = new HashMap<>();
+
+  private boolean cacheKeys = false;
+
   @Override
   public void init(String parameters) {
+    cacheKeys = Boolean.parseBoolean(parameters);
   }
 
   @Override
@@ -40,7 +47,14 @@ public Key getKey(String name) {
   public Key[] getKeys(String[] aliases) {
     Key[] result = new Key[aliases.length];
     for (int i = 0; i < aliases.length; i++) {
-      result[i] = new SecretKeySpec(Encryption.hash128(aliases[i]), "AES");
+      if (keys.containsKey(aliases[i])) {
+        result[i] = keys.get(aliases[i]);
+      } else {
+        result[i] = new SecretKeySpec(Encryption.hash128(aliases[i]), "AES");
+        if (cacheKeys) {
+          keys.put(aliases[i], result[i]);
+        }
+      }
     }
     return result;
   }

hbase-common/src/main/java/org/apache/hadoop/hbase/util/CommonFSUtils.java

@@ -288,17 +288,48 @@ public static String getPath(Path p) {
    * @throws IOException e
    */
   public static Path getRootDir(final Configuration c) throws IOException {
-    Path p = new Path(c.get(HConstants.HBASE_DIR));
+    return getRootDir(c, HConstants.HBASE_DIR);
+  }
+
+  /**
+   * Get the path for the original root data directory, which could be different from the current
+   * root directory, in case it was changed.
+   * @param c configuration
+   * @return {@link Path} to hbase original root directory from configuration as a qualified Path.
+   * @throws IOException e
+   */
+  public static Path getOriginalRootDir(final Configuration c) throws IOException {
+    return getRootDir(c,
+      c.get(HConstants.HBASE_ORIGINAL_DIR) == null
+        ? HConstants.HBASE_DIR
+        : HConstants.HBASE_ORIGINAL_DIR);
+  }
+
+  /**
+   * Get the path for the root data directory
+   * @param c           configuration
+   * @param rootDirProp the property name for the root directory
+   * @return {@link Path} to hbase root directory from configuration as a qualified Path.
+   * @throws IOException e
+   */
+  public static Path getRootDir(final Configuration c, final String rootDirProp)
+    throws IOException {
+    Path p = new Path(c.get(rootDirProp));
     FileSystem fs = p.getFileSystem(c);
     return p.makeQualified(fs.getUri(), fs.getWorkingDirectory());
   }
 
   public static void setRootDir(final Configuration c, final Path root) {
+    // Keep track of the original root dir.
+    if (c.get(HConstants.HBASE_ORIGINAL_DIR) == null && c.get(HConstants.HBASE_DIR) != null) {
+      c.set(HConstants.HBASE_ORIGINAL_DIR, c.get(HConstants.HBASE_DIR));
+    }
     c.set(HConstants.HBASE_DIR, root.toString());
   }
 
   public static Path getSystemKeyDir(final Configuration c) throws IOException {
-    return new Path(getRootDir(c), HConstants.SYSTEM_KEYS_DIRECTORY);
+    // Always use the original root dir for system key dir, in case it was changed..
+    return new Path(getOriginalRootDir(c), HConstants.SYSTEM_KEYS_DIRECTORY);
   }
 
   public static void setFsDefault(final Configuration c, final Path root) {