HADOOP-13827. Add reencryptEncryptedKey interface to KMS.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProvider.java

@@ -33,6 +33,8 @@ import java.util.Map;
 
 import com.google.gson.stream.JsonReader;
 import com.google.gson.stream.JsonWriter;
+import org.apache.commons.lang.builder.EqualsBuilder;
+import org.apache.commons.lang.builder.HashCodeBuilder;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
@@ -86,6 +88,7 @@ public abstract class KeyProvider {
       return material;
     }
 
+    @Override
     public String toString() {
       StringBuilder buf = new StringBuilder();
       buf.append("key(");
@@ -105,6 +108,31 @@ public abstract class KeyProvider {
       }
       return buf.toString();
     }
+
+    @Override
+    public boolean equals(Object rhs) {
+      if (this == rhs) {
+        return true;
+      }
+      if (rhs == null || getClass() != rhs.getClass()) {
+        return false;
+      }
+      final KeyVersion kv = (KeyVersion) rhs;
+      return new EqualsBuilder().
+          append(name, kv.name).
+          append(versionName, kv.versionName).
+          append(material, kv.material).
+          isEquals();
+    }
+
+    @Override
+    public int hashCode() {
+      return new HashCodeBuilder().
+          append(name).
+          append(versionName).
+          append(material).
+          toHashCode();
+    }
   }
 
   /**

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/KeyProviderCryptoExtension.java

@@ -188,8 +188,8 @@ public class KeyProviderCryptoExtension extends
     public void drain(String keyName);
 
     /**
-     * Generates a key material and encrypts it using the given key version name
+     * Generates a key material and encrypts it using the given key name.
-     * and initialization vector. The generated key material is of the same
+     * The generated key material is of the same
      * length as the <code>KeyVersion</code> material of the latest key version
      * of the key and is encrypted using the same cipher.
      * <p/>
@@ -210,7 +210,7 @@ public class KeyProviderCryptoExtension extends
         GeneralSecurityException;
 
     /**
-     * Decrypts an encrypted byte[] key material using the given a key version
+     * Decrypts an encrypted byte[] key material using the given key version
      * name and initialization vector.
      * 
      * @param encryptedKeyVersion
@@ -227,6 +227,26 @@ public class KeyProviderCryptoExtension extends
     public KeyVersion decryptEncryptedKey(
         EncryptedKeyVersion encryptedKeyVersion) throws IOException,
         GeneralSecurityException;
+
+    /**
+     * Re-encrypts an encrypted key version, using its initialization vector
+     * and key material, but with the latest key version name of its key name
+     * in the key provider.
+     * <p>
+     * If the latest key version name in the provider is the
+     * same as the one encrypted the passed-in encrypted key version, the same
+     * encrypted key version is returned.
+     * <p>
+     * NOTE: The generated key is not stored by the <code>KeyProvider</code>
+     *
+     * @param  ekv The EncryptedKeyVersion containing keyVersionName and IV.
+     * @return     The re-encrypted EncryptedKeyVersion.
+     * @throws IOException If the key material could not be re-encrypted.
+     * @throws GeneralSecurityException If the key material could not be
+     *                            re-encrypted because of a cryptographic issue.
+     */
+    EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+        throws IOException, GeneralSecurityException;
   }
 
   private static class DefaultCryptoExtension implements CryptoExtension {
@@ -258,24 +278,55 @@ public class KeyProviderCryptoExtension extends
       cc.generateSecureRandom(newKey);
       final byte[] iv = new byte[cc.getCipherSuite().getAlgorithmBlockSize()];
       cc.generateSecureRandom(iv);
+      Encryptor encryptor = cc.createEncryptor();
+      return generateEncryptedKey(encryptor, encryptionKey, newKey, iv);
+    }
+
+    private EncryptedKeyVersion generateEncryptedKey(final Encryptor encryptor,
+        final KeyVersion encryptionKey, final byte[] key, final byte[] iv)
+        throws IOException, GeneralSecurityException {
       // Encryption key IV is derived from new key's IV
       final byte[] encryptionIV = EncryptedKeyVersion.deriveIV(iv);
-      Encryptor encryptor = cc.createEncryptor();
       encryptor.init(encryptionKey.getMaterial(), encryptionIV);
-      int keyLen = newKey.length;
+      final int keyLen = key.length;
       ByteBuffer bbIn = ByteBuffer.allocateDirect(keyLen);
       ByteBuffer bbOut = ByteBuffer.allocateDirect(keyLen);
-      bbIn.put(newKey);
+      bbIn.put(key);
       bbIn.flip();
       encryptor.encrypt(bbIn, bbOut);
       bbOut.flip();
       byte[] encryptedKey = new byte[keyLen];
-      bbOut.get(encryptedKey);    
+      bbOut.get(encryptedKey);
-      return new EncryptedKeyVersion(encryptionKeyName,
+      return new EncryptedKeyVersion(encryptionKey.getName(),
           encryptionKey.getVersionName(), iv,
           new KeyVersion(encryptionKey.getName(), EEK, encryptedKey));
     }
 
+    @Override
+    public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+        throws IOException, GeneralSecurityException {
+      final String ekName = ekv.getEncryptionKeyName();
+      final KeyVersion ekNow = keyProvider.getCurrentKey(ekName);
+      Preconditions
+          .checkNotNull(ekNow, "KeyVersion name '%s' does not exist", ekName);
+      Preconditions.checkArgument(ekv.getEncryptedKeyVersion().getVersionName()
+              .equals(KeyProviderCryptoExtension.EEK),
+          "encryptedKey version name must be '%s', is '%s'",
+          KeyProviderCryptoExtension.EEK,
+          ekv.getEncryptedKeyVersion().getVersionName());
+
+      if (ekv.getEncryptedKeyVersion().equals(ekNow)) {
+        // no-op if same key version
+        return ekv;
+      }
+
+      final KeyVersion dek = decryptEncryptedKey(ekv);
+      final CryptoCodec cc = CryptoCodec.getInstance(keyProvider.getConf());
+      final Encryptor encryptor = cc.createEncryptor();
+      return generateEncryptedKey(encryptor, ekNow, dek.getMaterial(),
+          ekv.getEncryptedKeyIv());
+    }
+
     @Override
     public KeyVersion decryptEncryptedKey(
         EncryptedKeyVersion encryptedKeyVersion) throws IOException,
@@ -388,6 +439,28 @@ public class KeyProviderCryptoExtension extends
     return getExtension().decryptEncryptedKey(encryptedKey);
   }
 
+  /**
+   * Re-encrypts an encrypted key version, using its initialization vector
+   * and key material, but with the latest key version name of its key name
+   * in the key provider.
+   * <p>
+   * If the latest key version name in the provider is the
+   * same as the one encrypted the passed-in encrypted key version, the same
+   * encrypted key version is returned.
+   * <p>
+   * NOTE: The generated key is not stored by the <code>KeyProvider</code>
+   *
+   * @param  ekv The EncryptedKeyVersion containing keyVersionName and IV.
+   * @return     The re-encrypted EncryptedKeyVersion.
+   * @throws IOException If the key material could not be re-encrypted
+   * @throws GeneralSecurityException If the key material could not be
+   *                            re-encrypted because of a cryptographic issue.
+   */
+  public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+      throws IOException, GeneralSecurityException {
+    return getExtension().reencryptEncryptedKey(ekv);
+  }
+
   /**
    * Creates a <code>KeyProviderCryptoExtension</code> using a given
    * {@link KeyProvider}.

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/KMSClientProvider.java

@@ -146,7 +146,7 @@ public class KMSClientProvider extends KeyProvider implements CryptoExtension,
       List response = call(conn, null,
           HttpURLConnection.HTTP_OK, List.class);
       List<EncryptedKeyVersion> ekvs =
-          parseJSONEncKeyVersion(keyName, response);
+          parseJSONEncKeyVersions(keyName, response);
       keyQueue.addAll(ekvs);
     }
   }
@@ -221,39 +221,43 @@ public class KMSClientProvider extends KeyProvider implements CryptoExtension,
 
   @SuppressWarnings("rawtypes")
   private static List<EncryptedKeyVersion>
-      parseJSONEncKeyVersion(String keyName, List valueList) {
+      parseJSONEncKeyVersions(String keyName, List valueList) {
     List<EncryptedKeyVersion> ekvs = new LinkedList<EncryptedKeyVersion>();
     if (!valueList.isEmpty()) {
       for (Object values : valueList) {
         Map valueMap = (Map) values;
-
+        ekvs.add(parseJSONEncKeyVersion(keyName, valueMap));
-        String versionName = checkNotNull(
-                (String) valueMap.get(KMSRESTConstants.VERSION_NAME_FIELD),
-                KMSRESTConstants.VERSION_NAME_FIELD);
-
-        byte[] iv = Base64.decodeBase64(checkNotNull(
-                (String) valueMap.get(KMSRESTConstants.IV_FIELD),
-                KMSRESTConstants.IV_FIELD));
-
-        Map encValueMap = checkNotNull((Map)
-                valueMap.get(KMSRESTConstants.ENCRYPTED_KEY_VERSION_FIELD),
-                KMSRESTConstants.ENCRYPTED_KEY_VERSION_FIELD);
-
-        String encVersionName = checkNotNull((String)
-                encValueMap.get(KMSRESTConstants.VERSION_NAME_FIELD),
-                KMSRESTConstants.VERSION_NAME_FIELD);
-
-        byte[] encKeyMaterial = Base64.decodeBase64(checkNotNull((String)
-                encValueMap.get(KMSRESTConstants.MATERIAL_FIELD),
-                KMSRESTConstants.MATERIAL_FIELD));
-
-        ekvs.add(new KMSEncryptedKeyVersion(keyName, versionName, iv,
-            encVersionName, encKeyMaterial));
       }
     }
     return ekvs;
   }
 
+  private static EncryptedKeyVersion parseJSONEncKeyVersion(String keyName,
+      Map valueMap) {
+    String versionName = checkNotNull(
+        (String) valueMap.get(KMSRESTConstants.VERSION_NAME_FIELD),
+        KMSRESTConstants.VERSION_NAME_FIELD);
+
+    byte[] iv = Base64.decodeBase64(checkNotNull(
+        (String) valueMap.get(KMSRESTConstants.IV_FIELD),
+        KMSRESTConstants.IV_FIELD));
+
+    Map encValueMap = checkNotNull((Map)
+            valueMap.get(KMSRESTConstants.ENCRYPTED_KEY_VERSION_FIELD),
+        KMSRESTConstants.ENCRYPTED_KEY_VERSION_FIELD);
+
+    String encVersionName = checkNotNull((String)
+            encValueMap.get(KMSRESTConstants.VERSION_NAME_FIELD),
+        KMSRESTConstants.VERSION_NAME_FIELD);
+
+    byte[] encKeyMaterial = Base64.decodeBase64(checkNotNull((String)
+            encValueMap.get(KMSRESTConstants.MATERIAL_FIELD),
+        KMSRESTConstants.MATERIAL_FIELD));
+
+    return new KMSEncryptedKeyVersion(keyName, versionName, iv,
+        encVersionName, encKeyMaterial);
+  }
+
   private static KeyVersion parseJSONKeyVersion(Map valueMap) {
     KeyVersion keyVersion = null;
     if (!valueMap.isEmpty()) {
@@ -837,6 +841,35 @@ public class KMSClientProvider extends KeyProvider implements CryptoExtension,
     return parseJSONKeyVersion(response);
   }
 
+  @Override
+  public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+      throws IOException, GeneralSecurityException {
+    checkNotNull(ekv.getEncryptionKeyVersionName(), "versionName");
+    checkNotNull(ekv.getEncryptedKeyIv(), "iv");
+    checkNotNull(ekv.getEncryptedKeyVersion(), "encryptedKey");
+    Preconditions.checkArgument(ekv.getEncryptedKeyVersion().getVersionName()
+            .equals(KeyProviderCryptoExtension.EEK),
+        "encryptedKey version name must be '%s', is '%s'",
+        KeyProviderCryptoExtension.EEK,
+        ekv.getEncryptedKeyVersion().getVersionName());
+    final Map<String, String> params = new HashMap<>();
+    params.put(KMSRESTConstants.EEK_OP, KMSRESTConstants.EEK_REENCRYPT);
+    final Map<String, Object> jsonPayload = new HashMap<>();
+    jsonPayload.put(KMSRESTConstants.NAME_FIELD, ekv.getEncryptionKeyName());
+    jsonPayload.put(KMSRESTConstants.IV_FIELD,
+        Base64.encodeBase64String(ekv.getEncryptedKeyIv()));
+    jsonPayload.put(KMSRESTConstants.MATERIAL_FIELD,
+        Base64.encodeBase64String(ekv.getEncryptedKeyVersion().getMaterial()));
+    final URL url = createURL(KMSRESTConstants.KEY_VERSION_RESOURCE,
+        ekv.getEncryptionKeyVersionName(), KMSRESTConstants.EEK_SUB_RESOURCE,
+        params);
+    final HttpURLConnection conn = createConnection(url, HTTP_POST);
+    conn.setRequestProperty(CONTENT_TYPE, APPLICATION_JSON_MIME);
+    final Map response =
+        call(conn, jsonPayload, HttpURLConnection.HTTP_OK, Map.class);
+    return parseJSONEncKeyVersion(ekv.getEncryptionKeyName(), response);
+  }
+
   @Override
   public List<KeyVersion> getKeyVersions(String name) throws IOException {
     checkNotEmpty(name, "name");

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/KMSRESTConstants.java

@@ -42,6 +42,7 @@ public class KMSRESTConstants {
   public static final String EEK_GENERATE = "generate";
   public static final String EEK_DECRYPT = "decrypt";
   public static final String EEK_NUM_KEYS = "num_keys";
+  public static final String EEK_REENCRYPT = "reencrypt";
 
   public static final String IV_FIELD = "iv";
   public static final String NAME_FIELD = "name";

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/crypto/key/kms/LoadBalancingKMSClientProvider.java

@@ -218,6 +218,24 @@ public class LoadBalancingKMSClientProvider extends KeyProvider implements
     }
   }
 
+  public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion edek)
+      throws IOException, GeneralSecurityException {
+    try {
+      return doOp(new ProviderCallable<EncryptedKeyVersion>() {
+        @Override
+        public EncryptedKeyVersion call(KMSClientProvider provider)
+            throws IOException, GeneralSecurityException {
+          return provider.reencryptEncryptedKey(edek);
+        }
+      }, nextIdx());
+    } catch (WrapperException we) {
+      if (we.getCause() instanceof GeneralSecurityException) {
+        throw (GeneralSecurityException) we.getCause();
+      }
+      throw new IOException(we.getCause());
+    }
+  }
+
   @Override
   public KeyVersion getKeyVersion(final String versionName) throws IOException {
     return doOp(new ProviderCallable<KeyVersion>() {

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/crypto/key/TestKeyProviderCryptoExtension.java

@@ -33,7 +33,9 @@ import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension.EncryptedKeyVersion;
 import org.junit.Assert;
 import org.junit.BeforeClass;
+import org.junit.Rule;
 import org.junit.Test;
+import org.junit.rules.Timeout;
 
 import static org.apache.hadoop.crypto.key.KeyProvider.KeyVersion;
 import static org.junit.Assert.assertArrayEquals;
@@ -52,6 +54,9 @@ public class TestKeyProviderCryptoExtension {
   private static KeyProvider.Options options;
   private static KeyVersion encryptionKey;
 
+  @Rule
+  public Timeout testTimeout = new Timeout(180000);
+
   @BeforeClass
   public static void setup() throws Exception {
     conf = new Configuration();
@@ -138,6 +143,103 @@ public class TestKeyProviderCryptoExtension {
         manualMaterial, apiMaterial);
   }
 
+  @Test
+  public void testReencryptEncryptedKey() throws Exception {
+    // Generate a new EEK
+    final KeyProviderCryptoExtension.EncryptedKeyVersion ek1 =
+        kpExt.generateEncryptedKey(encryptionKey.getName());
+
+    // Decrypt EEK into an EK and check it
+    final KeyVersion k1 = kpExt.decryptEncryptedKey(ek1);
+    assertEquals(KeyProviderCryptoExtension.EK, k1.getVersionName());
+    assertEquals(encryptionKey.getMaterial().length, k1.getMaterial().length);
+    if (Arrays.equals(k1.getMaterial(), encryptionKey.getMaterial())) {
+      fail("Encrypted key material should not equal encryption key material");
+    }
+
+    // Roll the EK
+    kpExt.rollNewVersion(ek1.getEncryptionKeyName());
+
+    // Reencrypt ek1
+    final KeyProviderCryptoExtension.EncryptedKeyVersion ek2 =
+        kpExt.reencryptEncryptedKey(ek1);
+    assertEquals("Version name of EEK should be EEK",
+        KeyProviderCryptoExtension.EEK,
+        ek2.getEncryptedKeyVersion().getVersionName());
+    assertEquals("Name of EEK should be encryption key name",
+        ENCRYPTION_KEY_NAME, ek2.getEncryptionKeyName());
+    assertNotNull("Expected encrypted key material",
+        ek2.getEncryptedKeyVersion().getMaterial());
+    assertEquals("Length of encryption key material and EEK material should "
+            + "be the same", encryptionKey.getMaterial().length,
+        ek2.getEncryptedKeyVersion().getMaterial().length);
+    if (Arrays.equals(ek2.getEncryptedKeyVersion().getMaterial(),
+        encryptionKey.getMaterial())) {
+      fail("Encrypted key material should not equal decrypted key material");
+    }
+    if (Arrays.equals(ek2.getEncryptedKeyVersion().getMaterial(),
+        ek1.getEncryptedKeyVersion().getMaterial())) {
+      fail("Re-encrypted EEK should have different material");
+    }
+
+    // Decrypt the new EEK into an EK and check it
+    final KeyVersion k2 = kpExt.decryptEncryptedKey(ek2);
+    assertEquals(KeyProviderCryptoExtension.EK, k2.getVersionName());
+    assertEquals(encryptionKey.getMaterial().length, k2.getMaterial().length);
+    if (Arrays.equals(k2.getMaterial(), encryptionKey.getMaterial())) {
+      fail("Encrypted key material should not equal encryption key material");
+    }
+
+    // Re-encrypting the same EEK with the same EK should be deterministic
+    final KeyProviderCryptoExtension.EncryptedKeyVersion ek2a =
+        kpExt.reencryptEncryptedKey(ek1);
+    assertEquals("Version name of EEK should be EEK",
+        KeyProviderCryptoExtension.EEK,
+        ek2a.getEncryptedKeyVersion().getVersionName());
+    assertEquals("Name of EEK should be encryption key name",
+        ENCRYPTION_KEY_NAME, ek2a.getEncryptionKeyName());
+    assertNotNull("Expected encrypted key material",
+        ek2a.getEncryptedKeyVersion().getMaterial());
+    assertEquals("Length of encryption key material and EEK material should "
+            + "be the same", encryptionKey.getMaterial().length,
+        ek2a.getEncryptedKeyVersion().getMaterial().length);
+    if (Arrays.equals(ek2a.getEncryptedKeyVersion().getMaterial(),
+        encryptionKey.getMaterial())) {
+      fail("Encrypted key material should not equal decrypted key material");
+    }
+    if (Arrays.equals(ek2a.getEncryptedKeyVersion().getMaterial(),
+        ek1.getEncryptedKeyVersion().getMaterial())) {
+      fail("Re-encrypted EEK should have different material");
+    }
+    assertArrayEquals(ek2.getEncryptedKeyVersion().getMaterial(),
+        ek2a.getEncryptedKeyVersion().getMaterial());
+
+    // Re-encrypting an EEK with the same version EK should be no-op
+    final KeyProviderCryptoExtension.EncryptedKeyVersion ek3 =
+        kpExt.reencryptEncryptedKey(ek2);
+    assertEquals("Version name of EEK should be EEK",
+        KeyProviderCryptoExtension.EEK,
+        ek3.getEncryptedKeyVersion().getVersionName());
+    assertEquals("Name of EEK should be encryption key name",
+        ENCRYPTION_KEY_NAME, ek3.getEncryptionKeyName());
+    assertNotNull("Expected encrypted key material",
+        ek3.getEncryptedKeyVersion().getMaterial());
+    assertEquals("Length of encryption key material and EEK material should "
+            + "be the same", encryptionKey.getMaterial().length,
+        ek3.getEncryptedKeyVersion().getMaterial().length);
+    if (Arrays.equals(ek3.getEncryptedKeyVersion().getMaterial(),
+        encryptionKey.getMaterial())) {
+      fail("Encrypted key material should not equal decrypted key material");
+    }
+
+    if (Arrays.equals(ek3.getEncryptedKeyVersion().getMaterial(),
+        ek1.getEncryptedKeyVersion().getMaterial())) {
+      fail("Re-encrypted EEK should have different material");
+    }
+    assertArrayEquals(ek2.getEncryptedKeyVersion().getMaterial(),
+        ek3.getEncryptedKeyVersion().getMaterial());
+  }
+
   @Test
   public void testNonDefaultCryptoExtensionSelectionWithCachingKeyProvider()
           throws Exception {
@@ -233,6 +335,12 @@ public class TestKeyProviderCryptoExtension {
       return this.ekv;
     }
 
+    @Override
+    public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+        throws IOException, GeneralSecurityException {
+      return ekv;
+    }
+
     @Override
     public KeyVersion decryptEncryptedKey(
             EncryptedKeyVersion encryptedKeyVersion)
@@ -339,5 +447,11 @@ public class TestKeyProviderCryptoExtension {
             throws IOException, GeneralSecurityException {
       return kv;
     }
+
+    @Override
+    public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+        throws IOException, GeneralSecurityException {
+      return ekv;
+    }
   }
 }

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/EagerKeyGeneratorKeyProviderCryptoExtension.java

@@ -28,7 +28,6 @@ import java.util.concurrent.ExecutionException;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
-import org.apache.hadoop.crypto.key.KeyProvider;
 import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension;
 import org.apache.hadoop.crypto.key.kms.ValueQueue;
 import org.apache.hadoop.crypto.key.kms.ValueQueue.SyncGenerationPolicy;
@@ -136,6 +135,12 @@ public class EagerKeyGeneratorKeyProviderCryptoExtension
       return keyProviderCryptoExtension.decryptEncryptedKey(
           encryptedKeyVersion);
     }
+
+    @Override
+    public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+        throws IOException, GeneralSecurityException {
+      return keyProviderCryptoExtension.reencryptEncryptedKey(ekv);
+    }
   }
 
   /**

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/KMS.java

@@ -32,7 +32,6 @@ import org.apache.hadoop.security.token.delegation.web.HttpUserGroupInformation;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
@@ -65,7 +64,7 @@ public class KMS {
     CREATE_KEY, DELETE_KEY, ROLL_NEW_VERSION,
     GET_KEYS, GET_KEYS_METADATA,
     GET_KEY_VERSIONS, GET_METADATA, GET_KEY_VERSION, GET_CURRENT_KEY,
-    GENERATE_EEK, DECRYPT_EEK
+    GENERATE_EEK, DECRYPT_EEK, REENCRYPT_EEK
   }
 
   private KeyProviderCryptoExtension provider;
@@ -510,7 +509,7 @@ public class KMS {
   @Path(KMSRESTConstants.KEY_VERSION_RESOURCE + "/{versionName:.*}/" +
       KMSRESTConstants.EEK_SUB_RESOURCE)
   @Produces(MediaType.APPLICATION_JSON + "; " + JettyUtils.UTF_8)
-  public Response decryptEncryptedKey(
+  public Response handleEncryptedKeyOp(
       @PathParam("versionName") final String versionName,
       @QueryParam(KMSRESTConstants.EEK_OP) String eekOp,
       Map jsonPayload)
@@ -528,15 +527,15 @@ public class KMS {
       String ivStr = (String) jsonPayload.get(KMSRESTConstants.IV_FIELD);
       String encMaterialStr =
               (String) jsonPayload.get(KMSRESTConstants.MATERIAL_FIELD);
+      KMSClientProvider.checkNotNull(ivStr, KMSRESTConstants.IV_FIELD);
+      final byte[] iv = Base64.decodeBase64(ivStr);
+      KMSClientProvider.checkNotNull(encMaterialStr,
+          KMSRESTConstants.MATERIAL_FIELD);
+      final byte[] encMaterial = Base64.decodeBase64(encMaterialStr);
       Object retJSON;
       if (eekOp.equals(KMSRESTConstants.EEK_DECRYPT)) {
         assertAccess(KMSACLs.Type.DECRYPT_EEK, user, KMSOp.DECRYPT_EEK,
                 keyName);
-        KMSClientProvider.checkNotNull(ivStr, KMSRESTConstants.IV_FIELD);
-        final byte[] iv = Base64.decodeBase64(ivStr);
-        KMSClientProvider.checkNotNull(encMaterialStr,
-                KMSRESTConstants.MATERIAL_FIELD);
-        final byte[] encMaterial = Base64.decodeBase64(encMaterialStr);
 
         KeyProvider.KeyVersion retKeyVersion = user.doAs(
                 new PrivilegedExceptionAction<KeyVersion>() {
@@ -554,6 +553,23 @@ public class KMS {
 
         retJSON = KMSServerJSONUtils.toJSON(retKeyVersion);
         kmsAudit.ok(user, KMSOp.DECRYPT_EEK, keyName, "");
+      } else if (eekOp.equals(KMSRESTConstants.EEK_REENCRYPT)) {
+        assertAccess(KMSACLs.Type.GENERATE_EEK, user, KMSOp.REENCRYPT_EEK,
+            keyName);
+
+        EncryptedKeyVersion retEncryptedKeyVersion =
+            user.doAs(new PrivilegedExceptionAction<EncryptedKeyVersion>() {
+              @Override
+              public EncryptedKeyVersion run() throws Exception {
+                return provider.reencryptEncryptedKey(
+                    new KMSClientProvider.KMSEncryptedKeyVersion(keyName,
+                        versionName, iv, KeyProviderCryptoExtension.EEK,
+                        encMaterial));
+              }
+            });
+
+        retJSON = KMSServerJSONUtils.toJSON(retEncryptedKeyVersion);
+        kmsAudit.ok(user, KMSOp.REENCRYPT_EEK, keyName, "");
       } else {
         StringBuilder error;
         error = new StringBuilder("IllegalArgumentException Wrong ");
@@ -566,11 +582,11 @@ public class KMS {
         throw new IllegalArgumentException(error.toString());
       }
       KMSWebApp.getDecryptEEKCallsMeter().mark();
-      LOG.trace("Exiting decryptEncryptedKey method.");
+      LOG.trace("Exiting handleEncryptedKeyOp method.");
       return Response.ok().type(MediaType.APPLICATION_JSON).entity(retJSON)
-              .build();
+          .build();
     } catch (Exception e) {
-      LOG.debug("Exception in decryptEncryptedKey.", e);
+      LOG.debug("Exception in handleEncryptedKeyOp.", e);
       throw e;
     }
   }

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/KMSAudit.java

@@ -55,8 +55,8 @@ import java.util.Set;
 public class KMSAudit {
   @VisibleForTesting
   static final Set<KMS.KMSOp> AGGREGATE_OPS_WHITELIST = Sets.newHashSet(
-    KMS.KMSOp.GET_KEY_VERSION, KMS.KMSOp.GET_CURRENT_KEY,
+      KMS.KMSOp.GET_KEY_VERSION, KMS.KMSOp.GET_CURRENT_KEY,
-    KMS.KMSOp.DECRYPT_EEK, KMS.KMSOp.GENERATE_EEK
+      KMS.KMSOp.DECRYPT_EEK, KMS.KMSOp.GENERATE_EEK, KMS.KMSOp.REENCRYPT_EEK
   );
 
   private Cache<String, AuditEvent> cache;

hadoop-common-project/hadoop-kms/src/main/java/org/apache/hadoop/crypto/key/kms/server/KeyAuthorizationKeyProvider.java

@@ -264,6 +264,19 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
     }
   }
 
+  @Override
+  public EncryptedKeyVersion reencryptEncryptedKey(EncryptedKeyVersion ekv)
+      throws IOException, GeneralSecurityException {
+    readLock.lock();
+    try {
+      verifyKeyVersionBelongsToKey(ekv);
+      doAccessCheck(ekv.getEncryptionKeyName(), KeyOpType.GENERATE_EEK);
+      return provider.reencryptEncryptedKey(ekv);
+    } finally {
+      readLock.unlock();
+    }
+  }
+
   @Override
   public KeyVersion getKeyVersion(String versionName) throws IOException {
     readLock.lock();

hadoop-common-project/hadoop-kms/src/site/markdown/index.md.vm

@@ -169,7 +169,7 @@ Client-side can be changed via the following properties in the `etc/hadoop/core-
 
 $H3 KMS Aggregated Audit logs
 
-Audit logs are aggregated for API accesses to the GET\_KEY\_VERSION, GET\_CURRENT\_KEY, DECRYPT\_EEK, GENERATE\_EEK operations.
+Audit logs are aggregated for API accesses to the GET\_KEY\_VERSION, GET\_CURRENT\_KEY, DECRYPT\_EEK, GENERATE\_EEK, REENCRYPT\_EEK operations.
 
 Entries are grouped by the (user,key,operation) combined key for a configurable aggregation interval after which the number of accesses to the specified end-point by the user for a given key is flushed to the audit log.
 
@@ -481,7 +481,7 @@ $H5 Key ACLs
 KMS supports access control for all non-read operations at the Key level. All Key Access operations are classified as :
 
 * MANAGEMENT - createKey, deleteKey, rolloverNewVersion
-* GENERATE_EEK - generateEncryptedKey, warmUpEncryptedKeys
+* GENERATE_EEK - generateEncryptedKey, reencryptEncryptedKey, warmUpEncryptedKeys
 * DECRYPT_EEK - decryptEncryptedKey
 * READ - getKeyVersion, getKeyVersions, getMetadata, getKeysMetadata, getCurrentKey
 * ALL - all of the above
@@ -876,7 +876,7 @@ $H4 Generate Encrypted Key for Current KeyVersion
     Content-Type: application/json
     [
       {
-        "versionName"         : "encryptionVersionName",
+        "versionName"         : "<encryptionVersionName>",
         "iv"                  : "<iv>",          //base64
         "encryptedKeyVersion" : {
             "versionName"       : "EEK",
@@ -884,7 +884,7 @@ $H4 Generate Encrypted Key for Current KeyVersion
         }
       },
       {
-        "versionName"         : "encryptionVersionName",
+        "versionName"         : "<encryptionVersionName>",
         "iv"                  : "<iv>",          //base64
         "encryptedKeyVersion" : {
             "versionName"       : "EEK",
@@ -917,6 +917,37 @@ $H4 Decrypt Encrypted Key
       "material"    : "<material>",    //base64
     }
 
+$H4 Re-encrypt Encrypted Key With The Latest KeyVersion
+
+This command takes a previously generated encrypted key, and re-encrypts it using the latest KeyVersion encryption key in the KeyProvider. If the latest KeyVersion is the same as the one used to generate the encrypted key, the same encrypted key is returned.
+
+This is usually useful after a [Rollover](Rollover_Key) of an encryption key. Re-encrypting the encrypted key will allow it to be encrypted using the latest version of the encryption key, but still with the same key material and initialization vector.
+
+*REQUEST:*
+
+    POST http://HOST:PORT/kms/v1/keyversion/<version-name>/_eek?eek_op=reencrypt
+    Content-Type: application/json
+
+    {
+      "name"        : "<key-name>",
+      "iv"          : "<iv>",          //base64
+      "material"    : "<material>",    //base64
+    }
+
+*RESPONSE:*
+
+    200 OK
+    Content-Type: application/json
+
+    {
+      "versionName"         : "<encryptionVersionName>",
+      "iv"                  : "<iv>",            //base64
+      "encryptedKeyVersion" : {
+          "versionName"       : "EEK",
+          "material"          : "<material>",    //base64
+      }
+    }
+
 $H4 Get Key Version
 
 *REQUEST:*

hadoop-common-project/hadoop-kms/src/test/java/org/apache/hadoop/crypto/key/kms/server/TestKMS.java

@@ -68,6 +68,7 @@ import java.net.URI;
 import java.net.URL;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Collection;
 import java.util.Date;
 import java.util.HashMap;
@@ -79,6 +80,11 @@ import java.util.Set;
 import java.util.UUID;
 import java.util.concurrent.Callable;
 
+import static org.junit.Assert.assertArrayEquals;
+import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.assertFalse;
+import static org.junit.Assert.assertNotEquals;
+
 public class TestKMS {
   private static final Logger LOG = LoggerFactory.getLogger(TestKMS.class);
 
@@ -462,6 +468,7 @@ public class TestKMS {
   }
 
   @Test
+  @SuppressWarnings("checkstyle:methodlength")
   public void testKMSProvider() throws Exception {
     Configuration conf = new Configuration();
     conf.set("hadoop.security.authentication", "kerberos");
@@ -606,6 +613,25 @@ public class TestKMS {
         }
         Assert.assertFalse(isEq);
 
+        // test re-encrypt
+        kpExt.rollNewVersion(ek1.getEncryptionKeyName());
+        EncryptedKeyVersion ek1r = kpExt.reencryptEncryptedKey(ek1);
+        assertEquals(KeyProviderCryptoExtension.EEK,
+            ek1r.getEncryptedKeyVersion().getVersionName());
+        assertFalse(Arrays.equals(ek1.getEncryptedKeyVersion().getMaterial(),
+            ek1r.getEncryptedKeyVersion().getMaterial()));
+        assertEquals(kv.getMaterial().length,
+            ek1r.getEncryptedKeyVersion().getMaterial().length);
+        assertEquals(ek1.getEncryptionKeyName(), ek1r.getEncryptionKeyName());
+        assertArrayEquals(ek1.getEncryptedKeyIv(), ek1r.getEncryptedKeyIv());
+        assertNotEquals(ek1.getEncryptionKeyVersionName(),
+            ek1r.getEncryptionKeyVersionName());
+
+        KeyProvider.KeyVersion k1r = kpExt.decryptEncryptedKey(ek1r);
+        assertEquals(KeyProviderCryptoExtension.EK, k1r.getVersionName());
+        assertArrayEquals(k1.getMaterial(), k1r.getMaterial());
+        assertEquals(kv.getMaterial().length, k1r.getMaterial().length);
+
         // deleteKey()
         kp.deleteKey("k1");
 
@@ -721,6 +747,7 @@ public class TestKMS {
   }
 
   @Test
+  @SuppressWarnings("checkstyle:methodlength")
   public void testKeyACLs() throws Exception {
     Configuration conf = new Configuration();
     conf.set("hadoop.security.authentication", "kerberos");
@@ -969,17 +996,10 @@ public class TestKMS {
           @Override
           public Void run() throws Exception {
             KeyProvider kp = createProvider(uri, conf);
-            try {
+            KeyProviderCryptoExtension kpce =
-              KeyProviderCryptoExtension kpce =
+                KeyProviderCryptoExtension.createKeyProviderCryptoExtension(kp);
-                  KeyProviderCryptoExtension.createKeyProviderCryptoExtension(kp);
+            EncryptedKeyVersion ekv = kpce.generateEncryptedKey("k1");
-              try {
+            kpce.reencryptEncryptedKey(ekv);
-                kpce.generateEncryptedKey("k1");
-              } catch (Exception e) {
-                Assert.fail("User [GENERATE_EEK] should be allowed to generate_eek on k1");
-              }
-            } catch (Exception ex) {
-              Assert.fail(ex.getMessage());
-            }
             return null;
           }
         });
@@ -1170,6 +1190,7 @@ public class TestKMS {
   }
 
   @Test
+  @SuppressWarnings("checkstyle:methodlength")
   public void testACLs() throws Exception {
     Configuration conf = new Configuration();
     conf.set("hadoop.security.authentication", "kerberos");
@@ -1401,6 +1422,17 @@ public class TestKMS {
           }
         });
 
+        doAs("GENERATE_EEK", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            KeyProvider kp = createProvider(uri, conf);
+            KeyProviderCryptoExtension kpCE = KeyProviderCryptoExtension.
+                createKeyProviderCryptoExtension(kp);
+            kpCE.reencryptEncryptedKey(encKv);
+            return null;
+          }
+        });
+
         doAs("DECRYPT_EEK", new PrivilegedExceptionAction<Void>() {
           @Override
           public Void run() throws Exception {
@@ -1449,6 +1481,7 @@ public class TestKMS {
         // test ACL reloading
         Thread.sleep(10); // to ensure the ACLs file modifiedTime is newer
         conf.set(KMSACLs.Type.CREATE.getAclConfigKey(), "foo");
+        conf.set(KMSACLs.Type.GENERATE_EEK.getAclConfigKey(), "foo");
         writeConf(testDir, conf);
         Thread.sleep(1000);
 
@@ -1473,6 +1506,41 @@ public class TestKMS {
           }
         });
 
+        doAs("GENERATE_EEK", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            KeyProvider kp = createProvider(uri, conf);
+            try {
+              KeyProviderCryptoExtension kpCE = KeyProviderCryptoExtension.
+                  createKeyProviderCryptoExtension(kp);
+              kpCE.generateEncryptedKey("k1");
+            } catch (IOException ex) {
+              // This isn't an AuthorizationException because generate goes
+              // through the ValueQueue. See KMSCP#generateEncryptedKey.
+              if (ex.getCause().getCause() instanceof AuthorizationException) {
+                LOG.info("Caught expected exception.", ex);
+              } else {
+                throw ex;
+              }
+            }
+            return null;
+          }
+        });
+
+        doAs("GENERATE_EEK", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            KeyProvider kp = createProvider(uri, conf);
+            try {
+              KeyProviderCryptoExtension kpCE = KeyProviderCryptoExtension.
+                  createKeyProviderCryptoExtension(kp);
+              kpCE.reencryptEncryptedKey(encKv);
+            } catch (AuthorizationException ex) {
+              LOG.info("Caught expected exception.", ex);
+            }
+            return null;
+          }
+        });
         return null;
       }
     });

hadoop-common-project/hadoop-kms/src/test/java/org/apache/hadoop/crypto/key/kms/server/TestKMSAudit.java

@@ -95,6 +95,7 @@ public class TestKMSAudit {
   }
 
   @Test
+  @SuppressWarnings("checkstyle:linelength")
   public void testAggregation() throws Exception {
     UserGroupInformation luser = Mockito.mock(UserGroupInformation.class);
     Mockito.when(luser.getShortUserName()).thenReturn("luser");
@@ -109,6 +110,10 @@ public class TestKMSAudit {
     kmsAudit.evictCacheForTesting();
     kmsAudit.ok(luser, KMSOp.DECRYPT_EEK, "k1", "testmsg");
     kmsAudit.evictCacheForTesting();
+    kmsAudit.ok(luser, KMSOp.REENCRYPT_EEK, "k1", "testmsg");
+    kmsAudit.ok(luser, KMSOp.REENCRYPT_EEK, "k1", "testmsg");
+    kmsAudit.ok(luser, KMSOp.REENCRYPT_EEK, "k1", "testmsg");
+    kmsAudit.evictCacheForTesting();
     String out = getAndResetLogOutput();
     System.out.println(out);
     Assert.assertTrue(
@@ -119,10 +124,13 @@ public class TestKMSAudit {
             + "OK\\[op=ROLL_NEW_VERSION, key=k1, user=luser\\] testmsg"
             // Aggregated
             + "OK\\[op=DECRYPT_EEK, key=k1, user=luser, accessCount=6, interval=[^m]{1,4}ms\\] testmsg"
-            + "OK\\[op=DECRYPT_EEK, key=k1, user=luser, accessCount=1, interval=[^m]{1,4}ms\\] testmsg"));
+            + "OK\\[op=DECRYPT_EEK, key=k1, user=luser, accessCount=1, interval=[^m]{1,4}ms\\] testmsg"
+            + "OK\\[op=REENCRYPT_EEK, key=k1, user=luser, accessCount=1, interval=[^m]{1,4}ms\\] testmsg"
+            + "OK\\[op=REENCRYPT_EEK, key=k1, user=luser, accessCount=3, interval=[^m]{1,4}ms\\] testmsg"));
   }
 
   @Test
+  @SuppressWarnings("checkstyle:linelength")
   public void testAggregationUnauth() throws Exception {
     UserGroupInformation luser = Mockito.mock(UserGroupInformation.class);
     Mockito.when(luser.getShortUserName()).thenReturn("luser");
@@ -159,6 +167,7 @@ public class TestKMSAudit {
   }
 
   @Test
+  @SuppressWarnings("checkstyle:linelength")
   public void testAuditLogFormat() throws Exception {
     UserGroupInformation luser = Mockito.mock(UserGroupInformation.class);
     Mockito.when(luser.getShortUserName()).thenReturn("luser");