HADOOP-11341. KMS support for whitelist key ACLs. Contributed by Arun Suresh.

(cherry picked from commit 31b4d2daa1)

hadoop-common-project/hadoop-common/CHANGES.txt

@@ -39,6 +39,8 @@ Release 2.7.0 - UNRELEASED
     HADOOP-11257: Update "hadoop jar" documentation to warn against using it
     for launching yarn jars (iwasakims via cmccabe)
 
+    HADOOP-11341. KMS support for whitelist key ACLs. (Arun Suresh via wang)
+
   OPTIMIZATIONS
 
     HADOOP-11323. WritableComparator#compare keeps reference to byte array.

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

@@ -73,6 +73,8 @@ public class KMSACLs implements Runnable, KeyACLs {
   private volatile Map<String, HashMap<KeyOpType, AccessControlList>> keyAcls;
   private final Map<KeyOpType, AccessControlList> defaultKeyAcls =
       new HashMap<KeyOpType, AccessControlList>();
+  private final Map<KeyOpType, AccessControlList> whitelistKeyAcls =
+      new HashMap<KeyOpType, AccessControlList>();
   private ScheduledExecutorService executorService;
   private long lastReload;
 
@@ -151,11 +153,21 @@ public class KMSACLs implements Runnable, KeyACLs {
         String aclStr = conf.get(confKey);
         if (aclStr != null) {
           if (aclStr.equals("*")) {
-            LOG.info("Default Key ACL for  KEY_OP '{}' is set to '*'", keyOp);
+            LOG.info("Default Key ACL for KEY_OP '{}' is set to '*'", keyOp);
           }
           defaultKeyAcls.put(keyOp, new AccessControlList(aclStr));
         }
       }
+      if (!whitelistKeyAcls.containsKey(keyOp)) {
+        String confKey = KMSConfiguration.WHITELIST_KEY_ACL_PREFIX + keyOp;
+        String aclStr = conf.get(confKey);
+        if (aclStr != null) {
+          if (aclStr.equals("*")) {
+            LOG.info("Whitelist Key ACL for KEY_OP '{}' is set to '*'", keyOp);
+          }
+          whitelistKeyAcls.put(keyOp, new AccessControlList(aclStr));
+        }
+      }
     }
   }
 
@@ -229,13 +241,23 @@ public class KMSACLs implements Runnable, KeyACLs {
   @Override
   public boolean hasAccessToKey(String keyName, UserGroupInformation ugi,
       KeyOpType opType) {
+    return checkKeyAccess(keyName, ugi, opType)
+        || checkKeyAccess(whitelistKeyAcls, ugi, opType);
+  }
+
+  private boolean checkKeyAccess(String keyName, UserGroupInformation ugi,
+      KeyOpType opType) {
     Map<KeyOpType, AccessControlList> keyAcl = keyAcls.get(keyName);
     if (keyAcl == null) {
-      // Get KeyAcl map of DEFAULT KEY.
+      // If No key acl defined for this key, check to see if
+      // there are key defaults configured for this operation
       keyAcl = defaultKeyAcls;
     }
-    // If No key acl defined for this key, check to see if
+    return checkKeyAccess(keyAcl, ugi, opType);
-    // there are key defaults configured for this operation
+  }
+
+  private boolean checkKeyAccess(Map<KeyOpType, AccessControlList> keyAcl,
+      UserGroupInformation ugi, KeyOpType opType) {
     AccessControlList acl = keyAcl.get(opType);
     if (acl == null) {
       // If no acl is specified for this operation,
@@ -246,6 +268,7 @@ public class KMSACLs implements Runnable, KeyACLs {
     }
   }
 
+
   @Override
   public boolean isACLPresent(String keyName, KeyOpType opType) {
     return (keyAcls.containsKey(keyName) || defaultKeyAcls.containsKey(opType));

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

@@ -39,6 +39,7 @@ public class KMSConfiguration {
 
   public static final String KEY_ACL_PREFIX = "key.acl.";
   public static final String DEFAULT_KEY_ACL_PREFIX = "default.key.acl.";
+  public static final String WHITELIST_KEY_ACL_PREFIX = "whitelist.key.acl.";
 
   // Property to set the backing KeyProvider
   public static final String KEY_PROVIDER_URI = CONFIG_PREFIX +

hadoop-common-project/hadoop-kms/src/site/apt/index.apt.vm

@@ -467,10 +467,20 @@ $ keytool -genkey -alias tomcat -keyalg RSA
   is possible to configure a default key access control for a subset of the
   operation types.
 
+  It is also possible to configure a "whitelist" key ACL for a subset of the
+  operation types. The whitelist key ACL is a whitelist in addition to the
+  explicit or default per-key ACL. That is, if no per-key ACL is explicitly
+  set, a user will be granted access if they are present in the default per-key
+  ACL or the whitelist key ACL. If a per-key ACL is explicitly set, a user
+  will be granted access if they are present in the per-key ACL or the
+  whitelist key ACL.
+
   If no ACL is configured for a specific key AND no default ACL is configured
-  for the requested operation, then access will be DENIED.
+  AND no root key ACL is configured for the requested operation,
+  then access will be DENIED.
   
-  <<NOTE:>> The default ACL does not support <<<ALL>>> operation qualifier.
+  <<NOTE:>> The default and whitelist key ACL does not support <<<ALL>>>
+            operation qualifier.
   
 +---+
   <property>
@@ -491,7 +501,7 @@ $ keytool -genkey -alias tomcat -keyalg RSA
 
   <property>
     <name>key.acl.testKey3.DECRYPT_EEK</name>
-    <value>*</value>
+    <value>admink3</value>
     <description>
       ACL for decryptEncryptedKey operations.
     </description>
@@ -514,6 +524,28 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>whitelist.key.acl.MANAGEMENT</name>
+    <value>admin1</value>
+    <description>
+      whitelist ACL for MANAGEMENT operations for all keys.
+    </description>
+  </property>
+
+  <!--
+  'testKey3' key ACL is defined. Since a 'whitelist'
+  key is also defined for DECRYPT_EEK, in addition to
+  admink3, admin1 can also perform DECRYPT_EEK operations
+  on 'testKey3'
+  -->
+  <property>
+    <name>whitelist.key.acl.DECRYPT_EEK</name>
+    <value>admin1</value>
+    <description>
+      whitelist ACL for DECRYPT_EEK operations for all keys.
+    </description>
+  </property>
+
   <property>
     <name>default.key.acl.MANAGEMENT</name>
     <value>user1,user2</value>

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

@@ -617,13 +617,15 @@ public class TestKMS {
     for (KMSACLs.Type type : KMSACLs.Type.values()) {
       conf.set(type.getAclConfigKey(), type.toString());
     }
-    conf.set(KMSACLs.Type.CREATE.getAclConfigKey(),"CREATE,ROLLOVER,GET,SET_KEY_MATERIAL,GENERATE_EEK");
+    conf.set(KMSACLs.Type.CREATE.getAclConfigKey(),"CREATE,ROLLOVER,GET,SET_KEY_MATERIAL,GENERATE_EEK,DECRYPT_EEK");
-    conf.set(KMSACLs.Type.ROLLOVER.getAclConfigKey(),"CREATE,ROLLOVER,GET,SET_KEY_MATERIAL,GENERATE_EEK");
+    conf.set(KMSACLs.Type.ROLLOVER.getAclConfigKey(),"CREATE,ROLLOVER,GET,SET_KEY_MATERIAL,GENERATE_EEK,DECRYPT_EEK");
     conf.set(KMSACLs.Type.GENERATE_EEK.getAclConfigKey(),"CREATE,ROLLOVER,GET,SET_KEY_MATERIAL,GENERATE_EEK");
     conf.set(KMSACLs.Type.DECRYPT_EEK.getAclConfigKey(),"CREATE,ROLLOVER,GET,SET_KEY_MATERIAL,GENERATE_EEK");
 
-
     conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "test_key.MANAGEMENT", "CREATE");
+    conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "some_key.MANAGEMENT", "ROLLOVER");
+    conf.set(KMSConfiguration.WHITELIST_KEY_ACL_PREFIX + "MANAGEMENT", "DECRYPT_EEK");
+
     conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "all_access.ALL", "GENERATE_EEK");
     conf.set(KeyAuthorizationKeyProvider.KEY_ACL + "all_access.DECRYPT_EEK", "ROLLOVER");
     conf.set(KMSConfiguration.DEFAULT_KEY_ACL_PREFIX + "MANAGEMENT", "ROLLOVER");
@@ -676,6 +678,41 @@ public class TestKMS {
           }
         });
 
+        // Test whitelist key access..
+        // DECRYPT_EEK is whitelisted for MANAGEMENT operations only
+        doAs("DECRYPT_EEK", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            KeyProvider kp = new KMSClientProvider(uri, conf);
+            try {
+              Options options = new KeyProvider.Options(conf);
+              Map<String, String> attributes = options.getAttributes();
+              HashMap<String,String> newAttribs = new HashMap<String, String>(attributes);
+              newAttribs.put("key.acl.name", "some_key");
+              options.setAttributes(newAttribs);
+              KeyProvider.KeyVersion kv = kp.createKey("kk0", options);
+              Assert.assertNull(kv.getMaterial());
+              KeyVersion rollVersion = kp.rollNewVersion("kk0");
+              Assert.assertNull(rollVersion.getMaterial());
+              KeyProviderCryptoExtension kpce =
+                  KeyProviderCryptoExtension.createKeyProviderCryptoExtension(kp);
+              try {
+                kpce.generateEncryptedKey("kk0");
+                Assert.fail("User [DECRYPT_EEK] should not be allowed to generate_eek on kk0");
+              } catch (Exception e) {
+                // Ignore
+              }
+              newAttribs = new HashMap<String, String>(attributes);
+              newAttribs.put("key.acl.name", "all_access");
+              options.setAttributes(newAttribs);
+              kp.createKey("kkx", options);
+            } catch (Exception ex) {
+              Assert.fail(ex.getMessage());
+            }
+            return null;
+          }
+        });
+
         doAs("ROLLOVER", new PrivilegedExceptionAction<Void>() {
           @Override
           public Void run() throws Exception {