HDDS-1538. Update ozone protobuf message for ACLs. Contributed by Ajay Kumar. (#828)

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

@@ -52,12 +52,11 @@ public final class OzoneConsts {
   public static final String OZONE_ACL_USER_TYPE = "user";
   public static final String OZONE_ACL_GROUP_TYPE = "group";
   public static final String OZONE_ACL_WORLD_TYPE = "world";
+  public static final String OZONE_ACL_ANONYMOUS_TYPE = "anonymous";
   public static final String OZONE_ACL_IP_TYPE = "ip";
 
   public static final String OZONE_ACL_READ = "r";
   public static final String OZONE_ACL_WRITE = "w";
-  public static final String OZONE_ACL_READ_WRITE = "rw";
-  public static final String OZONE_ACL_WRITE_READ = "wr";
   public static final String OZONE_ACL_DELETE = "d";
   public static final String OZONE_ACL_LIST = "l";
   public static final String OZONE_ACL_ALL = "a";

hadoop-hdds/common/src/main/resources/ozone-default.xml

@@ -540,14 +540,6 @@
       the address of the OM.
     </description>
   </property>
-  <property>
-    <name>ozone.om.group.rights</name>
-    <value>READ_WRITE</value>
-    <tag>OM, SECURITY</tag>
-    <description>
-      Default group permissions in Ozone OM.
-    </description>
-  </property>
   <property>
     <name>ozone.om.handler.count.key</name>
     <value>20</value>
@@ -640,14 +632,6 @@
       of buckets or keys inside each bucket a user can create.
     </description>
   </property>
-  <property>
-    <name>ozone.om.user.rights</name>
-    <value>READ_WRITE</value>
-    <tag>OM, SECURITY</tag>
-    <description>
-      Default user permissions used in OM.
-    </description>
-  </property>
   <property>
     <name>ozone.om.db.dirs</name>
     <value/>

hadoop-ozone/client/src/main/java/org/apache/hadoop/ozone/client/rest/RestClient.java

@@ -112,7 +112,7 @@ public class RestClient implements ClientProtocol {
   private final URI ozoneRestUri;
   private final CloseableHttpClient httpClient;
   private final UserGroupInformation ugi;
-  private final OzoneAcl.OzoneACLRights userRights;
+  // private final OzoneAcl.OzoneACLRights userRights;
 
    /**
     * Creates RestClient instance with the given configuration.
@@ -161,8 +161,8 @@ public class RestClient implements ClientProtocol {
                   .build())
           .build();
 
-      this.userRights = conf.getEnum(OMConfigKeys.OZONE_OM_USER_RIGHTS,
+//      this.userRights = conf.getEnum(OMConfigKeys.OZONE_OM_USER_RIGHTS,
-          OMConfigKeys.OZONE_OM_USER_RIGHTS_DEFAULT);
+//          OMConfigKeys.OZONE_OM_USER_RIGHTS_DEFAULT);
 
       // TODO: Add new configuration parameter to configure RestServerSelector.
       RestServerSelector defaultSelector = new DefaultRestServerSelector();

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

@@ -73,7 +73,6 @@ import org.apache.hadoop.ozone.om.protocolPB
     .OzoneManagerProtocolClientSideTranslatorPB;
 import org.apache.hadoop.net.NetUtils;
 import org.apache.hadoop.ozone.OzoneAcl;
-import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.protocol.proto
     .OzoneManagerProtocolProtos.ServicePort;
 import org.apache.hadoop.hdds.protocol.proto.HddsProtos;
@@ -85,6 +84,9 @@ import org.apache.hadoop.hdds.scm.protocolPB
 import org.apache.hadoop.hdds.scm.protocolPB
     .StorageContainerLocationProtocolPB;
 import org.apache.hadoop.ozone.security.OzoneTokenIdentifier;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
+import org.apache.hadoop.ozone.security.acl.OzoneAclConfig;
 import org.apache.hadoop.ozone.web.utils.OzoneUtils;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
@@ -121,8 +123,8 @@ public class RpcClient implements ClientProtocol, KeyProviderTokenIssuer {
   private final int bytesPerChecksum;
   private boolean verifyChecksum;
   private final UserGroupInformation ugi;
-  private final OzoneAcl.OzoneACLRights userRights;
+  private final ACLType userRights;
-  private final OzoneAcl.OzoneACLRights groupRights;
+  private final ACLType groupRights;
   private final long streamBufferFlushSize;
   private final long streamBufferMaxSize;
   private final long blockSize;
@@ -141,10 +143,11 @@ public class RpcClient implements ClientProtocol, KeyProviderTokenIssuer {
     Preconditions.checkNotNull(conf);
     this.conf = new OzoneConfiguration(conf);
     this.ugi = UserGroupInformation.getCurrentUser();
-    this.userRights = conf.getEnum(OMConfigKeys.OZONE_OM_USER_RIGHTS,
+    // Get default acl rights for user and group.
-        OMConfigKeys.OZONE_OM_USER_RIGHTS_DEFAULT);
+    OzoneAclConfig aclConfig = this.conf.getObject(OzoneAclConfig.class);
-    this.groupRights = conf.getEnum(OMConfigKeys.OZONE_OM_GROUP_RIGHTS,
+    this.userRights = aclConfig.getUserDefaultRights();
-        OMConfigKeys.OZONE_OM_GROUP_RIGHTS_DEFAULT);
+    this.groupRights = aclConfig.getGroupDefaultRights();
+
     this.ozoneManagerClient = new OzoneManagerProtocolClientSideTranslatorPB(
         this.conf, clientId.toString(), ugi);
     long scmVersion =
@@ -256,13 +259,13 @@ public class RpcClient implements ClientProtocol, KeyProviderTokenIssuer {
         OzoneQuota.parseQuota(volArgs.getQuota()).sizeInBytes();
     List<OzoneAcl> listOfAcls = new ArrayList<>();
     //User ACL
-    listOfAcls.add(new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+    listOfAcls.add(new OzoneAcl(ACLIdentityType.USER,
             owner, userRights));
     //Group ACLs of the User
     List<String> userGroups = Arrays.asList(UserGroupInformation
         .createRemoteUser(owner).getGroupNames());
     userGroups.stream().forEach((group) -> listOfAcls.add(
-        new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, group, groupRights)));
+        new OzoneAcl(ACLIdentityType.GROUP, group, groupRights)));
     //ACLs from VolumeArgs
     if(volArgs.getAcls() != null) {
       listOfAcls.addAll(volArgs.getAcls());
@@ -403,13 +406,13 @@ public class RpcClient implements ClientProtocol, KeyProviderTokenIssuer {
 
     List<OzoneAcl> listOfAcls = new ArrayList<>();
     //User ACL
-    listOfAcls.add(new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+    listOfAcls.add(new OzoneAcl(ACLIdentityType.USER,
         ugi.getUserName(), userRights));
     //Group ACLs of the User
     List<String> userGroups = Arrays.asList(UserGroupInformation
         .createRemoteUser(ugi.getUserName()).getGroupNames());
     userGroups.stream().forEach((group) -> listOfAcls.add(
-        new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, group, groupRights)));
+        new OzoneAcl(ACLIdentityType.GROUP, group, groupRights)));
     //ACLs from BucketArgs
     if(bucketArgs.getAcls() != null) {
       listOfAcls.addAll(bucketArgs.getAcls());

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

@@ -19,6 +19,11 @@
 
 package org.apache.hadoop.ozone;
 
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
+
+import java.util.ArrayList;
+import java.util.List;
 import java.util.Objects;
 
 /**
@@ -32,9 +37,9 @@ import java.util.Objects;
  * </ul>
  */
 public class OzoneAcl {
-  private OzoneACLType type;
+  private ACLIdentityType type;
   private String name;
-  private OzoneACLRights rights;
+  private List<ACLType> rights;
 
   /**
    * Constructor for OzoneAcl.
@@ -47,16 +52,37 @@ public class OzoneAcl {
    *
    * @param type - Type
    * @param name - Name of user
-   * @param rights - Rights
+   * @param acl - Rights
    */
-  public OzoneAcl(OzoneACLType type, String name, OzoneACLRights rights) {
+  public OzoneAcl(ACLIdentityType type, String name, ACLType acl) {
     this.name = name;
-    this.rights = rights;
+    this.rights = new ArrayList<>();
+    this.rights.add(acl);
     this.type = type;
-    if (type == OzoneACLType.WORLD && name.length() != 0) {
+    if (type == ACLIdentityType.WORLD && name.length() != 0) {
       throw new IllegalArgumentException("Unexpected name part in world type");
     }
-    if (((type == OzoneACLType.USER) || (type == OzoneACLType.GROUP))
+    if (((type == ACLIdentityType.USER) || (type == ACLIdentityType.GROUP))
+        && (name.length() == 0)) {
+      throw new IllegalArgumentException("User or group name is required");
+    }
+  }
+
+  /**
+   * Constructor for OzoneAcl.
+   *
+   * @param type - Type
+   * @param name - Name of user
+   * @param acls - Rights
+   */
+  public OzoneAcl(ACLIdentityType type, String name, List<ACLType> acls) {
+    this.name = name;
+    this.rights = acls;
+    this.type = type;
+    if (type == ACLIdentityType.WORLD && name.length() != 0) {
+      throw new IllegalArgumentException("Unexpected name part in world type");
+    }
+    if (((type == ACLIdentityType.USER) || (type == ACLIdentityType.GROUP))
         && (name.length() == 0)) {
       throw new IllegalArgumentException("User or group name is required");
     }
@@ -78,17 +104,20 @@ public class OzoneAcl {
       throw new IllegalArgumentException("ACLs are not in expected format");
     }
 
-    OzoneACLType aclType = OzoneACLType.valueOf(parts[0].toUpperCase());
+    ACLIdentityType aclType = ACLIdentityType.valueOf(parts[0].toUpperCase());
-    OzoneACLRights rights = OzoneACLRights.getACLRight(parts[2].toLowerCase());
+    List<ACLType> acls = new ArrayList<>();
+    for (char ch : parts[2].toCharArray()) {
+      acls.add(ACLType.getACLRight(String.valueOf(ch)));
+    }
 
     // TODO : Support sanitation of these user names by calling into
     // userAuth Interface.
-    return new OzoneAcl(aclType, parts[1], rights);
+    return new OzoneAcl(aclType, parts[1], acls);
   }
 
   @Override
   public String toString() {
-    return type + ":" + name + ":" + OzoneACLRights.getACLRightsString(rights);
+    return type + ":" + name + ":" + ACLType.getACLString(rights);
   }
 
   /**
@@ -120,7 +149,7 @@ public class OzoneAcl {
    *
    * @return - Rights
    */
-  public OzoneACLRights getRights() {
+  public List<ACLType> getRights() {
     return rights;
   }
 
@@ -129,7 +158,7 @@ public class OzoneAcl {
    *
    * @return type
    */
-  public OzoneACLType getType() {
+  public ACLIdentityType getType() {
     return type;
   }
 
@@ -150,9 +179,7 @@ public class OzoneAcl {
       return false;
     }
     OzoneAcl otherAcl = (OzoneAcl) obj;
-    return otherAcl.getName().equals(this.getName()) &&
+    return otherAcl.toString().equals(this.toString());
-        otherAcl.getRights() == this.getRights() &&
-        otherAcl.getType() == this.getType();
   }
 
   /**
@@ -177,57 +204,4 @@ public class OzoneAcl {
       value = val;
     }
   }
-
-  /**
-   * ACL rights.
-   */
-  public enum OzoneACLRights {
-    READ, WRITE, READ_WRITE;
-
-    /**
-     * Returns the ACL rights based on passed in String.
-     *
-     * @param type ACL right string
-     *
-     * @return OzoneACLRights
-     */
-    public static OzoneACLRights getACLRight(String type) {
-      if (type == null || type.isEmpty()) {
-        throw new IllegalArgumentException("ACL right cannot be empty");
-      }
-
-      switch (type) {
-      case OzoneConsts.OZONE_ACL_READ:
-        return OzoneACLRights.READ;
-      case OzoneConsts.OZONE_ACL_WRITE:
-        return OzoneACLRights.WRITE;
-      case OzoneConsts.OZONE_ACL_READ_WRITE:
-      case OzoneConsts.OZONE_ACL_WRITE_READ:
-        return OzoneACLRights.READ_WRITE;
-      default:
-        throw new IllegalArgumentException("ACL right is not recognized");
-      }
-
-    }
-
-    /**
-     * Returns String representation of ACL rights.
-     * @param acl OzoneACLRights
-     * @return String representation of acl
-     */
-    public static String getACLRightsString(OzoneACLRights acl) {
-      switch(acl) {
-      case READ:
-        return OzoneConsts.OZONE_ACL_READ;
-      case WRITE:
-        return OzoneConsts.OZONE_ACL_WRITE;
-      case READ_WRITE:
-        return OzoneConsts.OZONE_ACL_READ_WRITE;
-      default:
-        throw new IllegalArgumentException("ACL right is not recognized");
-      }
-    }
-
-  }
-
 }

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

@@ -19,7 +19,6 @@ package org.apache.hadoop.ozone.om;
 
 import java.util.concurrent.TimeUnit;
 
-import org.apache.hadoop.ozone.OzoneAcl;
 import org.apache.ratis.util.TimeDuration;
 
 /**
@@ -78,17 +77,6 @@ public final class OMConfigKeys {
       "ozone.om.user.max.volume";
   public static final int OZONE_OM_USER_MAX_VOLUME_DEFAULT = 1024;
 
-  // OM Default user/group permissions
-  public static final String OZONE_OM_USER_RIGHTS =
-      "ozone.om.user.rights";
-  public static final OzoneAcl.OzoneACLRights OZONE_OM_USER_RIGHTS_DEFAULT =
-      OzoneAcl.OzoneACLRights.READ_WRITE;
-
-  public static final String OZONE_OM_GROUP_RIGHTS =
-      "ozone.om.group.rights";
-  public static final OzoneAcl.OzoneACLRights OZONE_OM_GROUP_RIGHTS_DEFAULT =
-      OzoneAcl.OzoneACLRights.READ_WRITE;
-
   public static final String OZONE_KEY_DELETING_LIMIT_PER_TASK =
       "ozone.key.deleting.limit.per.task";
   public static final int OZONE_KEY_DELETING_LIMIT_PER_TASK_DEFAULT = 1000;

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

@@ -37,7 +37,7 @@ import java.util.HashMap;
 @SuppressWarnings("ProtocolBufferOrdinal")
 public class OmOzoneAclMap {
   // per Acl Type user:rights map
-  private ArrayList<Map<String, OzoneAclRights>> aclMaps;
+  private ArrayList<Map<String, List<OzoneAclRights>>> aclMaps;
 
   OmOzoneAclMap() {
     aclMaps = new ArrayList<>();
@@ -46,51 +46,75 @@ public class OmOzoneAclMap {
     }
   }
 
-  private Map<String, OzoneAclRights> getMap(OzoneAclType type) {
+  private Map<String, List<OzoneAclRights>> getMap(OzoneAclType type) {
     return aclMaps.get(type.ordinal());
   }
 
   // For a given acl type and user, get the stored acl
-  private OzoneAclRights getAcl(OzoneAclType type, String user) {
+  private List<OzoneAclRights> getAcl(OzoneAclType type, String user) {
     return getMap(type).get(user);
   }
 
   // Add a new acl to the map
   public void addAcl(OzoneAclInfo acl) {
-    getMap(acl.getType()).put(acl.getName(), acl.getRights());
+    getMap(acl.getType()).put(acl.getName(), acl.getRightsList());
   }
 
   // for a given acl, check if the user has access rights
   public boolean hasAccess(OzoneAclInfo acl) {
-    OzoneAclRights storedRights = getAcl(acl.getType(), acl.getName());
+    if (acl == null) {
-    if (storedRights != null) {
+      return false;
-      switch (acl.getRights()) {
+    }
-      case READ:
+
-        return (storedRights == OzoneAclRights.READ)
+    List<OzoneAclRights> storedRights = getAcl(acl.getType(), acl.getName());
-            || (storedRights == OzoneAclRights.READ_WRITE);
+    if(storedRights == null) {
+      return false;
+    }
+
+    for (OzoneAclRights right : storedRights) {
+      switch (right) {
+      case CREATE:
+        return (right == OzoneAclRights.CREATE)
+            || (right == OzoneAclRights.ALL);
+      case LIST:
+        return (right == OzoneAclRights.LIST)
+            || (right == OzoneAclRights.ALL);
       case WRITE:
-        return (storedRights == OzoneAclRights.WRITE)
+        return (right == OzoneAclRights.WRITE)
-            || (storedRights == OzoneAclRights.READ_WRITE);
+            || (right == OzoneAclRights.ALL);
-      case READ_WRITE:
+      case READ:
-        return (storedRights == OzoneAclRights.READ_WRITE);
+        return (right == OzoneAclRights.READ)
+            || (right == OzoneAclRights.ALL);
+      case DELETE:
+        return (right == OzoneAclRights.DELETE)
+            || (right == OzoneAclRights.ALL);
+      case READ_ACL:
+        return (right == OzoneAclRights.READ_ACL)
+            || (right == OzoneAclRights.ALL);
+      case WRITE_ACL:
+        return (right == OzoneAclRights.WRITE_ACL)
+            || (right == OzoneAclRights.ALL);
+      case ALL:
+        return (right == OzoneAclRights.ALL);
+      case NONE:
+        return !(right == OzoneAclRights.NONE);
       default:
         return false;
       }
-    } else {
-      return false;
     }
+    return false;
   }
 
   // Convert this map to OzoneAclInfo Protobuf List
   public List<OzoneAclInfo> ozoneAclGetProtobuf() {
     List<OzoneAclInfo> aclList = new LinkedList<>();
     for (OzoneAclType type: OzoneAclType.values()) {
-      for (Map.Entry<String, OzoneAclRights> entry :
+      for (Map.Entry<String, List<OzoneAclRights>> entry :
           aclMaps.get(type.ordinal()).entrySet()) {
         OzoneAclInfo aclInfo = OzoneAclInfo.newBuilder()
             .setName(entry.getKey())
             .setType(type)
-            .setRights(entry.getValue())
+            .addAllRights(entry.getValue())
             .build();
         aclList.add(aclInfo);
       }

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

@@ -41,9 +41,15 @@ import org.apache.hadoop.ozone.protocol.proto
 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;
+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.List;
+import java.util.ArrayList;
+
 /**
  * Utilities for converting protobuf classes.
  */
@@ -59,7 +65,7 @@ public final class OMPBHelper {
    */
   public static OzoneAclInfo convertOzoneAcl(OzoneAcl acl) {
     OzoneAclInfo.OzoneAclType aclType;
-    switch(acl.getType()) {
+    switch (acl.getType()) {
     case USER:
       aclType = OzoneAclType.USER;
       break;
@@ -69,27 +75,24 @@ public final class OMPBHelper {
     case WORLD:
       aclType = OzoneAclType.WORLD;
       break;
+    case ANONYMOUS:
+      aclType = OzoneAclType.ANONYMOUS;
+      break;
+    case CLIENT_IP:
+      aclType = OzoneAclType.CLIENT_IP;
+      break;
     default:
       throw new IllegalArgumentException("ACL type is not recognized");
     }
-    OzoneAclInfo.OzoneAclRights aclRights;
+    List<OzoneAclRights> aclRights = new ArrayList<>();
-    switch(acl.getRights()) {
+
-    case READ:
+    for (ACLType right : acl.getRights()) {
-      aclRights = OzoneAclRights.READ;
+      aclRights.add(OzoneAclRights.valueOf(right.name()));
-      break;
-    case WRITE:
-      aclRights = OzoneAclRights.WRITE;
-      break;
-    case READ_WRITE:
-      aclRights = OzoneAclRights.READ_WRITE;
-      break;
-    default:
-      throw new IllegalArgumentException("ACL right is not recognized");
     }
 
     return OzoneAclInfo.newBuilder().setType(aclType)
         .setName(acl.getName())
-        .setRights(aclRights)
+        .addAllRights(aclRights)
         .build();
   }
 
@@ -98,35 +101,31 @@ public final class OMPBHelper {
    * @return OzoneAcl
    */
   public static OzoneAcl convertOzoneAcl(OzoneAclInfo aclInfo) {
-    OzoneAcl.OzoneACLType aclType;
+    ACLIdentityType aclType;
-    switch(aclInfo.getType()) {
+    switch (aclInfo.getType()) {
     case USER:
-      aclType = OzoneAcl.OzoneACLType.USER;
+      aclType = ACLIdentityType.USER;
       break;
     case GROUP:
-      aclType = OzoneAcl.OzoneACLType.GROUP;
+      aclType = ACLIdentityType.GROUP;
       break;
     case WORLD:
-      aclType = OzoneAcl.OzoneACLType.WORLD;
+      aclType = ACLIdentityType.WORLD;
+      break;
+    case ANONYMOUS:
+      aclType = ACLIdentityType.ANONYMOUS;
+      break;
+    case CLIENT_IP:
+      aclType = ACLIdentityType.CLIENT_IP;
       break;
     default:
       throw new IllegalArgumentException("ACL type is not recognized");
     }
-    OzoneAcl.OzoneACLRights aclRights;
-    switch(aclInfo.getRights()) {
-    case READ:
-      aclRights = OzoneAcl.OzoneACLRights.READ;
-      break;
-    case WRITE:
-      aclRights = OzoneAcl.OzoneACLRights.WRITE;
-      break;
-    case READ_WRITE:
-      aclRights = OzoneAcl.OzoneACLRights.READ_WRITE;
-      break;
-    default:
-      throw new IllegalArgumentException("ACL right is not recognized");
-    }
 
+    List<IAccessAuthorizer.ACLType> aclRights = new ArrayList<>();
+    for (OzoneAclRights acl : aclInfo.getRightsList()) {
+      aclRights.add(ACLType.valueOf(acl.name()));
+    }
     return new OzoneAcl(aclType, aclInfo.getName(), aclRights);
   }
 

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

@@ -20,6 +20,8 @@ import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.ozone.OzoneConsts;
 
+import java.util.List;
+
 /**
  * Public API for Ozone ACLs. Security providers providing support for Ozone
  * ACLs should implement this.
@@ -84,7 +86,8 @@ public interface IAccessAuthorizer {
       case OzoneConsts.OZONE_ACL_NONE:
         return ACLType.NONE;
       default:
-        throw new IllegalArgumentException("ACL right is not recognized");
+        throw new IllegalArgumentException(type + " ACL right is not " +
+            "recognized");
       }
 
     }
@@ -92,10 +95,18 @@ public interface IAccessAuthorizer {
     /**
      * Returns String representation of ACL rights.
      *
-     * @param acl ACLType
+     * @param acls ACLType
      * @return String representation of acl
      */
-    public static String getACLRightsString(ACLType acl) {
+    public static String getACLString(List<ACLType> acls) {
+      StringBuffer sb = new StringBuffer();
+      acls.forEach(acl -> {
+        sb.append(getAclString(acl));
+      });
+      return sb.toString();
+    }
+
+    public static String getAclString(ACLType acl) {
       switch (acl) {
       case READ:
         return OzoneConsts.OZONE_ACL_READ;
@@ -129,7 +140,8 @@ public interface IAccessAuthorizer {
     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);
+    WORLD(OzoneConsts.OZONE_ACL_WORLD_TYPE),
+    ANONYMOUS(OzoneConsts.OZONE_ACL_ANONYMOUS_TYPE);
 
     @Override
     public String toString() {

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

@@ -0,0 +1,65 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.hadoop.ozone.security.acl;
+
+import org.apache.hadoop.hdds.conf.Config;
+import org.apache.hadoop.hdds.conf.ConfigGroup;
+import org.apache.hadoop.hdds.conf.ConfigTag;
+import org.apache.hadoop.hdds.conf.ConfigType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
+
+/**
+ * Ozone ACL config pojo.
+ * */
+@ConfigGroup(prefix = "ozone.om")
+public class OzoneAclConfig {
+  // OM Default user/group permissions
+  private ACLType userDefaultRights = ACLType.ALL;
+  private ACLType groupDefaultRights = ACLType.ALL;
+
+  @Config(key = "user.rights",
+      defaultValue = "ALL",
+      type = ConfigType.STRING,
+      tags = {ConfigTag.OM, ConfigTag.SECURITY},
+      description = "Default user permissions set for an object in " +
+          "OzoneManager."
+  )
+  public void setUserDefaultRights(String userRights) {
+    this.userDefaultRights = ACLType.valueOf(userRights);
+  }
+
+  @Config(key = "group.rights",
+      defaultValue = "ALL",
+      type = ConfigType.STRING,
+      tags = {ConfigTag.OM, ConfigTag.SECURITY},
+      description = "Default group permissions set for an object in " +
+          "OzoneManager."
+  )
+  public void setGroupDefaultRights(String groupRights) {
+    this.groupDefaultRights = ACLType.valueOf(groupRights);
+  }
+
+  public ACLType getUserDefaultRights() {
+    return userDefaultRights;
+  }
+
+  public ACLType getGroupDefaultRights() {
+    return groupDefaultRights;
+  }
+
+}

hadoop-ozone/common/src/main/proto/OzoneManagerProtocol.proto

@@ -451,15 +451,24 @@ message OzoneAclInfo {
         USER = 1;
         GROUP = 2;
         WORLD = 3;
+        ANONYMOUS = 4;
+        CLIENT_IP = 5;
     }
+
     enum OzoneAclRights {
-        READ = 1;
+        CREATE = 1;
-        WRITE = 2;
+        LIST = 2;
-        READ_WRITE = 3;
+        DELETE = 3;
+        READ = 4;
+        WRITE = 5;
+        READ_ACL = 6;
+        WRITE_ACL = 7;
+        ALL = 8;
+        NONE = 9;
     }
     required OzoneAclType type = 1;
     required string name = 2;
-    required OzoneAclRights rights = 3;
+    repeated OzoneAclRights rights = 3;
 }
 
 message CreateBucketRequest {

hadoop-ozone/common/src/test/java/org/apache/hadoop/ozone/TestOzoneAcls.java

@@ -18,16 +18,20 @@
 
 package org.apache.hadoop.ozone;
 
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+
 import org.junit.Test;
 
+import java.util.Arrays;
 import java.util.HashMap;
 import java.util.Set;
 
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.*;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.fail;
 
 /**
- * This class is to test acl stoarge and retreival in ozone store.
+ * This class is to test acl storage and retrieval in ozone store.
  */
 public class TestOzoneAcls {
 
@@ -39,8 +43,8 @@ public class TestOzoneAcls {
     testMatrix.put("user:bilbo:r", Boolean.TRUE);
     testMatrix.put("user:bilbo:w", Boolean.TRUE);
     testMatrix.put("user:bilbo:rw", Boolean.TRUE);
-    testMatrix.put("user:bilbo:wr", Boolean.TRUE);
+    testMatrix.put("user:bilbo:a", Boolean.TRUE);
-    testMatrix.put("    user:bilbo:wr   ", Boolean.TRUE);
+    testMatrix.put("    user:bilbo:a   ", Boolean.TRUE);
 
 
     // ACLs makes no judgement on the quality of
@@ -53,7 +57,16 @@ public class TestOzoneAcls {
     testMatrix.put("", Boolean.FALSE);
     testMatrix.put(null, Boolean.FALSE);
     testMatrix.put(" user:bilbo:", Boolean.FALSE);
-    testMatrix.put(" user:bilbo:rx", Boolean.FALSE);
+    testMatrix.put(" user:bilbo:rx", Boolean.TRUE);
+    testMatrix.put(" user:bilbo:rwdlncxy", Boolean.TRUE);
+    testMatrix.put(" group:bilbo:rwdlncxy", Boolean.TRUE);
+    testMatrix.put(" world::rwdlncxy", Boolean.TRUE);
+    testMatrix.put(" user:bilbo:rncxy", Boolean.TRUE);
+    testMatrix.put(" group:bilbo:ncxy", Boolean.TRUE);
+    testMatrix.put(" world::ncxy", Boolean.TRUE);
+    testMatrix.put(" user:bilbo:rwcxy", Boolean.TRUE);
+    testMatrix.put(" group:bilbo:rwcxy", Boolean.TRUE);
+    testMatrix.put(" world::rwcxy", Boolean.TRUE);
     testMatrix.put(" user:bilbo:mk", Boolean.FALSE);
     testMatrix.put(" user::rw", Boolean.FALSE);
     testMatrix.put("user11:bilbo:rw", Boolean.FALSE);
@@ -62,12 +75,12 @@ public class TestOzoneAcls {
     testMatrix.put(" group:hobbit:r", Boolean.TRUE);
     testMatrix.put(" group:hobbit:w", Boolean.TRUE);
     testMatrix.put(" group:hobbit:rw", Boolean.TRUE);
-    testMatrix.put(" group:hobbit:wr", Boolean.TRUE);
+    testMatrix.put(" group:hobbit:a", Boolean.TRUE);
     testMatrix.put(" group:*:rw", Boolean.TRUE);
     testMatrix.put(" group:~!:rw", Boolean.TRUE);
 
     testMatrix.put(" group:hobbit:", Boolean.FALSE);
-    testMatrix.put(" group:hobbit:rx", Boolean.FALSE);
+    testMatrix.put(" group:hobbit:rx", Boolean.TRUE);
     testMatrix.put(" group:hobbit:mk", Boolean.FALSE);
     testMatrix.put(" group::", Boolean.FALSE);
     testMatrix.put(" group::rw", Boolean.FALSE);
@@ -77,14 +90,14 @@ public class TestOzoneAcls {
     testMatrix.put("JUNK group:hobbit:r", Boolean.FALSE);
     testMatrix.put("JUNK group:hobbit:w", Boolean.FALSE);
     testMatrix.put("JUNK group:hobbit:rw", Boolean.FALSE);
-    testMatrix.put("JUNK group:hobbit:wr", Boolean.FALSE);
+    testMatrix.put("JUNK group:hobbit:a", Boolean.FALSE);
     testMatrix.put("JUNK group:*:rw", Boolean.FALSE);
     testMatrix.put("JUNK group:~!:rw", Boolean.FALSE);
 
     testMatrix.put(" world::r", Boolean.TRUE);
     testMatrix.put(" world::w", Boolean.TRUE);
     testMatrix.put(" world::rw", Boolean.TRUE);
-    testMatrix.put(" world::wr", Boolean.TRUE);
+    testMatrix.put(" world::a", Boolean.TRUE);
 
     testMatrix.put(" world:bilbo:w", Boolean.FALSE);
     testMatrix.put(" world:bilbo:rw", Boolean.FALSE);
@@ -97,7 +110,7 @@ public class TestOzoneAcls {
         try {
           OzoneAcl.parseAcl(key);
           // should never get here since parseAcl will throw
-          fail("An exception was expected but did not happen.");
+          fail("An exception was expected but did not happen. Key: " + key);
         } catch (IllegalArgumentException e) {
           // nothing to do
         }
@@ -109,33 +122,51 @@ public class TestOzoneAcls {
   public void testAclValues() {
     OzoneAcl acl = OzoneAcl.parseAcl("user:bilbo:rw");
     assertEquals(acl.getName(), "bilbo");
-    assertEquals(OzoneAcl.OzoneACLRights.READ_WRITE, acl.getRights());
+    assertEquals(Arrays.asList(READ, WRITE), acl.getRights());
-    assertEquals(OzoneAcl.OzoneACLType.USER, acl.getType());
+    assertEquals(ACLIdentityType.USER, acl.getType());
 
-    acl = OzoneAcl.parseAcl("user:bilbo:wr");
+    acl = OzoneAcl.parseAcl("user:bilbo:a");
     assertEquals("bilbo", acl.getName());
-    assertEquals(OzoneAcl.OzoneACLRights.READ_WRITE, acl.getRights());
+    assertEquals(Arrays.asList(ALL), acl.getRights());
-    assertEquals(OzoneAcl.OzoneACLType.USER, acl.getType());
+    assertEquals(ACLIdentityType.USER, acl.getType());
 
     acl = OzoneAcl.parseAcl("user:bilbo:r");
     assertEquals("bilbo", acl.getName());
-    assertEquals(OzoneAcl.OzoneACLRights.READ, acl.getRights());
+    assertEquals(Arrays.asList(READ), acl.getRights());
-    assertEquals(OzoneAcl.OzoneACLType.USER, acl.getType());
+    assertEquals(ACLIdentityType.USER, acl.getType());
 
     acl = OzoneAcl.parseAcl("user:bilbo:w");
     assertEquals("bilbo", acl.getName());
-    assertEquals(OzoneAcl.OzoneACLRights.WRITE, acl.getRights());
+    assertEquals(Arrays.asList(WRITE), acl.getRights());
-    assertEquals(OzoneAcl.OzoneACLType.USER, acl.getType());
+    assertEquals(ACLIdentityType.USER, acl.getType());
 
-    acl = OzoneAcl.parseAcl("group:hobbit:wr");
+    acl = OzoneAcl.parseAcl("group:hobbit:a");
     assertEquals(acl.getName(), "hobbit");
-    assertEquals(OzoneAcl.OzoneACLRights.READ_WRITE, acl.getRights());
+    assertEquals(Arrays.asList(ALL), acl.getRights());
-    assertEquals(OzoneAcl.OzoneACLType.GROUP, acl.getType());
+    assertEquals(ACLIdentityType.GROUP, acl.getType());
 
-    acl = OzoneAcl.parseAcl("world::wr");
+    acl = OzoneAcl.parseAcl("world::a");
     assertEquals(acl.getName(), "");
-    assertEquals(OzoneAcl.OzoneACLRights.READ_WRITE, acl.getRights());
+    assertEquals(Arrays.asList(ALL), acl.getRights());
-    assertEquals(OzoneAcl.OzoneACLType.WORLD, acl.getType());
+    assertEquals(ACLIdentityType.WORLD, acl.getType());
+
+    acl = OzoneAcl.parseAcl("user:bilbo:rwdlncxy");
+    assertEquals(acl.getName(), "bilbo");
+    assertEquals(Arrays.asList(READ, WRITE, DELETE, LIST, NONE, CREATE,
+        READ_ACL, WRITE_ACL), acl.getRights());
+    assertEquals(ACLIdentityType.USER, acl.getType());
+
+    acl = OzoneAcl.parseAcl("group:hadoop:rwdlncxy");
+    assertEquals(acl.getName(), "hadoop");
+    assertEquals(Arrays.asList(READ, WRITE, DELETE, LIST, NONE, CREATE,
+        READ_ACL, WRITE_ACL), acl.getRights());
+    assertEquals(ACLIdentityType.GROUP, acl.getType());
+
+    acl = OzoneAcl.parseAcl("world::rwdlncxy");
+    assertEquals(acl.getName(), "");
+    assertEquals(Arrays.asList(READ, WRITE, DELETE, LIST, NONE, CREATE,
+        READ_ACL, WRITE_ACL), acl.getRights());
+    assertEquals(ACLIdentityType.WORLD, acl.getType());
   }
 
 }

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/client/rpc/TestOzoneRpcClientAbstract.java

@@ -82,6 +82,8 @@ import org.apache.hadoop.ozone.om.helpers.OmMultipartCommitUploadPartInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartInfo;
 import org.apache.hadoop.ozone.om.helpers.OmMultipartUploadCompleteInfo;
 import org.apache.hadoop.ozone.s3.util.OzoneS3Util;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
 import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.test.LambdaTestUtils;
 import org.apache.hadoop.util.Time;
@@ -420,8 +422,8 @@ public abstract class TestOzoneRpcClientAbstract {
       throws IOException, OzoneException {
     String volumeName = UUID.randomUUID().toString();
     String bucketName = UUID.randomUUID().toString();
-    OzoneAcl userAcl = new OzoneAcl(OzoneAcl.OzoneACLType.USER, "test",
+    OzoneAcl userAcl = new OzoneAcl(ACLIdentityType.USER, "test",
-        OzoneAcl.OzoneACLRights.READ_WRITE);
+        ACLType.READ);
     List<OzoneAcl> acls = new ArrayList<>();
     acls.add(userAcl);
     store.createVolume(volumeName);
@@ -439,8 +441,8 @@ public abstract class TestOzoneRpcClientAbstract {
       throws IOException, OzoneException {
     String volumeName = UUID.randomUUID().toString();
     String bucketName = UUID.randomUUID().toString();
-    OzoneAcl userAcl = new OzoneAcl(OzoneAcl.OzoneACLType.USER, "test",
+    OzoneAcl userAcl = new OzoneAcl(ACLIdentityType.USER, "test",
-        OzoneAcl.OzoneACLRights.READ_WRITE);
+        ACLType.ALL);
     List<OzoneAcl> acls = new ArrayList<>();
     acls.add(userAcl);
     store.createVolume(volumeName);
@@ -480,9 +482,7 @@ public abstract class TestOzoneRpcClientAbstract {
     OzoneVolume volume = store.getVolume(volumeName);
     volume.createBucket(bucketName);
     List<OzoneAcl> acls = new ArrayList<>();
-    acls.add(new OzoneAcl(
+    acls.add(new OzoneAcl(ACLIdentityType.USER, "test", ACLType.ALL));
-        OzoneAcl.OzoneACLType.USER, "test",
-        OzoneAcl.OzoneACLRights.READ_WRITE));
     OzoneBucket bucket = volume.getBucket(bucketName);
     bucket.addAcls(acls);
     OzoneBucket newBucket = volume.getBucket(bucketName);
@@ -495,8 +495,8 @@ public abstract class TestOzoneRpcClientAbstract {
       throws IOException, OzoneException {
     String volumeName = UUID.randomUUID().toString();
     String bucketName = UUID.randomUUID().toString();
-    OzoneAcl userAcl = new OzoneAcl(OzoneAcl.OzoneACLType.USER, "test",
+    OzoneAcl userAcl = new OzoneAcl(ACLIdentityType.USER, "test",
-        OzoneAcl.OzoneACLRights.READ_WRITE);
+        ACLType.ALL);
     List<OzoneAcl> acls = new ArrayList<>();
     acls.add(userAcl);
     store.createVolume(volumeName);

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/om/TestOzoneManager.java

@@ -55,6 +55,8 @@ import org.apache.hadoop.ozone.om.helpers.OmKeyInfo;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServicePort;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.VolumeList;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
 import org.apache.hadoop.ozone.util.OzoneVersionInfo;
 import org.apache.hadoop.ozone.web.handlers.BucketArgs;
 import org.apache.hadoop.ozone.web.handlers.KeyArgs;
@@ -356,30 +358,29 @@ public class TestOzoneManager {
     createVolumeArgs.setGroups(groupName);
     storageHandler.createVolume(createVolumeArgs);
 
-    OzoneAcl userAcl = new OzoneAcl(OzoneAcl.OzoneACLType.USER, userName,
+    OzoneAcl userAcl = new OzoneAcl(ACLIdentityType.USER, userName,
-        OzoneAcl.OzoneACLRights.READ_WRITE);
+        ACLType.READ);
     Assert.assertTrue(storageHandler.checkVolumeAccess(volumeName, userAcl));
-    OzoneAcl group = new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, groupName[0],
+    OzoneAcl group = new OzoneAcl(ACLIdentityType.GROUP, groupName[0],
-        OzoneAcl.OzoneACLRights.READ);
+        ACLType.READ);
     Assert.assertTrue(storageHandler.checkVolumeAccess(volumeName, group));
 
     // Create a different user and access should fail
     String falseUserName = "user" + RandomStringUtils.randomNumeric(5);
     OzoneAcl falseUserAcl =
-        new OzoneAcl(OzoneAcl.OzoneACLType.USER, falseUserName,
+        new OzoneAcl(ACLIdentityType.USER, falseUserName,
-            OzoneAcl.OzoneACLRights.READ_WRITE);
+            ACLType.ALL);
     Assert.assertFalse(storageHandler
         .checkVolumeAccess(volumeName, falseUserAcl));
     // Checking access with user name and Group Type should fail
-    OzoneAcl falseGroupAcl = new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, userName,
+    OzoneAcl falseGroupAcl = new OzoneAcl(ACLIdentityType.GROUP, userName,
-        OzoneAcl.OzoneACLRights.READ_WRITE);
+        ACLType.ALL);
     Assert.assertFalse(storageHandler
         .checkVolumeAccess(volumeName, falseGroupAcl));
 
     // Access for acl type world should also fail
     OzoneAcl worldAcl =
-        new OzoneAcl(OzoneAcl.OzoneACLType.WORLD, "",
+        new OzoneAcl(ACLIdentityType.WORLD, "", ACLType.READ);
-            OzoneAcl.OzoneACLRights.READ);
     Assert.assertFalse(storageHandler.checkVolumeAccess(volumeName, worldAcl));
 
     Assert.assertEquals(0, omMetrics.getNumVolumeCheckAccessFails());

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/ozShell/TestOzoneShell.java

@@ -44,8 +44,6 @@ import org.apache.hadoop.hdds.tracing.StringCodec;
 import org.apache.hadoop.hdfs.DFSUtil;
 import org.apache.hadoop.ozone.MiniOzoneCluster;
 import org.apache.hadoop.ozone.OzoneAcl;
-import org.apache.hadoop.ozone.OzoneAcl.OzoneACLRights;
-import org.apache.hadoop.ozone.OzoneAcl.OzoneACLType;
 import org.apache.hadoop.ozone.OzoneConsts;
 import org.apache.hadoop.ozone.client.OzoneBucket;
 import org.apache.hadoop.ozone.client.OzoneKey;
@@ -59,6 +57,8 @@ import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.exceptions.OMException.ResultCodes;
 import org.apache.hadoop.ozone.om.helpers.ServiceInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.ServicePort;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
 import org.apache.hadoop.ozone.web.ozShell.OzoneShell;
 import org.apache.hadoop.ozone.web.ozShell.Shell;
 import org.apache.hadoop.ozone.web.request.OzoneQuota;
@@ -744,8 +744,9 @@ public class TestOzoneShell {
 
     OzoneAcl acl = bucket.getAcls().get(aclSize);
     assertTrue(acl.getName().equals("frodo")
-        && acl.getType() == OzoneACLType.USER
+        && acl.getType() == ACLIdentityType.USER
-        && acl.getRights()== OzoneACLRights.READ_WRITE);
+        && acl.getRights().contains(ACLType.READ)
+        && acl.getRights().contains(ACLType.WRITE));
 
     args = new String[] {"bucket", "update",
         url + "/" + vol.getName() + "/" + bucketName, "--removeAcl",
@@ -756,8 +757,8 @@ public class TestOzoneShell {
     acl = bucket.getAcls().get(aclSize);
     assertEquals(1 + aclSize, bucket.getAcls().size());
     assertTrue(acl.getName().equals("samwise")
-        && acl.getType() == OzoneACLType.GROUP
+        && acl.getType() == ACLIdentityType.GROUP
-        && acl.getRights()== OzoneACLRights.READ);
+        && acl.getRights().contains(ACLType.READ));
 
     // test update bucket for a non-exist bucket
     args = new String[] {"bucket", "update",

hadoop-ozone/objectstore-service/src/main/java/org/apache/hadoop/ozone/web/storage/DistributedStorageHandler.java

@@ -44,8 +44,10 @@ import org.apache.hadoop.ozone.OzoneConsts.Versioning;
 import org.apache.hadoop.ozone.client.io.OzoneOutputStream;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos;
 import org.apache.hadoop.ozone.protocolPB.OMPBHelper;
-import org.apache.hadoop.ozone.om.OMConfigKeys;
 import org.apache.hadoop.ozone.OzoneAcl;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType;
+import org.apache.hadoop.ozone.security.acl.OzoneAclConfig;
 import org.apache.hadoop.ozone.web.request.OzoneQuota;
 import org.apache.hadoop.hdds.scm.ScmConfigKeys;
 import org.apache.hadoop.hdds.scm.XceiverClientManager;
@@ -80,8 +82,8 @@ public final class DistributedStorageHandler implements StorageHandler {
   private final OzoneManagerProtocol
       ozoneManagerClient;
   private final XceiverClientManager xceiverClientManager;
-  private final OzoneAcl.OzoneACLRights userRights;
+  private final ACLType userRights;
-  private final OzoneAcl.OzoneACLRights groupRights;
+  private final ACLType groupRights;
   private int chunkSize;
   private final long streamBufferFlushSize;
   private final long streamBufferMaxSize;
@@ -109,10 +111,10 @@ public final class DistributedStorageHandler implements StorageHandler {
 
     chunkSize = (int)conf.getStorageSize(ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_KEY,
         ScmConfigKeys.OZONE_SCM_CHUNK_SIZE_DEFAULT, StorageUnit.BYTES);
-    userRights = conf.getEnum(OMConfigKeys.OZONE_OM_USER_RIGHTS,
+    // Get default acl rights for user and group.
-        OMConfigKeys.OZONE_OM_USER_RIGHTS_DEFAULT);
+    OzoneAclConfig aclConfig = conf.getObject(OzoneAclConfig.class);
-    groupRights = conf.getEnum(OMConfigKeys.OZONE_OM_GROUP_RIGHTS,
+    this.userRights = aclConfig.getUserDefaultRights();
-        OMConfigKeys.OZONE_OM_GROUP_RIGHTS_DEFAULT);
+    this.groupRights = aclConfig.getGroupDefaultRights();
     if(chunkSize > OzoneConsts.OZONE_SCM_CHUNK_MAX_SIZE) {
       LOG.warn("The chunk size ({}) is not allowed to be more than"
               + " the maximum size ({}),"
@@ -176,8 +178,7 @@ public final class DistributedStorageHandler implements StorageHandler {
     long quota = args.getQuota() == null ?
         OzoneConsts.MAX_QUOTA_IN_BYTES : args.getQuota().sizeInBytes();
     OzoneAcl userAcl =
-        new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+        new OzoneAcl(ACLIdentityType.USER, args.getUserName(), userRights);
-            args.getUserName(), userRights);
     OmVolumeArgs.Builder builder = OmVolumeArgs.newBuilder();
     builder.setAdminName(args.getAdminName())
         .setOwnerName(args.getUserName())
@@ -187,7 +188,7 @@ public final class DistributedStorageHandler implements StorageHandler {
     if (args.getGroups() != null) {
       for (String group : args.getGroups()) {
         OzoneAcl groupAcl =
-            new OzoneAcl(OzoneAcl.OzoneACLType.GROUP, group, groupRights);
+            new OzoneAcl(ACLIdentityType.GROUP, group, groupRights);
         builder.addOzoneAcls(OMPBHelper.convertOzoneAcl(groupAcl));
       }
     }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/VolumeManagerImpl.java

@@ -20,6 +20,7 @@ import java.io.IOException;
 import java.util.ArrayList;
 import java.util.List;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.hdds.conf.OzoneConfiguration;
 import org.apache.hadoop.ozone.om.exceptions.OMException;
 import org.apache.hadoop.ozone.om.helpers.OmDeleteVolumeResponse;
@@ -484,7 +485,8 @@ public class VolumeManagerImpl implements VolumeManager {
     } catch (IOException ex) {
       if (!(ex instanceof OMException)) {
         LOG.error("Check volume access failed for volume:{} user:{} rights:{}",
-            volume, userAcl.getName(), userAcl.getRights(), ex);
+            volume, userAcl.getName(),
+            StringUtils.join(userAcl.getRightsList(), ","), ex);
       }
       throw ex;
     } finally {

hadoop-ozone/ozone-manager/src/test/java/org/apache/hadoop/ozone/om/TestBucketManagerImpl.java

@@ -42,6 +42,8 @@ import org.junit.runner.RunWith;
 import org.mockito.Mockito;
 import org.mockito.runners.MockitoJUnitRunner;
 
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.*;
+
 /**
  * Tests BucketManagerImpl, mocks OMMetadataManager for testing.
  */
@@ -217,8 +219,8 @@ public class TestBucketManagerImpl {
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
     List<OzoneAcl> acls = new LinkedList<>();
-    OzoneAcl ozoneAcl = new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+    OzoneAcl ozoneAcl = new OzoneAcl(ACLIdentityType.USER,
-        "root", OzoneAcl.OzoneACLRights.READ);
+        "root", ACLType.READ);
     acls.add(ozoneAcl);
     BucketManager bucketManager = new BucketManagerImpl(metaMgr);
     OmBucketInfo bucketInfo = OmBucketInfo.newBuilder()
@@ -235,8 +237,8 @@ public class TestBucketManagerImpl {
     Assert.assertEquals("bucketOne", result.getBucketName());
     Assert.assertEquals(1, result.getAcls().size());
     List<OzoneAcl> addAcls = new LinkedList<>();
-    OzoneAcl newAcl = new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+    OzoneAcl newAcl = new OzoneAcl(ACLIdentityType.USER,
-        "ozone", OzoneAcl.OzoneACLRights.READ);
+        "ozone", ACLType.READ);
     addAcls.add(newAcl);
     OmBucketArgs bucketArgs = OmBucketArgs.newBuilder()
         .setVolumeName("sampleVol")
@@ -256,10 +258,10 @@ public class TestBucketManagerImpl {
     OmMetadataManagerImpl metaMgr = createSampleVol();
 
     List<OzoneAcl> acls = new LinkedList<>();
-    OzoneAcl aclOne = new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+    OzoneAcl aclOne = new OzoneAcl(ACLIdentityType.USER,
-        "root", OzoneAcl.OzoneACLRights.READ);
+        "root", ACLType.READ);
-    OzoneAcl aclTwo = new OzoneAcl(OzoneAcl.OzoneACLType.USER,
+    OzoneAcl aclTwo = new OzoneAcl(ACLIdentityType.USER,
-        "ozone", OzoneAcl.OzoneACLRights.READ);
+        "ozone", ACLType.READ);
     acls.add(aclOne);
     acls.add(aclTwo);
     BucketManager bucketManager = new BucketManagerImpl(metaMgr);

hadoop-ozone/tools/src/main/java/org/apache/hadoop/ozone/scm/cli/SQLCLI.java

@@ -378,7 +378,7 @@ public class SQLCLI  extends Configured implements Tool {
       for (OzoneAclInfo aclInfo : volumeInfo.getVolumeAclsList()) {
         String insertAclInfo =
             String.format(INSERT_ACL_INFO, adminName, ownerName, volumeName,
-                aclInfo.getType(), aclInfo.getName(), aclInfo.getRights());
+                aclInfo.getType(), aclInfo.getName(), aclInfo.getRightsList());
         executeSQL(conn, insertAclInfo);
       }
       break;