HBASE-13002 Make encryption cipher configurable

Signed-off-by: Andrew Purtell <apurtell@apache.org>

Conflicts:
	hbase-common/src/main/java/org/apache/hadoop/hbase/HConstants.java
	hbase-common/src/test/java/org/apache/hadoop/hbase/io/crypto/TestEncryption.java

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

@@ -63,8 +63,7 @@ public class EncryptionUtil {
 
   /**
    * Protect a key by encrypting it with the secret key of the given subject.
-   * The configuration must be set up correctly for key alias resolution. Keys
+   * The configuration must be set up correctly for key alias resolution.
-   * are always wrapped using AES.
    * @param conf configuration
    * @param subject subject key alias
    * @param key the key
@@ -72,10 +71,12 @@ public class EncryptionUtil {
    */
   public static byte[] wrapKey(Configuration conf, String subject, Key key)
       throws IOException {
-    // Wrap the key with AES
+    // Wrap the key with the configured encryption algorithm.
-    Cipher cipher = Encryption.getCipher(conf, "AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    Cipher cipher = Encryption.getCipher(conf, algorithm);
     if (cipher == null) {
-      throw new RuntimeException("Cipher 'AES' not available");
+      throw new RuntimeException("Cipher '" + algorithm + "' not available");
     }
     EncryptionProtos.WrappedKey.Builder builder = EncryptionProtos.WrappedKey.newBuilder();
     builder.setAlgorithm(key.getAlgorithm());
@@ -100,8 +101,7 @@ public class EncryptionUtil {
 
   /**
    * 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. Keys
+   * The configuration must be set up correctly for key alias resolution.
-   * are always unwrapped using AES.
    * @param conf configuration
    * @param subject subject key alias
    * @param value the encrypted key bytes
@@ -113,9 +113,11 @@ public class EncryptionUtil {
       throws IOException, KeyException {
     EncryptionProtos.WrappedKey wrappedKey = EncryptionProtos.WrappedKey.PARSER
         .parseDelimitedFrom(new ByteArrayInputStream(value));
-    Cipher cipher = Encryption.getCipher(conf, "AES");
+    String algorithm = conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY,
+      HConstants.CIPHER_AES);
+    Cipher cipher = Encryption.getCipher(conf, algorithm);
     if (cipher == null) {
-      throw new RuntimeException("Algorithm 'AES' not available");
+      throw new RuntimeException("Cipher '" + algorithm + "' not available");
     }
     ByteArrayOutputStream out = new ByteArrayOutputStream();
     byte[] iv = wrappedKey.hasIv() ? wrappedKey.getIv().toByteArray() : null;

hbase-client/src/test/java/org/apache/hadoop/hbase/security/TestEncryptionUtil.java

@@ -47,7 +47,9 @@ public class TestEncryptionUtil {
     // generate a test key
     byte[] keyBytes = new byte[AES.KEY_LENGTH];
     new SecureRandom().nextBytes(keyBytes);
-    Key key = new SecretKeySpec(keyBytes, "AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    Key key = new SecretKeySpec(keyBytes, algorithm);
 
     // wrap the test key
     byte[] wrappedKeyBytes = EncryptionUtil.wrapKey(conf, "hbase", key);

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

@@ -931,7 +931,7 @@ public final class HConstants {
    *  NONE: no preference in destination of replicas
    *  ONE_SSD: place only one replica in SSD and the remaining in default storage
    *  and ALL_SSD: place all replica on SSD
-   *  
+   *
    * See http://hadoop.apache.org/docs/r2.6.0/hadoop-project-dist/hadoop-hdfs/ArchivalStorage.html*/
   public static final String WAL_STORAGE_POLICY = "hbase.wal.storage.policy";
   public static final String DEFAULT_WAL_STORAGE_POLICY = "NONE";
@@ -1043,6 +1043,9 @@ public final class HConstants {
 
   public static final long NO_NONCE = 0;
 
+  /** Default cipher for encryption */
+  public static final String CIPHER_AES = "AES";
+
   /** Configuration key for the crypto algorithm provider, a class name */
   public static final String CRYPTO_CIPHERPROVIDER_CONF_KEY = "hbase.crypto.cipherprovider";
 
@@ -1066,6 +1069,13 @@ public final class HConstants {
   /** Configuration key for the name of the master WAL encryption key for the cluster, a string */
   public static final String CRYPTO_WAL_KEY_NAME_CONF_KEY = "hbase.crypto.wal.key.name";
 
+  /** Configuration key for the algorithm used for creating jks key, a string */
+  public static final String CRYPTO_KEY_ALGORITHM_CONF_KEY = "hbase.crypto.key.algorithm";
+
+  /** Configuration key for the name of the alternate cipher algorithm for the cluster, a string */
+  public static final String CRYPTO_ALTERNATE_KEY_ALGORITHM_CONF_KEY =
+      "hbase.crypto.alternate.key.algorithm";
+
   /** Configuration key for enabling WAL encryption, a boolean */
   public static final String ENABLE_WAL_ENCRYPTION = "hbase.regionserver.wal.encryption";
 

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

@@ -469,9 +469,8 @@ public final class Encryption {
    * @param iv the initialization vector, can be null
    * @throws IOException
    */
-  public static void decryptWithSubjectKey(OutputStream out, InputStream in,
+  public static void decryptWithSubjectKey(OutputStream out, InputStream in, int outLen,
-      int outLen, String subject, Configuration conf, Cipher cipher,
+      String subject, Configuration conf, Cipher cipher, byte[] iv) throws IOException {
-      byte[] iv) throws IOException {
     Key key = getSecretKeyForSubject(subject, conf);
     if (key == null) {
       throw new IOException("No key found for subject '" + subject + "'");
@@ -479,7 +478,31 @@ public final class Encryption {
     Decryptor d = cipher.getDecryptor();
     d.setKey(key);
     d.setIv(iv); // can be null
-    decrypt(out, in, outLen, d);
+    try {
+      decrypt(out, in, outLen, d);
+    } 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)
+              + "'. 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);
+      }
+    }
   }
 
   private static ClassLoader getClassLoaderForClass(Class<?> c) {

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

@@ -140,11 +140,13 @@ public class TestCipherProvider {
     Configuration conf = HBaseConfiguration.create();
     CipherProvider provider = Encryption.getCipherProvider(conf);
     assertTrue(provider instanceof DefaultCipherProvider);
-    assertTrue(Arrays.asList(provider.getSupportedCiphers()).contains("AES"));
+    String algorithm =
-    Cipher a = Encryption.getCipher(conf, "AES");
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    assertTrue(Arrays.asList(provider.getSupportedCiphers()).contains(algorithm));
+    Cipher a = Encryption.getCipher(conf, algorithm);
     assertNotNull(a);
     assertTrue(a.getProvider() instanceof DefaultCipherProvider);
-    assertEquals(a.getName(), "AES");
+    assertEquals(a.getName(), algorithm);
     assertEquals(a.getKeyLength(), AES.KEY_LENGTH);
   }
 

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

@@ -28,6 +28,7 @@ import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.hbase.HBaseConfiguration;
+import org.apache.hadoop.hbase.HConstants;
 import org.apache.hadoop.hbase.testclassification.SmallTests;
 import org.apache.hadoop.hbase.util.Bytes;
 import org.junit.Test;
@@ -87,8 +88,10 @@ public class TestEncryption {
     LOG.info("checkTransformSymmetry: AES, plaintext length = " + plaintext.length);
 
     Configuration conf = HBaseConfiguration.create();
-    Cipher aes = Encryption.getCipher(conf, "AES");
+    String algorithm =
-    Key key = new SecretKeySpec(keyBytes, "AES");
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    Cipher aes = Encryption.getCipher(conf, algorithm);
+    Key key = new SecretKeySpec(keyBytes, algorithm);
 
     Encryptor e = aes.getEncryptor();
     e.setKey(key);

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/HStore.java

@@ -318,7 +318,7 @@ public class HStore implements Store {
         // Use the algorithm the key wants
         cipher = Encryption.getCipher(conf, key.getAlgorithm());
         if (cipher == null) {
-          throw new RuntimeException("Cipher '" + cipher + "' is not available");
+          throw new RuntimeException("Cipher '" + key.getAlgorithm() + "' is not available");
         }
         // Fail if misconfigured
         // We use the encryption type specified in the column schema as a sanity check on
@@ -332,7 +332,7 @@ public class HStore implements Store {
         // Family does not provide key material, create a random key
         cipher = Encryption.getCipher(conf, cipherName);
         if (cipher == null) {
-          throw new RuntimeException("Cipher '" + cipher + "' is not available");
+          throw new RuntimeException("Cipher '" + cipherName + "' is not available");
         }
         key = cipher.getRandomKey();
       }

hbase-server/src/main/java/org/apache/hadoop/hbase/regionserver/wal/SecureProtobufLogWriter.java

@@ -43,8 +43,6 @@ import org.apache.hadoop.hbase.security.User;
 public class SecureProtobufLogWriter extends ProtobufLogWriter {
 
   private static final Log LOG = LogFactory.getLog(SecureProtobufLogWriter.class);
-  private static final String DEFAULT_CIPHER = "AES";
-
   private Encryptor encryptor = null;
 
   @Override
@@ -56,7 +54,8 @@ public class SecureProtobufLogWriter extends ProtobufLogWriter {
       EncryptionTest.testCipherProvider(conf);
 
       // Get an instance of our cipher
-      final String cipherName = conf.get(HConstants.CRYPTO_WAL_ALGORITHM_CONF_KEY, DEFAULT_CIPHER);
+      final String cipherName =
+          conf.get(HConstants.CRYPTO_WAL_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
       Cipher cipher = Encryption.getCipher(conf, cipherName);
       if (cipher == null) {
         throw new RuntimeException("Cipher '" + cipherName + "' is not available");

hbase-server/src/test/java/org/apache/hadoop/hbase/io/hfile/TestHFileEncryption.java

@@ -69,7 +69,9 @@ public class TestHFileEncryption {
     fs = FileSystem.get(conf);
 
     cryptoContext = Encryption.newContext(conf);
-    Cipher aes = Encryption.getCipher(conf, "AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    Cipher aes = Encryption.getCipher(conf, algorithm);
     assertNotNull(aes);
     cryptoContext.setCipher(aes);
     byte[] key = new byte[aes.getKeyLength()];

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestEncryptionKeyRotation.java

@@ -66,9 +66,11 @@ public class TestEncryptionKeyRotation {
     SecureRandom rng = new SecureRandom();
     byte[] keyBytes = new byte[AES.KEY_LENGTH];
     rng.nextBytes(keyBytes);
-    initialCFKey = new SecretKeySpec(keyBytes, "AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    initialCFKey = new SecretKeySpec(keyBytes, algorithm);
     rng.nextBytes(keyBytes);
-    secondCFKey = new SecretKeySpec(keyBytes, "AES");
+    secondCFKey = new SecretKeySpec(keyBytes, algorithm);
   }
 
   @BeforeClass
@@ -94,7 +96,9 @@ public class TestEncryptionKeyRotation {
     HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("default",
       "testCFKeyRotation"));
     HColumnDescriptor hcd = new HColumnDescriptor("cf");
-    hcd.setEncryptionType("AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    hcd.setEncryptionType(algorithm);
     hcd.setEncryptionKey(EncryptionUtil.wrapKey(conf, "hbase", initialCFKey));
     htd.addFamily(hcd);
 
@@ -153,7 +157,9 @@ public class TestEncryptionKeyRotation {
     HTableDescriptor htd = new HTableDescriptor(TableName.valueOf("default",
       "testMasterKeyRotation"));
     HColumnDescriptor hcd = new HColumnDescriptor("cf");
-    hcd.setEncryptionType("AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    hcd.setEncryptionType(algorithm);
     hcd.setEncryptionKey(EncryptionUtil.wrapKey(conf, "hbase", initialCFKey));
     htd.addFamily(hcd);
 

hbase-server/src/test/java/org/apache/hadoop/hbase/regionserver/TestEncryptionRandomKeying.java

@@ -91,7 +91,9 @@ public class TestEncryptionRandomKeying {
     // Specify an encryption algorithm without a key
     htd = new HTableDescriptor(TableName.valueOf("default", "TestEncryptionRandomKeying"));
     HColumnDescriptor hcd = new HColumnDescriptor("cf");
-    hcd.setEncryptionType("AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    hcd.setEncryptionType(algorithm);
     htd.addFamily(hcd);
 
     // Start the minicluster

hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestEncryptionTest.java

@@ -74,10 +74,12 @@ public class TestEncryptionTest {
   public void testTestCipher() {
     Configuration conf = HBaseConfiguration.create();
     conf.set(HConstants.CRYPTO_KEYPROVIDER_CONF_KEY, KeyProviderForTesting.class.getName());
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
     try {
-      EncryptionTest.testEncryption(conf, "AES", null);
+      EncryptionTest.testEncryption(conf, algorithm, null);
     } catch (Exception e) {
-      fail("Test for cipher AES should have succeeded");
+      fail("Test for cipher " + algorithm + " should have succeeded");
     }
     try {
       EncryptionTest.testEncryption(conf, "foobar", null);

hbase-server/src/test/java/org/apache/hadoop/hbase/util/TestHBaseFsckEncryption.java

@@ -77,7 +77,9 @@ public class TestHBaseFsckEncryption {
     SecureRandom rng = new SecureRandom();
     byte[] keyBytes = new byte[AES.KEY_LENGTH];
     rng.nextBytes(keyBytes);
-    cfKey = new SecretKeySpec(keyBytes, "AES");
+    String algorithm =
+        conf.get(HConstants.CRYPTO_KEY_ALGORITHM_CONF_KEY, HConstants.CIPHER_AES);
+    cfKey = new SecretKeySpec(keyBytes,algorithm);
 
     // Start the minicluster
     TEST_UTIL.startMiniCluster(3);
@@ -85,7 +87,7 @@ public class TestHBaseFsckEncryption {
     // Create the table
     htd = new HTableDescriptor(TableName.valueOf("default", "TestHBaseFsckEncryption"));
     HColumnDescriptor hcd = new HColumnDescriptor("cf");
-    hcd.setEncryptionType("AES");
+    hcd.setEncryptionType(algorithm);
     hcd.setEncryptionKey(EncryptionUtil.wrapKey(conf,
       conf.get(HConstants.CRYPTO_MASTERKEY_NAME_CONF_KEY, User.getCurrent().getShortName()),
       cfKey));