HDFS 8748 rejected patch

HDFS-8748.patch

diff --git hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java
index e6570f5..0144ac4 100644
--- hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java
+++ hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java
@@ -20,6 +20,7 @@
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.HashSet;
+import java.util.Iterator;
 import java.util.Set;
 import java.util.Stack;
 
@@ -356,8 +357,10 @@ private void checkAccessAcl(INodeAttributes inode, String path,
       foundMatch = true;
     }
 
+    FsAction groupUnion = FsAction.NONE;
     // Check named user and group entries if user was not denied by owner entry.
     if (!foundMatch) {
+      // Bits for the union of all applicable groups
       for (int pos = 0, entry; pos < aclFeature.getEntriesSize(); pos++) {
         entry = aclFeature.getEntryAt(pos);
         if (AclEntryStatusFormat.getScope(entry) == AclEntryScope.DEFAULT) {
@@ -378,15 +381,17 @@ private void checkAccessAcl(INodeAttributes inode, String path,
             break;
           }
         } else if (type == AclEntryType.GROUP) {
-          // Use group entry (unnamed or named) with mask from permission bits
-          // applied if user is a member and entry grants access.  If user is a
-          // member of multiple groups that have entries that grant access, then
-          // it doesn't matter which is chosen, so exit early after first match.
+          // Calculate a unioned effective group permission entry as we go
+          // through each group entry (unnamed or named) with mask from permission bits.
+          // If at any point this unioned effective group permission entry
+          // grants access, it can't be taken away by subsequent group entries,
+          // so bail out early.
           String group = name == null ? inode.getGroupName() : name;
           if (getGroups().contains(group)) {
             FsAction masked = AclEntryStatusFormat.getPermission(entry).and(
                 mode.getGroupAction());
-            if (masked.implies(access)) {
+            groupUnion = groupUnion.or(masked);
+            if (groupUnion.implies(access)) {
               return;
             }
             foundMatch = true;
@@ -395,9 +400,17 @@ private void checkAccessAcl(INodeAttributes inode, String path,
       }
     }
 
-    // Use other entry if user was not denied by an earlier match.
-    if (!foundMatch && mode.getOtherAction().implies(access)) {
-      return;
+    // if owning user and named users did not determine access
+    // and at no point did the unioned group entry short circuit to allow
+    // access,
+    // then we'll check whether the groupUnion or other grants access
+    if (!foundMatch) {
+      if (groupUnion.implies(access)) {
+        return;
+      }
+      if (mode.getOtherAction().implies(access)) {
+        return;
+      }
     }
 
     throw new AccessControlException(
diff --git hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSAclBaseTest.java hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSAclBaseTest.java
index 002f7c0..f7a7e81 100644
--- hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSAclBaseTest.java
+++ hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSAclBaseTest.java
@@ -1294,6 +1294,34 @@ public void testGetAclStatusRequiresTraverseOrSuper() throws Exception {
   }
 
   @Test
+  public void testAclUnionsGroupPermissions() throws Exception {
+    Path path = new Path("/path");
+    fs.mkdirs(path);
+    fs.setOwner(path, BRUCE.getShortUserName(), "groupA");
+    fsAsBruce.setAcl(path, Lists.newArrayList(
+      aclEntry(ACCESS, USER, NONE),
+      aclEntry(ACCESS, GROUP, NONE),
+      aclEntry(ACCESS, OTHER, NONE)));
+    try {
+      fsAsBob.access(path, READ_EXECUTE);
+      fail("Access call should have failed");
+    } catch (AccessControlException e) {
+      // expected
+    }
+    fsAsBruce.modifyAclEntries(path, Lists.newArrayList(
+      aclEntry(ACCESS, GROUP, "groupY", READ)));
+    try {
+      fsAsBob.access(path, READ_EXECUTE);
+      fail("Access call should have failed");
+    } catch (AccessControlException e) {
+      // expected
+    }
+    fsAsBruce.modifyAclEntries(path, Lists.newArrayList(
+      aclEntry(ACCESS, GROUP, "groupZ", EXECUTE)));
+    fsAsBob.access(path, READ_EXECUTE);
+  }
+
+  @Test
   public void testAccess() throws IOException, InterruptedException {
     Path p1 = new Path("/p1");
     fs.mkdirs(p1);