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

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

@@ -340,7 +340,7 @@ private OMResponse submitRequest(OMRequest omRequest)
       if (omResponse.hasLeaderOMNodeId() && omFailoverProxyProvider != null) {
         String leaderOmId = omResponse.getLeaderOMNodeId();
 
-        // Failover to the OM node returned by OMReponse leaderOMNodeId if
+        // Failover to the OM node returned by OMResponse leaderOMNodeId if
         // current proxy is not pointing to that node.
         omFailoverProxyProvider.performFailoverIfRequired(leaderOmId);
       }

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

@@ -64,7 +64,7 @@ public static int getNoOfAcls() {
 
     public static ACLType getAclTypeFromOrdinal(int ordinal) {
       if (ordinal > length - 1 && ordinal > -1) {
-        throw new IllegalArgumentException("Ordinal greater than array lentgh" +
+        throw new IllegalArgumentException("Ordinal greater than array length" +
             ". ordinal:" + ordinal);
       }
       return vals[ordinal];

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

@@ -118,7 +118,7 @@ public void testBucketCreationPermissionDenied() throws Exception {
         () -> volume.createBucket(bucketName));
 
     assertTrue(logCapturer.getOutput()
-        .contains("doesn't have CREATE permission to access volume"));
+        .contains("doesn't have CREATE permission to access bucket"));
   }
 
   @Test
@@ -133,8 +133,8 @@ public void testFailureInKeyOp() throws Exception {
 
     OzoneTestUtils.expectOmException(ResultCodes.PERMISSION_DENIED,
         () -> TestDataUtil.createKey(bucket, "testKey", "testcontent"));
-    assertTrue(logCapturer.getOutput().contains("doesn't have WRITE " +
-        "permission to access bucket"));
+    assertTrue(logCapturer.getOutput().contains("doesn't have CREATE " +
+        "permission to access key"));
   }
 
   /**

hadoop-ozone/integration-test/src/test/java/org/apache/hadoop/ozone/security/acl/TestOzoneNativeAuthorizer.java

@@ -69,6 +69,8 @@
 import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType.USER;
 import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLIdentityType.WORLD;
 import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.ALL;
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.CREATE;
+import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.WRITE;
 import static org.apache.hadoop.ozone.security.acl.IAccessAuthorizer.ACLType.NONE;
 import static org.apache.hadoop.ozone.security.acl.OzoneObj.ResourceType.BUCKET;
 import static org.apache.hadoop.ozone.security.acl.OzoneObj.ResourceType.KEY;
@@ -107,6 +109,7 @@ public class TestOzoneNativeAuthorizer {
   private static OzoneObj buckObj;
   private static OzoneObj keyObj;
   private static OzoneObj prefixObj;
+  private static long keySessionId;
 
   @Parameterized.Parameters
   public static Collection<Object[]> data() {
@@ -192,6 +195,7 @@ private void createKey(String volume,
           keySession.getKeyInfo().getLatestVersionLocations()
               .getLocationList());
       keyManager.commitKey(keyArgs, keySession.getId());
+      keySessionId = keySession.getId();
     }
 
     keyObj = new OzoneObjInfo.Builder()
@@ -358,10 +362,21 @@ private void resetAclsAndValidateAccess(OzoneObj obj,
       List<ACLType> aclsToBeAdded =
           Arrays.stream(ACLType.values()).collect(Collectors.toList());
       aclsToBeValidated.remove(NONE);
+      // Do not validate "WRITE" since write acl type requires object to be
+      // present in OpenKeyTable.
+      aclsToBeValidated.remove(WRITE);
       aclsToBeValidated.remove(a1);
 
       aclsToBeAdded.remove(NONE);
       aclsToBeAdded.remove(ALL);
+      // AclType "CREATE" is skipped from access check on objects
+      // since the object will not exist during access check.
+      aclsToBeAdded.remove(CREATE);
+      // AclType "WRITE" is removed from being tested here,
+      // because object must always be present in OpenKeyTable for write
+      // acl requests. But, here the objects are already committed
+      // and will move to keyTable.
+      aclsToBeAdded.remove(WRITE);
 
       // Fetch acls again.
       for (ACLType a2 : aclsToBeAdded) {
@@ -370,7 +385,7 @@ private void resetAclsAndValidateAccess(OzoneObj obj,
           acls = aclImplementor.getAcl(obj);
           List right = acls.stream().map(a -> a.getAclList()).collect(
               Collectors.toList());
-          assertFalse("Do not expected client to have " + a2 + " acl. " +
+          assertFalse("Did not expect client to have " + a2 + " acl. " +
                   "Current acls found:" + right + ". Type:" + accessType + ","
                   + " name:" + (accessType == USER ? user : group),
               nativeAuthorizer.checkAccess(obj,
@@ -410,7 +425,7 @@ private void resetAclsAndValidateAccess(OzoneObj obj,
                   builder.setAclRights(a2).build()));
           aclsToBeValidated.remove(a2);
           for (ACLType a3 : aclsToBeValidated) {
-            if (!a3.equals(a1) && !a3.equals(a2)) {
+            if (!a3.equals(a1) && !a3.equals(a2) && !a3.equals(CREATE)) {
               assertFalse("User shouldn't have right " + a3 + ". " +
                       "Current acl rights for user:" + a1 + "," + a2,
                   nativeAuthorizer.checkAccess(obj,
@@ -420,7 +435,6 @@ private void resetAclsAndValidateAccess(OzoneObj obj,
         }
       }
     }
-
   }
 
   private String getAclName(ACLIdentityType identityType) {
@@ -462,6 +476,9 @@ private void validateNone(OzoneObj obj, RequestContext.Builder
       builder) throws OMException {
     List<ACLType> allAcls = new ArrayList<>(Arrays.asList(ACLType.values()));
     allAcls.remove(NONE);
+    // Removing CREATE, WRITE since they need special handling.
+    allAcls.remove(CREATE);
+    allAcls.remove(WRITE);
     for (ACLType a : allAcls) {
       assertFalse("User shouldn't have right " + a + ".", 
           nativeAuthorizer.checkAccess(obj, builder.setAclRights(a).build()));

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

@@ -89,6 +89,7 @@
 import org.apache.hadoop.ozone.om.helpers.RepeatedOmKeyInfo;
 import org.apache.hadoop.ozone.protocol.proto.OzoneManagerProtocolProtos.PartKeyInfo;
 import org.apache.hadoop.ozone.security.OzoneBlockTokenSecretManager;
+import org.apache.hadoop.ozone.security.acl.IAccessAuthorizer;
 import org.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.apache.hadoop.ozone.security.acl.RequestContext;
 import org.apache.hadoop.security.SecurityUtil;
@@ -1654,28 +1655,28 @@ public boolean checkAccess(OzoneObj ozObject, RequestContext context)
     metadataManager.getLock().acquireReadLock(BUCKET_LOCK, volume, bucket);
     try {
       validateBucket(volume, bucket);
-      OmKeyInfo keyInfo = null;
-      try {
-        OzoneFileStatus fileStatus = getFileStatus(args);
-        keyInfo = fileStatus.getKeyInfo();
-        if (keyInfo == null) {
-          // the key does not exist, but it is a parent "dir" of some key
-          // let access be determined based on volume/bucket/prefix ACL
-          if (LOG.isDebugEnabled()) {
-            LOG.debug("key:{} is non-existent parent, permit access to user:{}",
-                keyName, context.getClientUgi());
-          }
-          return true;
-        }
-      } catch (OMException e) {
-        if (e.getResult() == FILE_NOT_FOUND) {
-          keyInfo = metadataManager.getOpenKeyTable().get(objectKey);
+      OmKeyInfo keyInfo;
+
+      // For Acl Type "WRITE", the key can only be found in
+      // OpenKeyTable since appends to existing keys are not supported.
+      if (context.getAclRights() == IAccessAuthorizer.ACLType.WRITE) {
+        keyInfo = metadataManager.getOpenKeyTable().get(objectKey);
+      } else {
+        try {
+          OzoneFileStatus fileStatus = getFileStatus(args);
+          keyInfo = fileStatus.getKeyInfo();
+        } catch (IOException e) {
+          throw new OMException("Key not found, checkAccess failed. Key:" +
+              objectKey, KEY_NOT_FOUND);
         }
       }
 
       if (keyInfo == null) {
-        throw new OMException("Key not found, checkAccess failed. Key:" +
-            objectKey, KEY_NOT_FOUND);
+        // the key does not exist, but it is a parent "dir" of some key
+        // let access be determined based on volume/bucket/prefix ACL
+        LOG.debug("key:{} is non-existent parent, permit access to user:{}",
+            keyName, context.getClientUgi());
+        return true;
       }
 
       boolean hasAccess = OzoneAclUtil.checkAclRight(

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

@@ -301,6 +301,8 @@ public final class OzoneManager extends ServiceRuntimeInfoImpl
   private final boolean grpcBlockTokenEnabled;
   private final boolean useRatisForReplication;
 
+  private boolean isNativeAuthorizerEnabled;
+
   private OzoneManager(OzoneConfiguration conf) throws IOException,
       AuthenticationException {
     super(OzoneVersionInfo.OZONE_VERSION_INFO);
@@ -473,6 +475,7 @@ private void instantiateServices() throws IOException {
       if (accessAuthorizer instanceof OzoneNativeAuthorizer) {
         OzoneNativeAuthorizer authorizer =
             (OzoneNativeAuthorizer) accessAuthorizer;
+        isNativeAuthorizerEnabled = true;
         authorizer.setVolumeManager(volumeManager);
         authorizer.setBucketManager(bucketManager);
         authorizer.setKeyManager(keyManager);
@@ -3292,4 +3295,11 @@ public Collection<String> getOzoneAdmins() {
     return ozAdmins;
   }
 
+  /**
+   * Returns true if OzoneNativeAuthorizer is enabled and false if otherwise.
+   * @return if native authorizer is enabled.
+   */
+  public boolean isNativeAuthorizerEnabled() {
+    return isNativeAuthorizerEnabled;
+  }
 }

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

@@ -95,7 +95,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,8 @@
 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.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -126,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, OzoneObj.ResourceType.KEY);
 
       // 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,8 @@
 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.ozone.security.acl.OzoneObj;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -177,7 +179,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, OzoneObj.ResourceType.KEY);
 
       // acquire lock
       acquiredLock = omMetadataManager.getLock().acquireWriteLock(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);
+      checkKeyAclsInOpenKeyTable(ozoneManager, volumeName, bucketName, keyName,
+          IAccessAuthorizer.ACLType.WRITE, allocateBlockRequest.getClientID());
 
       OMMetadataManager omMetadataManager = ozoneManager.getMetadataManager();
       validateBucketAndVolume(omMetadataManager, volumeName,

hadoop-ozone/ozone-manager/src/main/java/org/apache/hadoop/ozone/om/request/key/OMKeyCommitRequest.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;
 
@@ -111,11 +112,14 @@ 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);
+      checkKeyAclsInOpenKeyTable(ozoneManager, volumeName, bucketName,
+          keyName, IAccessAuthorizer.ACLType.WRITE,
+          commitKeyRequest.getClientID());
 
       List<OmKeyLocationInfo> locationInfoList = commitKeyArgs
           .getKeyLocationsList().stream()
@@ -127,8 +131,8 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
       String dbOpenKey = omMetadataManager.getOpenKey(volumeName, bucketName,
           keyName, commitKeyRequest.getClientID());
 
-      omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK, volumeName,
-          bucketName);
+      bucketLockAcquired = omMetadataManager.getLock().acquireLock(BUCKET_LOCK,
+          volumeName, bucketName);
 
       validateBucketAndVolume(omMetadataManager, volumeName, bucketName);
       omKeyInfo = omMetadataManager.getOpenKeyTable().get(dbOpenKey);
@@ -166,8 +170,11 @@ public OMClientResponse validateAndUpdateCache(OzoneManager ozoneManager,
             ozoneManagerDoubleBufferHelper.add(omClientResponse,
                 transactionLogIndex));
       }
-      omMetadataManager.getLock().releaseWriteLock(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,8 @@
 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.ozone.security.acl.OzoneObj;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -162,7 +164,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, OzoneObj.ResourceType.KEY);
 
       acquireLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);

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

@@ -23,6 +23,8 @@
 
 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.apache.hadoop.ozone.security.acl.OzoneObj;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -108,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, OzoneObj.ResourceType.KEY);
 
       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,8 @@
 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.ozone.security.acl.OzoneObj;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -117,8 +119,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, OzoneObj.ResourceType.KEY);
+      checkKeyAcls(ozoneManager, volumeName, bucketName, toKeyName,
+          IAccessAuthorizer.ACLType.CREATE, OzoneObj.ResourceType.KEY);
 
       acquiredLock = omMetadataManager.getLock().acquireWriteLock(BUCKET_LOCK,
           volumeName, bucketName);