HADOOP-17304. KMS ACL: Allow DeleteKey Operation to Invalidate Cache. Contributed by Xiaoyu.

Reviewed-by: Ayush Saxena <ayushsaxena@apache.org>
Signed-off-by: He Xiaoqiao <hexiaoqiao@apache.org>

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

@@ -52,10 +52,12 @@ import java.io.IOException;
 import java.net.URI;
 import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
+import java.util.EnumSet;
 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 
+import static org.apache.hadoop.crypto.key.kms.server.KMSACLs.INVALIDATE_CACHE_TYPES;
 import static org.apache.hadoop.util.KMSUtil.checkNotEmpty;
 import static org.apache.hadoop.util.KMSUtil.checkNotNull;
 
@@ -95,6 +97,12 @@ public class KMS {
     KMSWebApp.getACLs().assertAccess(aclType, ugi, operation, key);
   }
 
+  private void assertAccess(EnumSet<KMSACLs.Type> aclTypes,
+      UserGroupInformation ugi, KMSOp operation, String key)
+      throws AccessControlException {
+    KMSWebApp.getACLs().assertAccess(aclTypes, ugi, operation, key);
+  }
+
   private static KeyProvider.KeyVersion removeKeyMaterial(
       KeyProvider.KeyVersion keyVersion) {
     return new KMSClientProvider.KMSKeyVersion(keyVersion.getName(),
@@ -270,7 +278,7 @@ public class KMS {
       KMSWebApp.getAdminCallsMeter().mark();
       checkNotEmpty(name, "name");
       UserGroupInformation user = HttpUserGroupInformation.get();
-      assertAccess(KMSACLs.Type.ROLLOVER, user, KMSOp.INVALIDATE_CACHE, name);
+      assertAccess(INVALIDATE_CACHE_TYPES, user, KMSOp.INVALIDATE_CACHE, name);
       LOG.debug("Invalidating cache with key name {}.", name);
 
       user.doAs(new PrivilegedExceptionAction<Void>() {

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

@@ -29,6 +29,7 @@ import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import java.util.EnumSet;
 import java.util.HashMap;
 import java.util.Map;
 import java.util.concurrent.Executors;
@@ -69,6 +70,10 @@ public class KMSACLs implements Runnable, KeyACLs {
 
   public static final int RELOADER_SLEEP_MILLIS = 1000;
 
+  // Allow both ROLLOVER and DELETE to invalidate cache.
+  public static final EnumSet<KMSACLs.Type> INVALIDATE_CACHE_TYPES =
+      EnumSet.of(KMSACLs.Type.ROLLOVER, KMSACLs.Type.DELETE);
+
   private volatile Map<Type, AccessControlList> acls;
   private volatile Map<Type, AccessControlList> blacklistedAcls;
   @VisibleForTesting
@@ -273,6 +278,27 @@ public class KMSACLs implements Runnable, KeyACLs {
     }
   }
 
+  public void assertAccess(EnumSet<Type> aclTypes,
+      UserGroupInformation ugi, KMSOp operation, String key)
+      throws AccessControlException {
+    boolean accessAllowed = false;
+    for (KMSACLs.Type type : aclTypes) {
+      if (KMSWebApp.getACLs().hasAccess(type, ugi)){
+        accessAllowed = true;
+        break;
+      }
+    }
+
+    if (!accessAllowed) {
+      KMSWebApp.getUnauthorizedCallsMeter().mark();
+      KMSWebApp.getKMSAudit().unauthorized(ugi, operation, key);
+      throw new AuthorizationException(String.format(
+          (key != null) ? UNAUTHORIZED_MSG_WITH_KEY
+              : UNAUTHORIZED_MSG_WITHOUT_KEY,
+          ugi.getShortUserName(), operation, key));
+    }
+  }
+
   @Override
   public boolean hasAccessToKey(String keyName, UserGroupInformation ugi,
       KeyOpType opType) {