HADOOP-10863. KMS should have a blacklist for decrypting EEKs. (asuresh via tucu)

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

@@ -493,6 +493,9 @@ Release 2.6.0 - UNRELEASED
 
     HADOOP-10990. Add missed NFSv3 request and response classes (brandonli)
 
+    HADOOP-10863. KMS should have a blacklist for decrypting EEKs. 
+    (asuresh via tucu)
+
   OPTIMIZATIONS
 
     HADOOP-10838. Byte array native checksumming. (James Thomas via todd)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/security/authorize/AccessControlList.java

@@ -221,7 +221,13 @@ public class AccessControlList implements Writable {
     return groups;
   }
 
-  public boolean isUserAllowed(UserGroupInformation ugi) {
+  /**
+   * Checks if a user represented by the provided {@link UserGroupInformation}
+   * is a member of the Access Control List
+   * @param ugi UserGroupInformation to check if contained in the ACL
+   * @return true if ugi is member of the list
+   */
+  public final boolean isUserInList(UserGroupInformation ugi) {
     if (allAllowed || users.contains(ugi.getShortUserName())) {
       return true;
     } else {
@@ -234,6 +240,10 @@ public class AccessControlList implements Writable {
     return false;
   }
 
+  public boolean isUserAllowed(UserGroupInformation ugi) {
+    return isUserInList(ugi);
+  }
+
   /**
    * Returns descriptive way of users and groups that are part of this ACL.
    * Use {@link #getAclString()} to get the exact String that can be given to

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

@@ -26,10 +26,10 @@ import org.apache.hadoop.crypto.key.KeyProviderCryptoExtension.EncryptedKeyVersi
 import org.apache.hadoop.crypto.key.kms.KMSRESTConstants;
 import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.apache.hadoop.crypto.key.kms.KMSClientProvider;
 import org.apache.hadoop.security.token.delegation.web.HttpUserGroupInformation;
 
+
 import javax.ws.rs.Consumes;
 import javax.ws.rs.DELETE;
 import javax.ws.rs.DefaultValue;
@@ -73,29 +73,14 @@ public class KMS {
     kmsAudit= KMSWebApp.getKMSAudit();
   }
 
-
-  private static final String UNAUTHORIZED_MSG_WITH_KEY = 
-      "User:%s not allowed to do '%s' on '%s'";
-  
-  private static final String UNAUTHORIZED_MSG_WITHOUT_KEY = 
-      "User:%s not allowed to do '%s'";
-
   private void assertAccess(KMSACLs.Type aclType, UserGroupInformation ugi,
       KMSOp operation) throws AccessControlException {
-    assertAccess(aclType, ugi, operation, null);
+    KMSWebApp.getACLs().assertAccess(aclType, ugi, operation, null);
   }
-
-  private void assertAccess(KMSACLs.Type aclType,
-      UserGroupInformation ugi, KMSOp operation, String key)
-      throws AccessControlException {
-    if (!KMSWebApp.getACLs().hasAccess(aclType, ugi)) {
-      KMSWebApp.getUnauthorizedCallsMeter().mark();
-      kmsAudit.unauthorized(ugi, operation, key);
-      throw new AuthorizationException(String.format(
-          (key != null) ? UNAUTHORIZED_MSG_WITH_KEY 
-                        : UNAUTHORIZED_MSG_WITHOUT_KEY,
-          ugi.getShortUserName(), operation, key));
-    }
+  
+  private void assertAccess(KMSACLs.Type aclType, UserGroupInformation ugi,
+      KMSOp operation, String key) throws AccessControlException {
+    KMSWebApp.getACLs().assertAccess(aclType, ugi, operation, key);
   }
 
   private static KeyProvider.KeyVersion removeKeyMaterial(

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

@@ -19,8 +19,11 @@ package org.apache.hadoop.crypto.key.kms.server;
 
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.crypto.key.kms.server.KMS.KMSOp;
+import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.authorize.AccessControlList;
+import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -39,14 +42,23 @@ import java.util.concurrent.TimeUnit;
 public class KMSACLs implements Runnable {
   private static final Logger LOG = LoggerFactory.getLogger(KMSACLs.class);
 
+  private static final String UNAUTHORIZED_MSG_WITH_KEY =
+      "User:%s not allowed to do '%s' on '%s'";
+
+  private static final String UNAUTHORIZED_MSG_WITHOUT_KEY =
+      "User:%s not allowed to do '%s'";
 
   public enum Type {
     CREATE, DELETE, ROLLOVER, GET, GET_KEYS, GET_METADATA,
     SET_KEY_MATERIAL, GENERATE_EEK, DECRYPT_EEK;
 
-    public String getConfigKey() {
+    public String getAclConfigKey() {
       return KMSConfiguration.CONFIG_PREFIX + "acl." + this.toString();
     }
+
+    public String getBlacklistConfigKey() {
+      return KMSConfiguration.CONFIG_PREFIX + "blacklist." + this.toString();
+    }
   }
 
   public static final String ACL_DEFAULT = AccessControlList.WILDCARD_ACL_VALUE;
@@ -54,6 +66,7 @@ public class KMSACLs implements Runnable {
   public static final int RELOADER_SLEEP_MILLIS = 1000;
 
   private volatile Map<Type, AccessControlList> acls;
+  private volatile Map<Type, AccessControlList> blacklistedAcls;
   private ScheduledExecutorService executorService;
   private long lastReload;
 
@@ -70,12 +83,20 @@ public class KMSACLs implements Runnable {
 
   private void setACLs(Configuration conf) {
     Map<Type, AccessControlList> tempAcls = new HashMap<Type, AccessControlList>();
+    Map<Type, AccessControlList> tempBlacklist = new HashMap<Type, AccessControlList>();
     for (Type aclType : Type.values()) {
-      String aclStr = conf.get(aclType.getConfigKey(), ACL_DEFAULT);
+      String aclStr = conf.get(aclType.getAclConfigKey(), ACL_DEFAULT);
       tempAcls.put(aclType, new AccessControlList(aclStr));
+      String blacklistStr = conf.get(aclType.getBlacklistConfigKey());
+      if (blacklistStr != null) {
+        // Only add if blacklist is present
+        tempBlacklist.put(aclType, new AccessControlList(blacklistStr));
+        LOG.info("'{}' Blacklist '{}'", aclType, blacklistStr);
+      }
       LOG.info("'{}' ACL '{}'", aclType, aclStr);
     }
     acls = tempAcls;
+    blacklistedAcls = tempBlacklist;
   }
 
   @Override
@@ -109,12 +130,38 @@ public class KMSACLs implements Runnable {
     lastReload = System.currentTimeMillis();
     Configuration conf = KMSConfiguration.getACLsConf();
     // triggering the resource loading.
-    conf.get(Type.CREATE.getConfigKey());
+    conf.get(Type.CREATE.getAclConfigKey());
     return conf;
   }
 
+  /**
+   * First Check if user is in ACL for the KMS operation, if yes, then
+   * return true if user is not present in any configured blacklist for
+   * the operation
+   * @param type KMS Operation
+   * @param ugi UserGroupInformation of user
+   * @return true is user has access
+   */
   public boolean hasAccess(Type type, UserGroupInformation ugi) {
-    return acls.get(type).isUserAllowed(ugi);
+    boolean access = acls.get(type).isUserAllowed(ugi);
+    if (access) {
+      AccessControlList blacklist = blacklistedAcls.get(type);
+      access = (blacklist == null) || !blacklist.isUserInList(ugi);
+    }
+    return access;
+  }
+
+  public void assertAccess(KMSACLs.Type aclType,
+      UserGroupInformation ugi, KMSOp operation, String key)
+      throws AccessControlException {
+    if (!KMSWebApp.getACLs().hasAccess(aclType, ugi)) {
+      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));
+    }
   }
 
 }

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

@@ -274,8 +274,13 @@ $ keytool -genkey -alias tomcat -keyalg RSA
   KMS ACLs configuration are defined in the KMS <<<etc/hadoop/kms-acls.xml>>>
   configuration file. This file is hot-reloaded when it changes.
 
-  KMS supports a fine grained access control via a set ACL
-  configuration properties:
+  KMS supports both fine grained access control as well as blacklist for kms
+  operations via a set ACL configuration properties.
+
+  A user accessing KMS is first checked for inclusion in the Access Control
+  List for the requested operation and then checked for exclusion in the
+  Black list for the operation before access is granted.
+
 
 +---+
   <property>
@@ -288,6 +293,16 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.CREATE</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for create-key operations.
+      If the user does is in the Blacklist, the key material is not returned
+      as part of the response.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.DELETE</name>
     <value>*</value>
@@ -296,6 +311,14 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.DELETE</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for delete-key operations.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.ROLLOVER</name>
     <value>*</value>
@@ -306,6 +329,14 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.ROLLOVER</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for rollover-key operations.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.GET</name>
     <value>*</value>
@@ -314,6 +345,14 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.GET</name>
+    <value>hdfs,foo</value>
+    <description>
+      ACL for get-key-version and get-current-key operations.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.GET_KEYS</name>
     <value>*</value>
@@ -322,6 +361,14 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.GET_KEYS</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for get-keys operation.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.GET_METADATA</name>
     <value>*</value>
@@ -330,6 +377,14 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.GET_METADATA</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for get-key-metadata and get-keys-metadata operations.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.SET_KEY_MATERIAL</name>
     <value>*</value>
@@ -339,6 +394,15 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.SET_KEY_MATERIAL</name>
+    <value>hdfs,foo</value>
+    <description>
+        Complimentary Blacklist for CREATE and ROLLOVER operation to allow the client
+        to provide the key material when creating or rolling a key.
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.GENERATE_EEK</name>
     <value>*</value>
@@ -348,6 +412,15 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 
+  <property>
+    <name>hadoop.kms.blacklist.GENERATE_EEK</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for generateEncryptedKey
+      CryptoExtension operations
+    </description>
+  </property>
+
   <property>
     <name>hadoop.kms.acl.DECRYPT_EEK</name>
     <value>*</value>
@@ -357,6 +430,17 @@ $ keytool -genkey -alias tomcat -keyalg RSA
     </description>
   </property>
 </configuration>
+
+  <property>
+    <name>hadoop.kms.blacklist.DECRYPT_EEK</name>
+    <value>hdfs,foo</value>
+    <description>
+      Blacklist for decrypt EncryptedKey
+      CryptoExtension operations
+    </description>
+  </property>
+</configuration>
+
 +---+
 
 ** KMS Delegation Token Configuration

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

@@ -268,6 +268,8 @@ public class TestKMS {
     List<String> principals = new ArrayList<String>();
     principals.add("HTTP/localhost");
     principals.add("client");
+    principals.add("hdfs");
+    principals.add("otheradmin");
     principals.add("client/host");
     principals.add("client1");
     for (KMSACLs.Type type : KMSACLs.Type.values()) {
@@ -621,12 +623,12 @@ public class TestKMS {
     conf.set("hadoop.kms.authentication.kerberos.name.rules", "DEFAULT");
 
     for (KMSACLs.Type type : KMSACLs.Type.values()) {
-      conf.set(type.getConfigKey(), type.toString());
+      conf.set(type.getAclConfigKey(), type.toString());
     }
-    conf.set(KMSACLs.Type.CREATE.getConfigKey(),
+    conf.set(KMSACLs.Type.CREATE.getAclConfigKey(),
         KMSACLs.Type.CREATE.toString() + ",SET_KEY_MATERIAL");
 
-    conf.set(KMSACLs.Type.ROLLOVER.getConfigKey(),
+    conf.set(KMSACLs.Type.ROLLOVER.getAclConfigKey(),
         KMSACLs.Type.ROLLOVER.toString() + ",SET_KEY_MATERIAL");
 
     writeConf(testDir, conf);
@@ -884,7 +886,7 @@ public class TestKMS {
 
         // test ACL reloading
         Thread.sleep(10); // to ensure the ACLs file modifiedTime is newer
-        conf.set(KMSACLs.Type.CREATE.getConfigKey(), "foo");
+        conf.set(KMSACLs.Type.CREATE.getAclConfigKey(), "foo");
         writeConf(testDir, conf);
         Thread.sleep(1000);
 
@@ -914,6 +916,92 @@ public class TestKMS {
     });
   }
 
+  @Test
+  public void testKMSBlackList() throws Exception {
+    Configuration conf = new Configuration();
+    conf.set("hadoop.security.authentication", "kerberos");
+    UserGroupInformation.setConfiguration(conf);
+    File testDir = getTestDir();
+    conf = createBaseKMSConf(testDir);
+    conf.set("hadoop.kms.authentication.type", "kerberos");
+    conf.set("hadoop.kms.authentication.kerberos.keytab",
+        keytab.getAbsolutePath());
+    conf.set("hadoop.kms.authentication.kerberos.principal", "HTTP/localhost");
+    conf.set("hadoop.kms.authentication.kerberos.name.rules", "DEFAULT");
+    for (KMSACLs.Type type : KMSACLs.Type.values()) {
+      conf.set(type.getAclConfigKey(), " ");
+    }
+    conf.set(KMSACLs.Type.CREATE.getAclConfigKey(), "client,hdfs,otheradmin");
+    conf.set(KMSACLs.Type.GENERATE_EEK.getAclConfigKey(), "client,hdfs,otheradmin");
+    conf.set(KMSACLs.Type.DECRYPT_EEK.getAclConfigKey(), "client,hdfs,otheradmin");
+    conf.set(KMSACLs.Type.DECRYPT_EEK.getBlacklistConfigKey(), "hdfs,otheradmin");
+
+    writeConf(testDir, conf);
+
+    runServer(null, null, testDir, new KMSCallable() {
+      @Override
+      public Void call() throws Exception {
+        final Configuration conf = new Configuration();
+        conf.setInt(KeyProvider.DEFAULT_BITLENGTH_NAME, 128);
+        final URI uri = createKMSUri(getKMSUrl());
+
+        doAs("client", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            try {
+              KMSClientProvider kp = new KMSClientProvider(uri, conf);
+              KeyProvider.KeyVersion kv = kp.createKey("ck0",
+                  new KeyProvider.Options(conf));
+              EncryptedKeyVersion eek =
+                  kp.generateEncryptedKey("ck0");
+              kp.decryptEncryptedKey(eek);
+              Assert.assertNull(kv.getMaterial());
+            } catch (Exception ex) {
+              Assert.fail(ex.getMessage());
+            }
+            return null;
+          }
+        });
+
+        doAs("hdfs", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            try {
+              KMSClientProvider kp = new KMSClientProvider(uri, conf);
+              KeyProvider.KeyVersion kv = kp.createKey("ck1",
+                  new KeyProvider.Options(conf));
+              EncryptedKeyVersion eek =
+                  kp.generateEncryptedKey("ck1");
+              kp.decryptEncryptedKey(eek);
+              Assert.fail("admin user must not be allowed to decrypt !!");
+            } catch (Exception ex) {
+            }
+            return null;
+          }
+        });
+
+        doAs("otheradmin", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            try {
+              KMSClientProvider kp = new KMSClientProvider(uri, conf);
+              KeyProvider.KeyVersion kv = kp.createKey("ck2",
+                  new KeyProvider.Options(conf));
+              EncryptedKeyVersion eek =
+                  kp.generateEncryptedKey("ck2");
+              kp.decryptEncryptedKey(eek);
+              Assert.fail("admin user must not be allowed to decrypt !!");
+            } catch (Exception ex) {
+            }
+            return null;
+          }
+        });
+
+        return null;
+      }
+    });
+  }
+
   @Test
   public void testServicePrincipalACLs() throws Exception {
     Configuration conf = new Configuration();
@@ -927,9 +1015,9 @@ public class TestKMS {
     conf.set("hadoop.kms.authentication.kerberos.principal", "HTTP/localhost");
     conf.set("hadoop.kms.authentication.kerberos.name.rules", "DEFAULT");
     for (KMSACLs.Type type : KMSACLs.Type.values()) {
-      conf.set(type.getConfigKey(), " ");
+      conf.set(type.getAclConfigKey(), " ");
     }
-    conf.set(KMSACLs.Type.CREATE.getConfigKey(), "client");
+    conf.set(KMSACLs.Type.CREATE.getAclConfigKey(), "client");
 
     writeConf(testDir, conf);
 

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

@@ -37,7 +37,7 @@ public class TestKMSACLs {
   public void testCustom() {
     Configuration conf = new Configuration(false);
     for (KMSACLs.Type type : KMSACLs.Type.values()) {
-      conf.set(type.getConfigKey(), type.toString() + " ");
+      conf.set(type.getAclConfigKey(), type.toString() + " ");
     }
     KMSACLs acls = new KMSACLs(conf);
     for (KMSACLs.Type type : KMSACLs.Type.values()) {