[ALLUXIO-3305] Fix loadMetadata bug related to ACL (#7813)

* add defaultAcl to the information returned by ufs

* change the way we write to ufs acl and default ACL so default acl won't overwrite the acl entries

* fix compilation

* remove an masking exception handler and fixed a NPE

* style fix

* fix immutable list, use a mutable list

* remove comments

* Generate children's access acl based on umask and the parent's default acl.

* fix equals test for CreateDirectoryOptions

* fix tests

* fix checkstyle

* fix setDefaultFacl test according to new default acl inheritance rules.

* remove unnecessary logging

* address comments

* address comments

core/common/src/main/java/alluxio/security/authorization/DefaultAccessControlList.java

@@ -24,7 +24,6 @@
  */
 public class DefaultAccessControlList extends AccessControlList {
   private static final long serialVersionUID = 8649647787531425489L;
-
   public static final DefaultAccessControlList EMPTY_DEFAULT_ACL = new DefaultAccessControlList();
 
   /**
@@ -60,27 +59,46 @@ public DefaultAccessControlList(AccessControlList acl) {
 
   /**
    * create a child file 's accessACL based on the default ACL.
+   * @param umask file's umask
    * @return child file's access ACL
    */
-  public AccessControlList generateChildFileACL() {
+  public AccessControlList generateChildFileACL(Short umask) {
+    Mode defaultMode = new Mode(umask);
     AccessControlList acl = new AccessControlList();
     acl.mOwningUser = mOwningUser;
     acl.mOwningGroup = mOwningGroup;
     acl.mMode = mMode;
     if (mExtendedEntries == null) {
       acl.mExtendedEntries = null;
+      // minimal acl so we use defaultMode to filter user/group/other
+      acl.mMode = Mode.and(new Mode(mMode), defaultMode).toShort();
     } else {
+      // Rules for having extended entries, we need to modify user, mask and others'
+      // permission to be filtered by the defaultMode
       acl.mExtendedEntries = new ExtendedACLEntries(mExtendedEntries);
+
+      // mask is filtered by the group bits from the umask parameter
+      AclActions mask = acl.mExtendedEntries.getMask();
+      AclActions groupAction = new AclActions();
+      groupAction.updateByModeBits(defaultMode.getGroupBits());
+      mask.mask(groupAction);
+      // user is filtered by the user bits from the umask parameter
+      // other is filtered by the other bits from the umask parameter
+      Mode updateMode = new Mode(mMode);
+      updateMode.setOwnerBits(updateMode.getOwnerBits().and(defaultMode.getOwnerBits()));
+      updateMode.setOtherBits(updateMode.getOtherBits().and(defaultMode.getOtherBits()));
+      acl.mMode = updateMode.toShort();
     }
     return acl;
   }
 
   /**
    * Creates a child directory's access ACL and default ACL based on the default ACL.
+   * @param umask child's umask
    * @return child directory's access ACL and default ACL
    */
-  public Pair<AccessControlList, DefaultAccessControlList> generateChildDirACL() {
-    AccessControlList acl = generateChildFileACL();
+  public Pair<AccessControlList, DefaultAccessControlList> generateChildDirACL(Short umask) {
+    AccessControlList acl = generateChildFileACL(umask);
     DefaultAccessControlList dAcl = new DefaultAccessControlList(acl);
     dAcl.setEmpty(false);
     dAcl.mOwningUser = mOwningUser;

core/common/src/main/java/alluxio/security/authorization/Mode.java

@@ -104,6 +104,18 @@ public Mode(Mode mode) {
     set(mode.mOwnerBits, mode.mGroupBits, mode.mOtherBits);
   }
 
+  /**
+   * @param mode1 first mode of the and operation
+   * @param mode2 second mode of the and operation
+   * @return the AND result of the two Modes
+   */
+  public static Mode and(Mode mode1, Mode mode2) {
+    Bits u = mode1.mOwnerBits.and(mode2.mOwnerBits);
+    Bits g = mode1.mGroupBits.and(mode2.mGroupBits);
+    Bits o = mode1.mOtherBits.and(mode2.mOtherBits);
+    return new Mode(u, g, o);
+  }
+
   /**
    * @return the owner {@link Bits}
    */

core/common/src/main/java/alluxio/underfs/BaseUnderFileSystem.java

@@ -15,6 +15,8 @@
 import alluxio.Constants;
 import alluxio.collections.Pair;
 import alluxio.security.authorization.AccessControlList;
+import alluxio.security.authorization.AclEntry;
+import alluxio.security.authorization.DefaultAccessControlList;
 import alluxio.underfs.options.CreateOptions;
 import alluxio.underfs.options.DeleteOptions;
 import alluxio.underfs.options.ListOptions;
@@ -79,24 +81,27 @@ public boolean exists(String path) throws IOException {
   }
 
   @Override
-  public AccessControlList getAcl(String path) throws IOException {
+  public Pair<AccessControlList, DefaultAccessControlList> getAclPair(String path)
+      throws IOException {
     return null;
   }
 
   @Override
-  public void setAcl(String path, AccessControlList acl) throws IOException{
+  public void setAclEntries(String path, List<AclEntry> aclEntries) throws IOException{
     // Noop here by default
   }
 
   @Override
   public String getFingerprint(String path) {
+    // TODO(yuzhu): include default ACL in the fingerprint
     try {
       UfsStatus status = getStatus(path);
-      AccessControlList acl = getAcl(path);
-      if (acl == null || !acl.hasExtended()) {
+      Pair<AccessControlList, DefaultAccessControlList> aclPair = getAclPair(path);
+
+      if (aclPair == null || aclPair.getFirst() == null || !aclPair.getFirst().hasExtended()) {
         return Fingerprint.create(getUnderFSType(), status).serialize();
       } else {
-        return Fingerprint.create(getUnderFSType(), status, acl).serialize();
+        return Fingerprint.create(getUnderFSType(), status, aclPair.getFirst()).serialize();
       }
 
     } catch (Exception e) {

core/common/src/main/java/alluxio/underfs/UnderFileSystem.java

@@ -15,7 +15,10 @@
 import alluxio.Configuration;
 import alluxio.PropertyKey;
 import alluxio.annotation.PublicApi;
+import alluxio.collections.Pair;
 import alluxio.security.authorization.AccessControlList;
+import alluxio.security.authorization.AclEntry;
+import alluxio.security.authorization.DefaultAccessControlList;
 import alluxio.underfs.options.CreateOptions;
 import alluxio.underfs.options.DeleteOptions;
 import alluxio.underfs.options.FileLocationOptions;
@@ -267,13 +270,14 @@ enum UfsMode {
   boolean exists(String path) throws IOException;
 
   /**
-   * Gets the access control list of a file or directory in under file system.
+   * Gets the ACL and the Default ACL of a file or directory in under file system.
    *
    * @param path the path to the file or directory
-   * @return the access control list, or null if ACL is unsupported or disabled
+   * @return the access control list, along with a Default ACL if it is a directory
+   *         return null if ACL is unsupported or disabled
    * @throws IOException if ACL is supported and enabled but cannot be retrieved
    */
-  AccessControlList getAcl(String path) throws IOException;
+  Pair<AccessControlList, DefaultAccessControlList> getAclPair(String path) throws IOException;
 
   /**
    * Gets the block size of a file in under file system, in bytes.
@@ -522,11 +526,12 @@ enum UfsMode {
   /**
    * Sets the access control list of a file or directory in under file system.
    * if the ufs does not support acls, this is a noop.
+   * This will overwrite the ACL and defaultACL in the UFS.
    *
    * @param path the path to the file or directory
-   * @param acl the access control list
+   * @param aclEntries the access control list + default acl represented in a list of acl entries
    */
-  void setAcl(String path, AccessControlList acl) throws IOException;
+  void setAclEntries(String path, List<AclEntry> aclEntries) throws IOException;
 
   /**
    * Changes posix file mode.

core/common/src/main/java/alluxio/underfs/UnderFileSystemWithLogging.java

@@ -13,13 +13,16 @@
 
 import alluxio.AlluxioURI;
 import alluxio.Constants;
+import alluxio.collections.Pair;
 import alluxio.exception.status.UnimplementedException;
 import alluxio.security.authorization.AccessControlList;
 import alluxio.metrics.CommonMetrics;
 import alluxio.metrics.Metric;
 import alluxio.metrics.MetricsSystem;
 import alluxio.metrics.WorkerMetrics;
 import alluxio.security.authentication.AuthenticatedClientUser;
+import alluxio.security.authorization.AclEntry;
+import alluxio.security.authorization.DefaultAccessControlList;
 import alluxio.underfs.options.CreateOptions;
 import alluxio.underfs.options.DeleteOptions;
 import alluxio.underfs.options.FileLocationOptions;
@@ -223,11 +226,12 @@ public String toString() {
   }
 
   @Override
-  public AccessControlList getAcl(String path) throws IOException, UnimplementedException {
-    return call(new UfsCallable<AccessControlList>() {
+  public Pair<AccessControlList, DefaultAccessControlList> getAclPair(String path)
+      throws IOException, UnimplementedException {
+    return call(new UfsCallable<Pair<AccessControlList, DefaultAccessControlList>>() {
       @Override
-      public AccessControlList call() throws IOException {
-        return mUnderFileSystem.getAcl(path);
+      public Pair<AccessControlList, DefaultAccessControlList> call() throws IOException {
+        return mUnderFileSystem.getAclPair(path);
       }
 
       @Override
@@ -567,17 +571,17 @@ public AlluxioURI resolveUri(AlluxioURI ufsBaseUri, String alluxioPath) {
   }
 
   @Override
-  public void setAcl(String path, AccessControlList acl) throws IOException {
+  public void setAclEntries(String path, List<AclEntry> aclEntries) throws IOException {
     call(new UfsCallable<Void>() {
       @Override
       public Void call() throws IOException {
-        mUnderFileSystem.setAcl(path, acl);
+        mUnderFileSystem.setAclEntries(path, aclEntries);
         return null;
       }
 
       @Override
       public String toString() {
-        return String.format("SetAcl: path=%s, ACL=%s", path, acl);
+        return String.format("SetAcl: path=%s, ACLEntries=%s", path, aclEntries);
       }
     });
   }

core/server/master/src/main/java/alluxio/master/file/DefaultFileSystemMaster.java

@@ -98,6 +98,7 @@
 import alluxio.security.authorization.AccessControlList;
 import alluxio.security.authorization.AclEntry;
 import alluxio.security.authorization.AclEntryType;
+import alluxio.security.authorization.DefaultAccessControlList;
 import alluxio.security.authorization.Mode;
 import alluxio.thrift.CommandType;
 import alluxio.thrift.FileSystemCommand;
@@ -2332,7 +2333,16 @@ private void loadFileMetadataInternal(RpcContext rpcContext, LockedInodePath ino
       ufsLength = ufsStatus.getContentLength();
 
       if (isAclEnabled()) {
-        acl = ufs.getAcl(ufsUri.toString());
+        Pair<AccessControlList, DefaultAccessControlList> aclPair
+            = ufs.getAclPair(ufsUri.toString());
+        if (aclPair != null) {
+          acl = aclPair.getFirst();
+          // DefaultACL should be null, because it is a file
+          if (aclPair.getSecond() != null) {
+            LOG.warn("File {} has default ACL in the UFS", inodePath.getUri());
+          }
+        }
+
       }
     }
 
@@ -2405,12 +2415,22 @@ private void loadDirectoryMetadata(RpcContext rpcContext, LockedInodePath inodeP
         .setTtl(options.getTtl()).setTtlAction(options.getTtlAction());
     MountTable.Resolution resolution = mMountTable.resolve(inodePath.getUri());
     UfsStatus ufsStatus = options.getUfsStatus();
-    if (ufsStatus == null) {
-      AlluxioURI ufsUri = resolution.getUri();
-      try (CloseableResource<UnderFileSystem> ufsResource = resolution.acquireUfsResource()) {
-        UnderFileSystem ufs = ufsResource.get();
+
+    AlluxioURI ufsUri = resolution.getUri();
+    AccessControlList acl = null;
+    DefaultAccessControlList defaultAcl = null;
+
+    try (CloseableResource<UnderFileSystem> ufsResource = resolution.acquireUfsResource()) {
+      UnderFileSystem ufs = ufsResource.get();
+      if (ufsStatus == null) {
         ufsStatus = ufs.getDirectoryStatus(ufsUri.toString());
       }
+      Pair<AccessControlList, DefaultAccessControlList> aclPair =
+          ufs.getAclPair(ufsUri.toString());
+      if (aclPair != null) {
+        acl = aclPair.getFirst();
+        defaultAcl = aclPair.getSecond();
+      }
     }
     String ufsOwner = ufsStatus.getOwner();
     String ufsGroup = ufsStatus.getGroup();
@@ -2422,6 +2442,13 @@ private void loadDirectoryMetadata(RpcContext rpcContext, LockedInodePath inodeP
     }
     createDirectoryOptions = createDirectoryOptions.setOwner(ufsOwner).setGroup(ufsGroup)
             .setMode(mode).setUfsStatus(ufsStatus);
+    if (acl != null) {
+      createDirectoryOptions.setAcl(acl.getEntries());
+    }
+
+    if (defaultAcl != null) {
+      createDirectoryOptions.setDefaultAcl(defaultAcl.getEntries());
+    }
     if (lastModifiedTime != null) {
       createDirectoryOptions.setOperationTimeMs(lastModifiedTime);
     }
@@ -2465,7 +2492,8 @@ private void loadMetadataIfNotExist(RpcContext rpcContext, LockedInodePath inode
     if (!inodeExists || loadDirectChildren) {
       try {
         loadMetadataInternal(rpcContext, inodePath, options);
-      } catch (Exception e) {
+      } catch (IOException | InvalidPathException | FileDoesNotExistException | BlockInfoException
+          | FileAlreadyCompletedException | InvalidFileSizeException | AccessControlException e) {
         // NOTE, this may be expected when client tries to get info (e.g. exists()) for a file
         // existing neither in Alluxio nor UFS.
         LOG.debug("Failed to load metadata for path from UFS: {}", inodePath.getUri());
@@ -2725,11 +2753,12 @@ private void setUfsAcl(LockedInodePath inodePath)
             + "UFS: " + ufsUri + ". This has no effect on the underlying object.");
       } else {
         try {
-          ufs.setAcl(ufsUri, inode.getACL());
+          List<AclEntry> entries = new ArrayList<>(inode.getACL().getEntries());
           if (inode.isDirectory()) {
             InodeDirectoryView inodeDirectory = (InodeDirectoryView) inode;
-            ufs.setAcl(ufsUri, inodeDirectory.getDefaultACL());
+            entries.addAll(inodeDirectory.getDefaultACL().getEntries());
           }
+          ufs.setAclEntries(ufsUri, entries);
         } catch (IOException e) {
           throw new AccessControlException("Could not setAcl for UFS file: " + ufsUri);
         }

core/server/master/src/main/java/alluxio/master/file/meta/Inode.java

@@ -402,8 +402,9 @@ public T setMode(short mode) {
   /**
    * @param acl set the default ACL associated with this inode
    * @throws UnsupportedOperationException if the inode is a file
+   * @return the updated object
    */
-  public abstract void setDefaultACL(DefaultAccessControlList acl)
+  public abstract T setDefaultACL(DefaultAccessControlList acl)
       throws UnsupportedOperationException;
 
   /**
@@ -414,6 +415,9 @@ public abstract void setDefaultACL(DefaultAccessControlList acl)
    * @return the updated object
    */
   public T setAcl(List<AclEntry> entries) {
+    if (entries == null || entries.isEmpty()) {
+      return getThis();
+    }
     for (AclEntry entry : entries) {
       if (entry.isDefault()) {
         getDefaultACL().setEntry(entry);

core/server/master/src/main/java/alluxio/master/file/meta/InodeDirectory.java

@@ -216,8 +216,9 @@ public synchronized InodeDirectory setDirectChildrenLoaded(boolean directChildre
   }
 
   @Override
-  public void setDefaultACL(DefaultAccessControlList acl) {
+  public InodeDirectory setDefaultACL(DefaultAccessControlList acl) {
     mDefaultAcl = acl;
+    return getThis();
   }
 
   /**
@@ -336,6 +337,8 @@ public static InodeDirectory create(long id, long parentId, String name,
         .setGroup(options.getGroup())
         .setMode(options.getMode().toShort())
         .setAcl(options.getAcl())
+        // SetAcl call is also setting default AclEntries
+        .setAcl(options.getDefaultAcl())
         .setMountPoint(options.isMountPoint());
   }
 

core/server/master/src/main/java/alluxio/master/file/meta/InodeFile.java

@@ -109,7 +109,8 @@ public DefaultAccessControlList getDefaultACL() throws UnsupportedOperationExcep
   }
 
   @Override
-  public void setDefaultACL(DefaultAccessControlList acl) throws UnsupportedOperationException {
+  public InodeFile setDefaultACL(DefaultAccessControlList acl)
+      throws UnsupportedOperationException {
     throw new UnsupportedOperationException("setDefaultACL: File does not have default ACL");
   }