HADOOP-11322. key based ACL check in KMS always check KeyOpType.MANAGEMENT even actual KeyOpType is not MANAGEMENT. (Dian Fu via yliu)

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

@@ -462,6 +462,9 @@ Release 2.7.0 - UNRELEASED
     HADOOP-11201. Hadoop Archives should support globs resolving to files.
     (Gera Shegalov via cnauroth)
 
+    HADOOP-11322. key based ACL check in KMS always check KeyOpType.MANAGEMENT
+    even actual KeyOpType is not MANAGEMENT. (Dian Fu via yliu)
+
 Release 2.6.0 - 2014-11-18
 
   INCOMPATIBLE CHANGES

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

@@ -132,7 +132,7 @@ public class KeyAuthorizationKeyProvider extends KeyProviderCryptoExtension {
       KeyOpType opType) throws AuthorizationException {
     Preconditions.checkNotNull(aclName, "Key ACL name cannot be null");
     Preconditions.checkNotNull(ugi, "UserGroupInformation cannot be null");
-    if (acls.isACLPresent(aclName, KeyOpType.MANAGEMENT) &&
+    if (acls.isACLPresent(aclName, opType) &&
         (acls.hasAccessToKey(aclName, ugi, opType)
             || acls.hasAccessToKey(aclName, ugi, KeyOpType.ALL))) {
       return;

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

@@ -33,11 +33,9 @@ import org.apache.hadoop.minikdc.MiniKdc;
 import org.apache.hadoop.security.Credentials;
 import org.apache.hadoop.security.SecurityUtil;
 import org.apache.hadoop.security.UserGroupInformation;
-import org.apache.hadoop.security.authentication.client.AuthenticatedURL;
 import org.apache.hadoop.security.authorize.AuthorizationException;
 import org.apache.hadoop.security.ssl.KeyStoreTestUtil;
 import org.apache.hadoop.security.token.Token;
-import org.apache.hadoop.security.token.delegation.web.DelegationTokenAuthenticatedURL;
 import org.junit.AfterClass;
 import org.junit.Assert;
 import org.junit.Before;
@@ -53,8 +51,6 @@ import java.io.File;
 import java.io.FileWriter;
 import java.io.IOException;
 import java.io.Writer;
-import java.lang.reflect.Field;
-import java.lang.reflect.Method;
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.ServerSocket;
@@ -798,6 +794,40 @@ public class TestKMS {
         return null;
       }
     });
+
+    conf.set(KMSConfiguration.DEFAULT_KEY_ACL_PREFIX + "MANAGEMENT", "");
+    conf.set(KMSConfiguration.DEFAULT_KEY_ACL_PREFIX + "GENERATE_EEK", "*");
+    writeConf(testDir, conf);
+
+    runServer(null, null, testDir, new KMSCallable<Void>() {
+
+      @Override
+      public Void call() throws Exception {
+        final Configuration conf = new Configuration();
+        conf.setInt(KeyProvider.DEFAULT_BITLENGTH_NAME, 128);
+        final URI uri = createKMSUri(getKMSUrl());
+
+        doAs("GENERATE_EEK", new PrivilegedExceptionAction<Void>() {
+          @Override
+          public Void run() throws Exception {
+            KeyProvider kp = new KMSClientProvider(uri, conf);
+            try {
+              KeyProviderCryptoExtension kpce =
+                  KeyProviderCryptoExtension.createKeyProviderCryptoExtension(kp);
+              try {
+                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;
+          }
+        });
+        return null;
+      }
+    });
   }
 
   @Test