HDFS-5737. Replacing only the default ACL can fail to copy unspecified base entries from the access ACL. Contributed by Chris Nauroth.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-4685@1556652 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES-HDFS-4685.txt

@@ -31,3 +31,6 @@ HDFS-4685 (Unreleased)
   OPTIMIZATIONS
 
   BUG FIXES
+
+    HDFS-5737. Replacing only the default ACL can fail to copy unspecified base
+    entries from the access ACL. (cnauroth)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/AclTransformation.java

@@ -62,6 +62,7 @@ import org.apache.hadoop.hdfs.protocol.AclException;
 @InterfaceAudience.Private
 final class AclTransformation {
   private static final int MAX_ENTRIES = 32;
+  private static final int PIVOT_NOT_FOUND = -1;
 
   /**
    * Filters (discards) any existing ACL entries that have the same scope, type
@@ -276,7 +277,7 @@ final class AclTransformation {
     }
     aclBuilder.trimToSize();
     Collections.sort(aclBuilder, ACL_ENTRY_COMPARATOR);
-    AclEntry userEntry = null, groupEntry = null, otherEntry = null;
+    // Full iteration to check for duplicates and invalid named entries.
     AclEntry prevEntry = null;
     for (AclEntry entry: aclBuilder) {
       if (prevEntry != null &&
@@ -289,22 +290,28 @@ final class AclTransformation {
         throw new AclException(
           "Invalid ACL: this entry type must not have a name: " + entry + ".");
       }
-      if (entry.getScope() == ACCESS) {
-        if (entry.getType() == USER && entry.getName() == null) {
-          userEntry = entry;
-        }
-        if (entry.getType() == GROUP && entry.getName() == null) {
-          groupEntry = entry;
-        }
-        if (entry.getType() == OTHER && entry.getName() == null) {
-          otherEntry = entry;
-        }
-      }
       prevEntry = entry;
     }
-    if (userEntry == null || groupEntry == null || otherEntry == null) {
-      throw new AclException(
-        "Invalid ACL: the user, group and other entries are required.");
+    // Search for the required base access entries.  If there is a default ACL,
+    // then do the same check on the default entries.
+    int pivot = calculatePivotOnDefaultEntries(aclBuilder);
+    for (AclEntryType type: EnumSet.of(USER, GROUP, OTHER)) {
+      AclEntry accessEntryKey = new AclEntry.Builder().setScope(ACCESS)
+        .setType(type).build();
+      if (Collections.binarySearch(aclBuilder, accessEntryKey,
+          ACL_ENTRY_COMPARATOR) < 0) {
+        throw new AclException(
+          "Invalid ACL: the user, group and other entries are required.");
+      }
+      if (pivot != PIVOT_NOT_FOUND) {
+        AclEntry defaultEntryKey = new AclEntry.Builder().setScope(DEFAULT)
+          .setType(type).build();
+        if (Collections.binarySearch(aclBuilder, defaultEntryKey,
+            ACL_ENTRY_COMPARATOR) < 0) {
+          throw new AclException(
+            "Invalid default ACL: the user, group and other entries are required.");
+        }
+      }
     }
     return Collections.unmodifiableList(aclBuilder);
   }
@@ -372,6 +379,23 @@ final class AclTransformation {
     }
   }
 
+  /**
+   * Returns the pivot point in the list between the access entries and the
+   * default entries.  This is the index of the first element in the list that is
+   * a default entry.
+   *
+   * @param aclBuilder ArrayList<AclEntry> containing entries to build
+   * @return int pivot point, or -1 if list contains no default entries
+   */
+  private static int calculatePivotOnDefaultEntries(List<AclEntry> aclBuilder) {
+    for (int i = 0; i < aclBuilder.size(); ++i) {
+      if (aclBuilder.get(i).getScope() == DEFAULT) {
+        return i;
+      }
+    }
+    return PIVOT_NOT_FOUND;
+  }
+
   /**
    * Adds unspecified default entries by copying permissions from the
    * corresponding access entries.
@@ -379,14 +403,9 @@ final class AclTransformation {
    * @param aclBuilder ArrayList<AclEntry> containing entries to build
    */
   private static void copyDefaultsIfNeeded(List<AclEntry> aclBuilder) {
-    int pivot = -1;
-    for (int i = 0; i < aclBuilder.size(); ++i) {
-      if (aclBuilder.get(i).getScope() == DEFAULT) {
-        pivot = i;
-        break;
-      }
-    }
-    if (pivot > -1) {
+    Collections.sort(aclBuilder, ACL_ENTRY_COMPARATOR);
+    int pivot = calculatePivotOnDefaultEntries(aclBuilder);
+    if (pivot != PIVOT_NOT_FOUND) {
       List<AclEntry> accessEntries = aclBuilder.subList(0, pivot);
       List<AclEntry> defaultEntries = aclBuilder.subList(pivot,
         aclBuilder.size());

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestAclTransformation.java

@@ -1058,6 +1058,28 @@ public class TestAclTransformation {
     assertEquals(expected, replaceAclEntries(existing, aclSpec));
   }
 
+  @Test
+  public void testReplaceAclEntriesOnlyDefaults() throws AclException {
+    List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()
+      .add(aclEntry(ACCESS, USER, ALL))
+      .add(aclEntry(ACCESS, GROUP, READ))
+      .add(aclEntry(ACCESS, OTHER, NONE))
+      .build();
+    List<AclEntry> aclSpec = Lists.newArrayList(
+      aclEntry(DEFAULT, USER, "bruce", READ));
+    List<AclEntry> expected = new ImmutableList.Builder<AclEntry>()
+      .add(aclEntry(ACCESS, USER, ALL))
+      .add(aclEntry(ACCESS, GROUP, READ))
+      .add(aclEntry(ACCESS, OTHER, NONE))
+      .add(aclEntry(DEFAULT, USER, ALL))
+      .add(aclEntry(DEFAULT, USER, "bruce", READ))
+      .add(aclEntry(DEFAULT, GROUP, READ))
+      .add(aclEntry(DEFAULT, MASK, READ))
+      .add(aclEntry(DEFAULT, OTHER, NONE))
+      .build();
+    assertEquals(expected, replaceAclEntries(existing, aclSpec));
+  }
+
   @Test(expected=AclException.class)
   public void testReplaceAclEntriesInputTooLarge() throws AclException {
     List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()