HDDS-1611. Evaluate ACL on volume bucket key and prefix to authorize access. Contributed by Ajay Kumar.

hadoop-hdds/common/src/main/java/org/apache/hadoop/ozone/OzoneConfigKeys.java

@@ -120,6 +120,10 @@ public final class OzoneConfigKeys {
    * */
   public static final String OZONE_ADMINISTRATORS =
       "ozone.administrators";
+  /**
+   * Used only for testing purpose. Results in making every user an admin.
+   * */
+  public static final String OZONE_ADMINISTRATORS_WILDCARD = "*";
 
   public static final String OZONE_CLIENT_PROTOCOL =
       "ozone.client.protocol";
@@ -390,6 +394,8 @@ public final class OzoneConfigKeys {
       "ozone.acl.authorizer.class";
   public static final String OZONE_ACL_AUTHORIZER_CLASS_DEFAULT =
       "org.apache.hadoop.ozone.security.acl.OzoneAccessAuthorizer";
+  public static final String OZONE_ACL_AUTHORIZER_CLASS_NATIVE =
+      "org.apache.hadoop.ozone.security.acl.OzoneNativeAuthorizer";
   public static final String OZONE_ACL_ENABLED =
       "ozone.acl.enabled";
   public static final boolean OZONE_ACL_ENABLED_DEFAULT =

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/diskbalancer/TestDiskBalancer.java

@@ -219,7 +219,6 @@ public void testDiskBalancerWithFederatedCluster() throws Exception {
     } finally {
       cluster.shutdown();
     }
-
   }
 
   @Test

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rpc/RpcClient.java

@@ -892,6 +892,7 @@ public OmMultipartUploadCompleteInfo completeMultipartUpload(
         .setBucketName(bucketName)
         .setKeyName(keyName)
         .setMultipartUploadID(uploadID)
+        .setAcls(getAclList())
         .build();
 
     OmMultipartUploadList omMultipartUploadList = new OmMultipartUploadList(

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/OzoneAcl.java

@@ -20,8 +20,8 @@
 package org.apache.hadoop.ozone;
 
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
+import com.google.protobuf.ByteString;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo;
-import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclRights;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclType;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
@@ -30,6 +30,7 @@
 import java.util.BitSet;
 import java.util.List;
 import java.util.Objects;
+import java.util.stream.Collectors;
 
 /**
  * OzoneACL classes define bucket ACLs used in OZONE.
@@ -46,6 +47,7 @@ public class OzoneAcl {
   private ACLIdentityType type;
   private String name;
   private BitSet aclBitSet;
+  private static final List<ACLType> EMPTY_LIST = new ArrayList<>(0);
   public static final BitSet ZERO_BITSET = new BitSet(0);
 
   /**
@@ -66,8 +68,16 @@ public OzoneAcl(ACLIdentityType type, String name, ACLType acl) {
     this.aclBitSet = new BitSet(ACLType.getNoOfAcls());
     aclBitSet.set(acl.ordinal(), true);
     this.type = type;
-    if (type == ACLIdentityType.WORLD && name.length() != 0) {
-      throw new IllegalArgumentException("Unexpected name part in world type");
+    if (type == ACLIdentityType.WORLD || type == ACLIdentityType.ANONYMOUS) {
+      if (!name.equals(ACLIdentityType.WORLD.name()) &&
+          !name.equals(ACLIdentityType.ANONYMOUS.name()) &&
+          name.length() != 0) {
+        throw new IllegalArgumentException("Unexpected name:{" + name +
+            "} for type WORLD, ANONYMOUS. It should be WORLD & " +
+            "ANONYMOUS respectively.");
+      }
+      // For type WORLD and ANONYMOUS we allow only one acl to be set.
+      this.name = type.name();
     }
     if (((type == ACLIdentityType.USER) || (type == ACLIdentityType.GROUP))
         && (name.length() == 0)) {
@@ -91,14 +101,20 @@ public OzoneAcl(ACLIdentityType type, String name, BitSet acls) {
           "size. bitset size:" + acls.cardinality() + ", bitset:"
           + acls.toString());
     }
-
     this.aclBitSet = (BitSet) acls.clone();
-    acls.stream().forEach(a -> aclBitSet.set(a));
 
     this.name = name;
     this.type = type;
-    if (type == ACLIdentityType.WORLD && name.length() != 0) {
-      throw new IllegalArgumentException("Unexpected name part in world type");
+    if (type == ACLIdentityType.WORLD || type == ACLIdentityType.ANONYMOUS) {
+      if (!name.equals(ACLIdentityType.WORLD.name()) &&
+          !name.equals(ACLIdentityType.ANONYMOUS.name()) &&
+          name.length() != 0) {
+        throw new IllegalArgumentException("Unexpected name:{" + name +
+            "} for type WORLD, ANONYMOUS. It should be WORLD & " +
+            "ANONYMOUS respectively.");
+      }
+      // For type WORLD and ANONYMOUS we allow only one acl to be set.
+      this.name = type.name();
     }
     if (((type == ACLIdentityType.USER) || (type == ACLIdentityType.GROUP))
         && (name.length() == 0)) {
@@ -161,17 +177,13 @@ public static List<OzoneAcl> parseAcls(String acls)
   public static OzoneAclInfo toProtobuf(OzoneAcl acl) {
     OzoneAclInfo.Builder builder = OzoneAclInfo.newBuilder()
         .setName(acl.getName())
-        .setType(OzoneAclType.valueOf(acl.getType().name()));
-    acl.getAclBitSet().stream().forEach(a ->
-        builder.addRights(OzoneAclRights.valueOf(ACLType.values()[a].name())));
+        .setType(OzoneAclType.valueOf(acl.getType().name()))
+        .setRights(ByteString.copyFrom(acl.getAclBitSet().toByteArray()));
     return builder.build();
   }
 
   public static OzoneAcl fromProtobuf(OzoneAclInfo protoAcl) {
-    BitSet aclRights = new BitSet(ACLType.getNoOfAcls());
-    protoAcl.getRightsList().parallelStream().forEach(a ->
-        aclRights.set(a.ordinal()));
-
+    BitSet aclRights = BitSet.valueOf(protoAcl.getRights().toByteArray());
     return new OzoneAcl(ACLIdentityType.valueOf(protoAcl.getType().name()),
         protoAcl.getName(), aclRights);
   }
@@ -215,11 +227,11 @@ public BitSet getAclBitSet() {
   }
 
   public List<ACLType> getAclList() {
-    List<ACLType> acls = new ArrayList<>(ACLType.getNoOfAcls());
     if(aclBitSet !=  null) {
-      aclBitSet.stream().forEach(a -> acls.add(ACLType.values()[a]));
+      return aclBitSet.stream().mapToObj(a ->
+          ACLType.values()[a]).collect(Collectors.toList());
     }
-    return acls;
+    return EMPTY_LIST;
   }
 
   /**

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/helpers/OmOzoneAclMap.java

@@ -18,15 +18,17 @@
 
 package org.apache.hadoop.ozone.om.helpers;
 
+import com.google.protobuf.ByteString;
 import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.protocol.proto
     .OzoneManagerProtocolProtos.OzoneAclInfo;
-import org.apache.hadoop.ozone.protocol.proto
-    .OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclRights;
 import org.apache.hadoop.ozone.protocol.proto
     .OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclType;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
+import org.apache.hadoop.ozone.web.utils.OzoneUtils;
+import org.apache.hadoop.security.UserGroupInformation;
 
 import java.util.BitSet;
 import java.util.List;
@@ -38,7 +40,8 @@
 
 import static org.apache.hadoop.ozone.OzoneAcl.ZERO_BITSET;
 import static org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes.INVALID_REQUEST;
-import static org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclRights.ALL;
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.ALL;
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.NONE;
 
 /**
  * This helper class keeps a map of all user and their permissions.
@@ -92,7 +95,7 @@ public void addAcl(OzoneAcl acl) throws OMException {
         throw new OMException("Acl " + acl + " already exist.",
             INVALID_REQUEST);
       }
-      getMap(aclType).get(acl.getName()).or(acl.getAclBitSet());
+      getMap(aclType).replace(acl.getName(), temp);
     }
   }
 
@@ -143,8 +146,7 @@ public void removeAcl(OzoneAcl acl) throws OMException {
   public void addAcl(OzoneAclInfo acl) throws OMException {
     Objects.requireNonNull(acl, "Acl should not be null.");
     if (!getMap(acl.getType()).containsKey(acl.getName())) {
-      BitSet acls = new BitSet(OzoneAclRights.values().length);
-      acl.getRightsList().parallelStream().forEach(a -> acls.set(a.ordinal()));
+      BitSet acls = BitSet.valueOf(acl.getRights().toByteArray());
       getMap(acl.getType()).put(acl.getName(), acls);
     } else {
       // throw exception if acl is already added.
@@ -163,11 +165,66 @@ public boolean hasAccess(OzoneAclInfo acl) {
     if (aclBitSet == null) {
       return false;
     }
+    BitSet result = BitSet.valueOf(acl.getRights().toByteArray());
+    result.and(aclBitSet);
+    return (!result.equals(ZERO_BITSET) || aclBitSet.get(ALL.ordinal()))
+        && !aclBitSet.get(NONE.ordinal());
+  }
+
+  /**
+   * For a given acl, check if the user has access rights.
+   * Acl's are checked in followoing order:
+   * 1. Acls for USER.
+   * 2. Acls for GROUPS.
+   * 3. Acls for WORLD.
+   * 4. Acls for ANONYMOUS.
+   * @param acl
+   * @param ugi
+   *
+   * @return true if given ugi has acl set, else false.
+   * */
+  public boolean hasAccess(ACLType acl, UserGroupInformation ugi) {
+    if (acl == null) {
+      return false;
+    }
+    if (ugi == null) {
+      return false;
+    }
 
-    for (OzoneAclRights right : acl.getRightsList()) {
-      if (aclBitSet.get(right.ordinal()) || aclBitSet.get(ALL.ordinal())) {
+    // Check acls in user acl list.
+    return checkAccessForOzoneAclType(OzoneAclType.USER, acl, ugi)
+        || checkAccessForOzoneAclType(OzoneAclType.GROUP, acl, ugi)
+        || checkAccessForOzoneAclType(OzoneAclType.WORLD, acl, ugi)
+        || checkAccessForOzoneAclType(OzoneAclType.ANONYMOUS, acl, ugi);
+  }
+
+  /**
+   * Helper function to check acl access for OzoneAclType.
+   * */
+  private boolean checkAccessForOzoneAclType(OzoneAclType identityType,
+      ACLType acl, UserGroupInformation ugi) {
+
+    switch (identityType) {
+    case USER:
+      return OzoneUtils.checkIfAclBitIsSet(acl, getAcl(identityType,
+          ugi.getUserName()));
+    case GROUP:
+      // Check access for user groups.
+      for (String userGroup : ugi.getGroupNames()) {
+        if (OzoneUtils.checkIfAclBitIsSet(acl, getAcl(identityType,
+            userGroup))) {
+          // Return true if any user group has required permission.
+          return true;
+        }
+      }
+      break;
+    default:
+      // For type WORLD and ANONYMOUS we set acl type as name.
+      if(OzoneUtils.checkIfAclBitIsSet(acl, getAcl(identityType,
+          identityType.name()))) {
         return true;
       }
+
     }
     return false;
   }
@@ -180,13 +237,12 @@ public List<OzoneAclInfo> ozoneAclGetProtobuf() {
           aclMaps.get(type.ordinal()).entrySet()) {
         OzoneAclInfo.Builder builder = OzoneAclInfo.newBuilder()
             .setName(entry.getKey())
-            .setType(type);
-        entry.getValue().stream().forEach(a ->
-            builder.addRights(OzoneAclRights.values()[a]));
+            .setType(type)
+            .setRights(ByteString.copyFrom(entry.getValue().toByteArray()));
+
         aclList.add(builder.build());
       }
     }
-
     return aclList;
   }
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/om/protocolPB/OzoneManagerProtocolClientSideTranslatorPB.java

@@ -1073,6 +1073,8 @@ public OmMultipartUploadCompleteInfo completeMultipartUpload(
         .setVolumeName(omKeyArgs.getVolumeName())
         .setBucketName(omKeyArgs.getBucketName())
         .setKeyName(omKeyArgs.getKeyName())
+        .addAllAcls(omKeyArgs.getAcls().stream().map(a ->
+            OzoneAcl.toProtobuf(a)).collect(Collectors.toList()))
         .setMultipartUploadID(omKeyArgs.getMultipartUploadID());
 
     multipartUploadCompleteRequest.setKeyArgs(keyArgs.build());

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/protocolPB/OMPBHelper.java

@@ -38,17 +38,12 @@
     .OzoneManagerProtocolProtos.OzoneAclInfo;
 import org.apache.hadoop.ozone.protocol.proto
     .OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclType;
-import org.apache.hadoop.ozone.protocol.proto
-    .OzoneManagerProtocolProtos.OzoneAclInfo.OzoneAclRights;
 import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
-import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
 import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
 import org.apache.hadoop.security.proto.SecurityProtos.TokenProto;
 import org.apache.hadoop.security.token.Token;
 
 import java.util.BitSet;
-import java.util.List;
-import java.util.ArrayList;
 
 /**
  * Utilities for converting protobuf classes.
@@ -84,14 +79,10 @@ public static OzoneAclInfo convertOzoneAcl(OzoneAcl acl) {
     default:
       throw new IllegalArgumentException("ACL type is not recognized");
     }
-    List<OzoneAclRights> ozAclRights =
-        new ArrayList<>(acl.getAclBitSet().cardinality());
-    acl.getAclBitSet().stream().forEach(a -> ozAclRights.add(
-        OzoneAclRights.valueOf(ACLType.values()[a].name())));
 
     return OzoneAclInfo.newBuilder().setType(aclType)
         .setName(acl.getName())
-        .addAllRights(ozAclRights)
+        .setRights(ByteString.copyFrom(acl.getAclBitSet().toByteArray()))
         .build();
   }
 
@@ -121,9 +112,7 @@ public static OzoneAcl convertOzoneAcl(OzoneAclInfo aclInfo) {
       throw new IllegalArgumentException("ACL type is not recognized");
     }
 
-    BitSet aclRights = new BitSet(ACLType.getNoOfAcls());
-    aclInfo.getRightsList().stream().forEach(a ->
-        aclRights.set(ACLType.valueOf(a.name()).ordinal()));
+    BitSet aclRights = BitSet.valueOf(aclInfo.getRights().toByteArray());
     return new OzoneAcl(aclType, aclInfo.getName(), aclRights);
   }
 

hadoop-ozone/common/src/main/java/org/apache/hadoop/ozone/security/acl/IAccessAuthorizer.java

@@ -56,11 +56,20 @@ enum ACLType {
     ALL,
     NONE;
     private static int length = ACLType.values().length;
+    private static ACLType[] vals = ACLType.values();
 
     public static int getNoOfAcls() {
       return length;
     }
 
+    public static ACLType getAclTypeFromOrdinal(int ordinal) {
+      if (ordinal > length - 1 && ordinal > -1) {
+        throw new IllegalArgumentException("Ordinal greater than array lentgh" +
+            ". ordinal:" + ordinal);
+      }
+      return vals[ordinal];
+    }
+
     /**
      * Returns the ACL rights based on passed in String.
      *
@@ -145,9 +154,11 @@ public static String getAclString(ACLType acl) {
   enum ACLIdentityType {
     USER(OzoneConsts.OZONE_ACL_USER_TYPE),
     GROUP(OzoneConsts.OZONE_ACL_GROUP_TYPE),
-    CLIENT_IP(OzoneConsts.OZONE_ACL_IP_TYPE),
     WORLD(OzoneConsts.OZONE_ACL_WORLD_TYPE),
-    ANONYMOUS(OzoneConsts.OZONE_ACL_ANONYMOUS_TYPE);
+    ANONYMOUS(OzoneConsts.OZONE_ACL_ANONYMOUS_TYPE),
+    CLIENT_IP(OzoneConsts.OZONE_ACL_IP_TYPE);
+
+    // TODO: Add support for acl checks based on CLIENT_IP.
 
     @Override
     public String toString() {