HDFS-7582. Enforce maximum number of ACL entries separately per access and default. (Contributed by Vinayakumar B)

(cherry picked from commit 29cf887b226f4ab3c336a6e681db5e8e70699d66)

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

@@ -719,6 +719,9 @@ Release 2.8.0 - UNRELEASED
     HDFS-8344. NameNode doesn't recover lease for files with missing blocks
     (raviprak)
 
+    HDFS-7582. Enforce maximum number of ACL entries separately per access
+    and default. (vinayakumarb)
+
 Release 2.7.2 - UNRELEASED
 
   INCOMPATIBLE CHANGES

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

@@ -271,10 +271,6 @@ final class AclTransformation {
    */
   private static List<AclEntry> buildAndValidateAcl(
       ArrayList<AclEntry> aclBuilder) throws AclException {
-    if (aclBuilder.size() > MAX_ENTRIES) {
-      throw new AclException("Invalid ACL: ACL has " + aclBuilder.size() +
-        " entries, which exceeds maximum of " + MAX_ENTRIES + ".");
-    }
     aclBuilder.trimToSize();
     Collections.sort(aclBuilder, ACL_ENTRY_COMPARATOR);
     // Full iteration to check for duplicates and invalid named entries.
@@ -292,9 +288,12 @@ final class AclTransformation {
       }
       prevEntry = entry;
     }
+
+    ScopedAclEntries scopedEntries = new ScopedAclEntries(aclBuilder);
+    checkMaxEntries(scopedEntries);
+
     // Search for the required base access entries.  If there is a default ACL,
     // then do the same check on the default entries.
-    ScopedAclEntries scopedEntries = new ScopedAclEntries(aclBuilder);
     for (AclEntryType type: EnumSet.of(USER, GROUP, OTHER)) {
       AclEntry accessEntryKey = new AclEntry.Builder().setScope(ACCESS)
         .setType(type).build();
@@ -316,6 +315,22 @@ final class AclTransformation {
     return Collections.unmodifiableList(aclBuilder);
   }
 
+  // Check the max entries separately on access and default entries
+  // HDFS-7582
+  private static void checkMaxEntries(ScopedAclEntries scopedEntries)
+      throws AclException {
+    List<AclEntry> accessEntries = scopedEntries.getAccessEntries();
+    List<AclEntry> defaultEntries = scopedEntries.getDefaultEntries();
+    if (accessEntries.size() > MAX_ENTRIES) {
+      throw new AclException("Invalid ACL: ACL has " + accessEntries.size()
+          + " access entries, which exceeds maximum of " + MAX_ENTRIES + ".");
+    }
+    if (defaultEntries.size() > MAX_ENTRIES) {
+      throw new AclException("Invalid ACL: ACL has " + defaultEntries.size()
+          + " default entries, which exceeds maximum of " + MAX_ENTRIES + ".");
+    }
+  }
+
   /**
    * Calculates mask entries required for the ACL.  Mask calculation is performed
    * separately for each scope: access and default.  This method is responsible
@@ -444,11 +459,8 @@ final class AclTransformation {
      * @throws AclException if validation fails
      */
     public ValidatedAclSpec(List<AclEntry> aclSpec) throws AclException {
-      if (aclSpec.size() > MAX_ENTRIES) {
-        throw new AclException("Invalid ACL: ACL spec has " + aclSpec.size() +
-          " entries, which exceeds maximum of " + MAX_ENTRIES + ".");
-      }
       Collections.sort(aclSpec, ACL_ENTRY_COMPARATOR);
+      checkMaxEntries(new ScopedAclEntries(aclSpec));
       this.aclSpec = aclSpec;
     }
 

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

@@ -31,11 +31,8 @@ import com.google.common.collect.Lists;
 import org.junit.Test;
 
 import org.apache.hadoop.fs.permission.AclEntry;
-import org.apache.hadoop.fs.permission.AclEntryScope;
-import org.apache.hadoop.fs.permission.AclEntryType;
 import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.hdfs.protocol.AclException;
-import org.apache.hadoop.hdfs.server.namenode.AclTransformation;
 
 /**
  * Tests operations that modify ACLs.  All tests in this suite have been
@@ -45,10 +42,13 @@ import org.apache.hadoop.hdfs.server.namenode.AclTransformation;
 public class TestAclTransformation {
 
   private static final List<AclEntry> ACL_SPEC_TOO_LARGE;
+  private static final List<AclEntry> ACL_SPEC_DEFAULT_TOO_LARGE;
   static {
     ACL_SPEC_TOO_LARGE = Lists.newArrayListWithCapacity(33);
+    ACL_SPEC_DEFAULT_TOO_LARGE = Lists.newArrayListWithCapacity(33);
     for (int i = 0; i < 33; ++i) {
       ACL_SPEC_TOO_LARGE.add(aclEntry(ACCESS, USER, "user" + i, ALL));
+      ACL_SPEC_DEFAULT_TOO_LARGE.add(aclEntry(DEFAULT, USER, "user" + i, ALL));
     }
   }
 
@@ -351,6 +351,17 @@ public class TestAclTransformation {
     filterAclEntriesByAclSpec(existing, ACL_SPEC_TOO_LARGE);
   }
 
+  @Test(expected = AclException.class)
+  public void testFilterDefaultAclEntriesByAclSpecInputTooLarge()
+      throws AclException {
+    List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()
+        .add(aclEntry(DEFAULT, USER, ALL))
+        .add(aclEntry(DEFAULT, GROUP, READ))
+        .add(aclEntry(DEFAULT, OTHER, NONE))
+        .build();
+    filterAclEntriesByAclSpec(existing, ACL_SPEC_DEFAULT_TOO_LARGE);
+  }
+
   @Test
   public void testFilterDefaultAclEntries() throws AclException {
     List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()
@@ -719,6 +730,16 @@ public class TestAclTransformation {
     mergeAclEntries(existing, ACL_SPEC_TOO_LARGE);
   }
 
+  @Test(expected=AclException.class)
+  public void testMergeAclDefaultEntriesInputTooLarge() throws AclException {
+    List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()
+      .add(aclEntry(DEFAULT, USER, ALL))
+      .add(aclEntry(DEFAULT, GROUP, READ))
+      .add(aclEntry(DEFAULT, OTHER, NONE))
+      .build();
+    mergeAclEntries(existing, ACL_SPEC_DEFAULT_TOO_LARGE);
+  }
+
   @Test(expected=AclException.class)
   public void testMergeAclEntriesResultTooLarge() throws AclException {
     ImmutableList.Builder<AclEntry> aclBuilder =
@@ -737,6 +758,24 @@ public class TestAclTransformation {
     mergeAclEntries(existing, aclSpec);
   }
 
+  @Test(expected = AclException.class)
+  public void testMergeAclDefaultEntriesResultTooLarge() throws AclException {
+    ImmutableList.Builder<AclEntry> aclBuilder =
+        new ImmutableList.Builder<AclEntry>()
+        .add(aclEntry(DEFAULT, USER, ALL));
+    for (int i = 1; i <= 28; ++i) {
+      aclBuilder.add(aclEntry(DEFAULT, USER, "user" + i, READ));
+    }
+    aclBuilder
+    .add(aclEntry(DEFAULT, GROUP, READ))
+    .add(aclEntry(DEFAULT, MASK, READ))
+    .add(aclEntry(DEFAULT, OTHER, NONE));
+    List<AclEntry> existing = aclBuilder.build();
+    List<AclEntry> aclSpec = Lists.newArrayList(
+         aclEntry(DEFAULT, USER, "bruce", READ));
+    mergeAclEntries(existing, aclSpec);
+  }
+
   @Test(expected=AclException.class)
   public void testMergeAclEntriesDuplicateEntries() throws AclException {
     List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()
@@ -1091,6 +1130,16 @@ public class TestAclTransformation {
     replaceAclEntries(existing, ACL_SPEC_TOO_LARGE);
   }
 
+  @Test(expected=AclException.class)
+  public void testReplaceAclDefaultEntriesInputTooLarge() throws AclException {
+    List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()
+      .add(aclEntry(DEFAULT, USER, ALL))
+      .add(aclEntry(DEFAULT, GROUP, READ))
+      .add(aclEntry(DEFAULT, OTHER, NONE))
+      .build();
+    replaceAclEntries(existing, ACL_SPEC_DEFAULT_TOO_LARGE);
+  }
+
   @Test(expected=AclException.class)
   public void testReplaceAclEntriesResultTooLarge() throws AclException {
     List<AclEntry> existing = new ImmutableList.Builder<AclEntry>()