HADOOP-10650. Add ability to specify a reverse ACL (black list) of users and groups. (Contributed by Benoy Antony)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1618482 13f79535-47bb-0310-9956-ffa450edef68

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

@@ -521,6 +521,9 @@ Release 2.6.0 - UNRELEASED
     HADOOP-10231. Add some components in Native Libraries document (Akira 
     AJISAKA via aw)
 
+    HADOOP-10650. Add ability to specify a reverse ACL (black list) of users
+    and groups. (Benoy Antony via Arpit Agarwal)
+
   OPTIMIZATIONS
 
     HADOOP-10838. Byte array native checksumming. (James Thomas via todd)

hadoop-common-project/hadoop-common/src/main/java/org/apache/hadoop/fs/CommonConfigurationKeys.java

@@ -134,6 +134,9 @@ public class CommonConfigurationKeys extends CommonConfigurationKeysPublic {
   HADOOP_SECURITY_SERVICE_AUTHORIZATION_DEFAULT_ACL = 
       "security.service.authorization.default.acl";
   public static final String 
+  HADOOP_SECURITY_SERVICE_AUTHORIZATION_DEFAULT_BLOCKED_ACL =
+      "security.service.authorization.default.acl.blocked";
+  public static final String
   HADOOP_SECURITY_SERVICE_AUTHORIZATION_REFRESH_POLICY = 
       "security.refresh.policy.protocol.acl";
   public static final String 

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

@@ -43,10 +43,14 @@ import com.google.common.annotations.VisibleForTesting;
 @InterfaceAudience.LimitedPrivate({"HDFS", "MapReduce"})
 @InterfaceStability.Evolving
 public class ServiceAuthorizationManager {
+  static final String BLOCKED = ".blocked";
+
   private static final String HADOOP_POLICY_FILE = "hadoop-policy.xml";
 
-  private volatile Map<Class<?>, AccessControlList> protocolToAcl =
-    new IdentityHashMap<Class<?>, AccessControlList>();
+  // For each class, first ACL in the array specifies the allowed entries
+  // and second ACL specifies blocked entries.
+  private volatile Map<Class<?>, AccessControlList[]> protocolToAcls =
+    new IdentityHashMap<Class<?>, AccessControlList[]>();
   
   /**
    * Configuration key for controlling service-level authorization for Hadoop.
@@ -80,8 +84,8 @@ public class ServiceAuthorizationManager {
                                Configuration conf,
                                InetAddress addr
                                ) throws AuthorizationException {
-    AccessControlList acl = protocolToAcl.get(protocol);
-    if (acl == null) {
+    AccessControlList[] acls = protocolToAcls.get(protocol);
+    if (acls == null) {
       throw new AuthorizationException("Protocol " + protocol + 
                                        " is not known.");
     }
@@ -104,7 +108,7 @@ public class ServiceAuthorizationManager {
       }
     }
     if((clientPrincipal != null && !clientPrincipal.equals(user.getUserName())) || 
-        !acl.isUserAllowed(user)) {
+       acls.length != 2  || !acls[0].isUserAllowed(user) || acls[1].isUserAllowed(user)) {
       AUDITLOG.warn(AUTHZ_FAILED_FOR + user + " for protocol=" + protocol
           + ", expected client Kerberos principal is " + clientPrincipal);
       throw new AuthorizationException("User " + user + 
@@ -129,13 +133,16 @@ public class ServiceAuthorizationManager {
   @Private
   public void refreshWithLoadedConfiguration(Configuration conf,
       PolicyProvider provider) {
-    final Map<Class<?>, AccessControlList> newAcls =
-        new IdentityHashMap<Class<?>, AccessControlList>();
+    final Map<Class<?>, AccessControlList[]> newAcls =
+      new IdentityHashMap<Class<?>, AccessControlList[]>();
     
     String defaultAcl = conf.get(
         CommonConfigurationKeys.HADOOP_SECURITY_SERVICE_AUTHORIZATION_DEFAULT_ACL,
         AccessControlList.WILDCARD_ACL_VALUE);
 
+    String defaultBlockedAcl = conf.get(
+      CommonConfigurationKeys.HADOOP_SECURITY_SERVICE_AUTHORIZATION_DEFAULT_BLOCKED_ACL, "");
+
     // Parse the config file
     Service[] services = provider.getServices();
     if (services != null) {
@@ -145,21 +152,30 @@ public class ServiceAuthorizationManager {
                 conf.get(service.getServiceKey(),
                     defaultAcl)
             );
-        newAcls.put(service.getProtocol(), acl);
+        AccessControlList blockedAcl =
+           new AccessControlList(
+           conf.get(service.getServiceKey() + BLOCKED,
+           defaultBlockedAcl));
+        newAcls.put(service.getProtocol(), new AccessControlList[] {acl, blockedAcl});
       }
     }
 
     // Flip to the newly parsed permissions
-    protocolToAcl = newAcls;
+    protocolToAcls = newAcls;
   }
 
   @VisibleForTesting
   public Set<Class<?>> getProtocolsWithAcls() {
-    return protocolToAcl.keySet();
+    return protocolToAcls.keySet();
   }
 
   @VisibleForTesting
   public AccessControlList getProtocolsAcls(Class<?> className) {
-    return protocolToAcl.get(className);
+    return protocolToAcls.get(className)[0];
+  }
+
+  @VisibleForTesting
+  public AccessControlList getProtocolsBlockedAcls(Class<?> className) {
+    return protocolToAcls.get(className)[1];
   }
 }

hadoop-common-project/hadoop-common/src/site/apt/ServiceLevelAuth.apt.vm

@@ -110,6 +110,27 @@ security.ha.service.protocol.acl      | ACL for HAService protocol used by HAAdm
    <<<security.service.authorization.default.acl>>> is applied. If 
    <<<security.service.authorization.default.acl>>> is not defined, <<<*>>>  is applied.
 
+ ** Blocked Access Control Lists
+
+   In some cases, it is required to specify blocked access control list for a service. This specifies
+   the list of users and groups who are not authorized to access the service. The format of
+   the blocked access control list is same as that of access control list. The blocked access
+   control list can be specified via <<<${HADOOP_CONF_DIR}/hadoop-policy.xml>>>. The property name
+   is derived by suffixing with ".blocked".
+
+   Example: The property name of blocked access control list for <<<security.client.protocol.acl>>
+   will be <<<security.client.protocol.acl.blocked>>>
+
+   For a service, it is possible to specify both an access control list and a blocked control
+   list. A user is authorized to access the service if the user is in the access control and not in
+   the blocked access control list.
+
+   If blocked access control list is not defined for a service, the value of
+   <<<security.service.authorization.default.acl.blocked>>> is applied. If
+   <<<security.service.authorization.default.acl.blocked>>> is not defined,
+   empty blocked access control list is applied.
+
+
 ** Refreshing Service Level Authorization Configuration
 
    The service-level authorization configuration for the NameNode and

hadoop-common-project/hadoop-common/src/test/java/org/apache/hadoop/security/authorize/TestServiceAuthorization.java

@@ -18,16 +18,22 @@
 package org.apache.hadoop.security.authorize;
 
 import static org.junit.Assert.assertEquals;
+import static org.junit.Assert.fail;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeys;
 import org.apache.hadoop.ipc.TestRPC.TestProtocol;
+import org.apache.hadoop.security.UserGroupInformation;
 import org.junit.Test;
 
 public class TestServiceAuthorization {
 
   private static final String ACL_CONFIG = "test.protocol.acl";
   private static final String ACL_CONFIG1 = "test.protocol1.acl";
+  private static final String ADDRESS =  "0.0.0.0";
 
   public interface TestProtocol1 extends TestProtocol {};
 
@@ -64,4 +70,115 @@ public class TestServiceAuthorization {
     acl = serviceAuthorizationManager.getProtocolsAcls(TestProtocol1.class);
     assertEquals("user2 group2", acl.getAclString());
   }
+
+  @Test
+  public void testBlockedAcl() throws UnknownHostException {
+    UserGroupInformation drwho =
+        UserGroupInformation.createUserForTesting("drwho@EXAMPLE.COM",
+            new String[] { "group1", "group2" });
+
+    ServiceAuthorizationManager serviceAuthorizationManager =
+        new ServiceAuthorizationManager();
+    Configuration conf = new Configuration ();
+
+    //test without setting a blocked acl
+    conf.set(ACL_CONFIG, "user1 group1");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+    } catch (AuthorizationException e) {
+      fail();
+    }
+    //now set a blocked acl with another user and another group
+    conf.set(ACL_CONFIG + ServiceAuthorizationManager.BLOCKED, "drwho2 group3");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+    } catch (AuthorizationException e) {
+      fail();
+    }
+    //now set a blocked acl with the user and another group
+    conf.set(ACL_CONFIG + ServiceAuthorizationManager.BLOCKED, "drwho group3");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+      fail();
+    } catch (AuthorizationException e) {
+
+    }
+    //now set a blocked acl with another user and another group
+    conf.set(ACL_CONFIG + ServiceAuthorizationManager.BLOCKED, "drwho2 group3");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+    } catch (AuthorizationException e) {
+      fail();
+    }
+    //now set a blocked acl with another user and group that the user belongs to
+    conf.set(ACL_CONFIG + ServiceAuthorizationManager.BLOCKED, "drwho2 group2");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+      fail();
+    } catch (AuthorizationException e) {
+      //expects Exception
+    }
+    //reset blocked acl so that there is no blocked ACL
+    conf.set(ACL_CONFIG + ServiceAuthorizationManager.BLOCKED, "");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+    } catch (AuthorizationException e) {
+      fail();
+    }
+  }
+
+  @Test
+  public void testDefaultBlockedAcl() throws UnknownHostException {
+    UserGroupInformation drwho =
+        UserGroupInformation.createUserForTesting("drwho@EXAMPLE.COM",
+            new String[] { "group1", "group2" });
+
+    ServiceAuthorizationManager serviceAuthorizationManager =
+        new ServiceAuthorizationManager();
+    Configuration conf = new Configuration ();
+
+    //test without setting a default blocked acl
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol1.class, conf,
+          InetAddress.getByName(ADDRESS));
+    } catch (AuthorizationException e) {
+      fail();
+    }
+
+    //set a restrictive default blocked acl and an non-restricting blocked acl for TestProtocol
+    conf.set(
+        CommonConfigurationKeys.HADOOP_SECURITY_SERVICE_AUTHORIZATION_DEFAULT_BLOCKED_ACL,
+        "user2 group2");
+    conf.set(ACL_CONFIG + ServiceAuthorizationManager.BLOCKED, "user2");
+    serviceAuthorizationManager.refresh(conf, new TestPolicyProvider());
+    //drwho is authorized to access TestProtocol
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol.class, conf,
+          InetAddress.getByName(ADDRESS));
+    } catch (AuthorizationException e) {
+      fail();
+    }
+    //drwho is not authorized to access TestProtocol1 because it uses the default blocked acl.
+    try {
+      serviceAuthorizationManager.authorize(drwho, TestProtocol1.class, conf,
+          InetAddress.getByName(ADDRESS));
+      fail();
+    } catch (AuthorizationException e) {
+      //expects Exception
+    }
+  }
+
 }