HDDS-2181. Ozone Manager should send correct ACL type in ACL requests…

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

@@ -52,6 +52,8 @@
 
 import static org.apache.hadoop.hdds.HddsUtils.getHostNameFromConfigKeys;
 import static org.apache.hadoop.hdds.HddsUtils.getPortNumberFromConfigKeys;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS;
+import static org.apache.hadoop.ozone.OzoneConfigKeys.OZONE_ACL_AUTHORIZER_CLASS_NATIVE;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_ADDRESS_KEY;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_BIND_HOST_DEFAULT;
 import static org.apache.hadoop.ozone.om.OMConfigKeys.OZONE_OM_HTTPS_ADDRESS_KEY;
@@ -530,4 +532,16 @@ public static RepeatedOmKeyInfo prepareKeyForDelete(OmKeyInfo keyInfo,
 
     return repeatedOmKeyInfo;
   }
+
+  /**
+   * Returns true if OzoneNativeAuthorizer is configured in the configuration.
+   * @param configuration ozone configuration
+   * @return true if OzoneNativeAuthorizer is configured in the configuration;
+   * else false.
+   */
+  public static boolean isNativeAuthorizerEnabled(Configuration configuration) {
+    String authorizer = configuration.get(OZONE_ACL_AUTHORIZER_CLASS);
+    return authorizer != null &&
+        authorizer.equals(OZONE_ACL_AUTHORIZER_CLASS_NATIVE);
+  }
 }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketCreateRequest.java

@@ -143,7 +143,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     try {
       // check Acl
       if (ozoneManager.getAclsEnabled()) {
-        checkAcls(ozoneManager, OzoneObj.ResourceType.VOLUME,
+        checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET,
             OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.CREATE,
             volumeName, bucketName, null);
       }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/bucket/OMBucketDeleteRequest.java

@@ -93,7 +93,7 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       // check Acl
       if (ozoneManager.getAclsEnabled()) {
         checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET,
-            OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE,
+            OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.DELETE,
             volumeName, bucketName, null);
       }
 

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMDirectoryCreateRequest.java

@@ -32,6 +32,7 @@
 import org.apache.hadoop.ozone.om.helpers.OzoneAclUtil;
 import org.apache.hadoop.ozone.om.helpers.OzoneFSUtils;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -127,7 +128,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     OMClientResponse omClientResponse = null;
     try {
       // check Acl
-      checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, keyName,
+          IAccessAuthorizer.ACLType.CREATE);
 
       // Check if this is the root of the filesystem.
       if (keyName.length() == 0) {

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/file/OMFileCreateRequest.java

@@ -31,6 +31,7 @@
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -177,7 +178,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     OMClientResponse omClientResponse = null;
     try {
       // check Acl
-      checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, keyName,
+          IAccessAuthorizer.ACLType.CREATE);
 
       // acquire lock
       acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMAllocateBlockRequest.java

@@ -26,6 +26,7 @@
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.util.Time;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -169,7 +170,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     OmKeyInfo omKeyInfo = null;
     try {
       // check Acl
-      checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, keyName,
+          IAccessAuthorizer.ACLType.WRITE);
 
       OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
       validateBucketAndVolume(omMetadataManager, volumeName,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.java

@@ -25,7 +25,11 @@
 
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.hdds.conf.OzoneConfiguration;
+import org.apache.hadoop.ozone.OmUtils;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -111,11 +115,21 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     IOException exception = null;
     OmKeyInfo omKeyInfo = null;
     OMClientResponse omClientResponse = null;
+    boolean bucketLockAcquired = false;
 
     OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
     try {
       // check Acl
-      checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
+      // Native authorizer requires client id as part of keyname to check
+      // write ACL on key. Add client id to key name if ozone native
+      // authorizer is configured.
+      Configuration config = new OzoneConfiguration();
+      String keyNameForAclCheck = keyName;
+      if (OmUtils.isNativeAuthorizerEnabled(config)) {
+        keyNameForAclCheck = keyName + "/" + commitKeyRequest.getClientID();
+      }
+      checkKeyAcls(ozoneManager, volumeName, bucketName, keyNameForAclCheck,
+          IAccessAuthorizer.ACLType.WRITE);
 
       List<OmKeyLocationInfo> locationInfoList = commitKeyArgs
           .getKeyLocationsList().stream()
@@ -127,8 +141,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       String dbOpenKey = omMetadataManager.getOpenKey(volumeName, bucketName,
           keyName, commitKeyRequest.getClientID());
 
-      omMetadataManager.getLock().acquireLock(BUCKET_LOCK, volumeName,
-          bucketName);
+      bucketLockAcquired = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+          volumeName, bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
       omKeyInfo = omMetadataManager.getOpenKeyTable().get(dbOpenKey);
@@ -166,8 +180,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
             ozoneManagerDoubleBufferHelper.add(omClientResponse,
                 transactionLogIndex));
       }
-      omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
-          bucketName);
+
+      if(bucketLockAcquired) {
+        omMetadataManager.getLock().releaseLock(BUCKET_LOCK, volumeName,
+            bucketName);
+      }
     }
 
     // Performing audit logging outside of the lock.

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCreateRequest.java

@@ -26,6 +26,7 @@
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -162,7 +163,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     OMClientResponse omClientResponse = null;
     try {
       // check Acl
-      checkBucketAcls(ozoneManager, volumeName, bucketName, keyName);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, keyName,
+          IAccessAuthorizer.ACLType.CREATE);
 
       acquireLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
           volumeName, bucketName);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyDeleteRequest.java

@@ -23,6 +23,7 @@
 
 import com.google.common.base.Optional;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -109,7 +110,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
     OMClientResponse omClientResponse = null;
     try {
       // check Acl
-      checkKeyAcls(ozoneManager, volumeName, bucketName, keyName);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, keyName,
+          IAccessAuthorizer.ACLType.DELETE);
 
       String objectKey = omMetadataManager.getOzoneKey(
           volumeName, bucketName, keyName);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRenameRequest.java

@@ -24,6 +24,7 @@
 import com.google.common.base.Optional;
 import com.google.common.base.Preconditions;
 import org.apache.hadoop.ozone.om.ratis.utils.OzoneManagerDoubleBufferHelper;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -117,8 +118,12 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
         throw new OMException("Key name is empty",
             OMException.ResultCodes.INVALID_KEY_NAME);
       }
-      // check Acl
-      checkKeyAcls(ozoneManager, volumeName, bucketName, fromKeyName);
+      // check Acls to see if user has access to perform delete operation on
+      // old key and create operation on new key
+      checkKeyAcls(ozoneManager, volumeName, bucketName, fromKeyName,
+          IAccessAuthorizer.ACLType.DELETE);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, toKeyName,
+          IAccessAuthorizer.ACLType.CREATE);
 
       acquiredLock = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
           volumeName, bucketName);

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyRequest.java

@@ -507,10 +507,11 @@ private OMClientResponse createKeyErrorResponse(@Nonnull OMMetrics omMetrics,
    * @throws IOException
    */
   protected void checkBucketAcls(OzoneManager ozoneManager, String volume,
-      String bucket, String key) throws IOException {
+      String bucket, String key, IAccessAuthorizer.ACLType aclType)
+      throws IOException {
     if (ozoneManager.getAclsEnabled()) {
       checkAcls(ozoneManager, OzoneObj.ResourceType.BUCKET,
-          OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE,
+          OzoneObj.StoreType.OZONE, aclType,
           volume, bucket, key);
     }
   }
@@ -525,10 +526,11 @@ protected void checkBucketAcls(OzoneManager ozoneManager, String volume,
    * @throws IOException
    */
   protected void checkKeyAcls(OzoneManager ozoneManager, String volume,
-      String bucket, String key) throws IOException {
+      String bucket, String key, IAccessAuthorizer.ACLType aclType)
+      throws IOException {
     if (ozoneManager.getAclsEnabled()) {
       checkAcls(ozoneManager, OzoneObj.ResourceType.KEY,
-          OzoneObj.StoreType.OZONE, IAccessAuthorizer.ACLType.WRITE,
+          OzoneObj.StoreType.OZONE, aclType,
           volume, bucket, key);
     }
   }

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/security/acl/OzoneNativeAuthorizer.java

@@ -69,6 +69,9 @@ public boolean checkAccess(IOzoneObj ozObject, RequestContext context)
     Objects.requireNonNull(ozObject);
     Objects.requireNonNull(context);
     OzoneObjInfo objInfo;
+    RequestContext parentContext;
+    boolean isACLTypeCreate = (context.getAclRights() == ACLType.CREATE);
+    boolean isACLTypeDelete = (context.getAclRights() == ACLType.DELETE);
 
     if (ozObject instanceof OzoneObjInfo) {
       objInfo = (OzoneObjInfo) ozObject;
@@ -77,25 +80,52 @@ public boolean checkAccess(IOzoneObj ozObject, RequestContext context)
           "configured to work with OzoneObjInfo type only.", INVALID_REQUEST);
     }
 
+    // For CREATE and DELETE acl requests, the parents need to be checked
+    // for WRITE acl. If Key create request is received, then we need to
+    // check if user has WRITE acl set on Bucket and Volume. In all other cases
+    // the parents also need to be checked for the same acl type.
+    if (isACLTypeCreate || isACLTypeDelete) {
+      parentContext = RequestContext.newBuilder()
+        .setClientUgi(context.getClientUgi())
+        .setIp(context.getIp())
+        .setAclType(context.getAclType())
+        .setAclRights(ACLType.WRITE)
+        .build();
+    } else {
+      parentContext = context;
+    }
+
     switch (objInfo.getResourceType()) {
     case VOLUME:
       LOG.trace("Checking access for volume:" + objInfo);
       return volumeManager.checkAccess(objInfo, context);
     case BUCKET:
       LOG.trace("Checking access for bucket:" + objInfo);
-      return (bucketManager.checkAccess(objInfo, context)
-          && volumeManager.checkAccess(objInfo, context));
+      // Skip bucket access check for CREATE acl since
+      // bucket will not exist at the time of creation
+      boolean bucketAccess = isACLTypeCreate
+          || bucketManager.checkAccess(objInfo, context);
+      return (bucketAccess
+          && volumeManager.checkAccess(objInfo, parentContext));
     case KEY:
       LOG.trace("Checking access for Key:" + objInfo);
-      return (keyManager.checkAccess(objInfo, context)
-          && prefixManager.checkAccess(objInfo, context)
-          && bucketManager.checkAccess(objInfo, context)
-          && volumeManager.checkAccess(objInfo, context));
+      // Skip key access check for CREATE acl since
+      // key will not exist at the time of creation
+      boolean keyAccess = isACLTypeCreate
+          || keyManager.checkAccess(objInfo, context);
+      return (keyAccess
+          && prefixManager.checkAccess(objInfo, parentContext)
+          && bucketManager.checkAccess(objInfo, parentContext)
+          && volumeManager.checkAccess(objInfo, parentContext));
     case PREFIX:
       LOG.trace("Checking access for Prefix:" + objInfo);
-      return (prefixManager.checkAccess(objInfo, context)
-          && bucketManager.checkAccess(objInfo, context)
-          && volumeManager.checkAccess(objInfo, context));
+      // Skip prefix access check for CREATE acl since
+      // prefix will not exist at the time of creation
+      boolean prefixAccess = isACLTypeCreate
+          || prefixManager.checkAccess(objInfo, context);
+      return (prefixAccess
+          && bucketManager.checkAccess(objInfo, parentContext)
+          && volumeManager.checkAccess(objInfo, parentContext));
     default:
       throw new OMException("Unexpected object type:" +
           objInfo.getResourceType(), INVALID_REQUEST);