HADOOP-14935. Azure: POSIX permissions are taking effect in access() method even when authorization is enabled.

Contributed by Santhosh G Nayak

Cherry-picked from commit 9fcc3a1fc8cab873034f5c308ceb2d5671a954e8

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/NativeAzureFileSystem.java

@@ -67,9 +67,11 @@ import org.apache.hadoop.fs.azure.metrics.AzureFileSystemMetricsSystem;
 import org.apache.hadoop.fs.azure.security.Constants;
 import org.apache.hadoop.fs.azure.security.RemoteWasbDelegationTokenManager;
 import org.apache.hadoop.fs.azure.security.WasbDelegationTokenManager;
+import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.fs.permission.PermissionStatus;
 import org.apache.hadoop.io.IOUtils;
+import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.token.delegation.web.DelegationTokenAuthenticatedURL;
@@ -2658,22 +2660,6 @@ public class NativeAzureFileSystem extends FileSystem {
   public FileStatus getFileStatus(Path f) throws FileNotFoundException, IOException {
 
     LOG.debug("Getting the file status for {}", f.toString());
-
-    // Capture the absolute path and the path to key.
-    Path absolutePath = makeAbsolute(f);
-
-    if (!isRenamePendingFile(absolutePath)) {
-      Path ancestor = getAncestor(absolutePath);
-      if (ancestor.equals(absolutePath) && !ancestor.equals(new Path("/"))) {
-        performAuthCheck(ancestor.getParent(), WasbAuthorizationOperations.READ,
-            "getFileStatus", absolutePath);
-      }
-      else {
-        performAuthCheck(ancestor, WasbAuthorizationOperations.READ,
-            "getFileStatus", absolutePath);
-      }
-    }
-
     return getFileStatusInternal(f);
   }
 
@@ -2701,6 +2687,14 @@ public class NativeAzureFileSystem extends FileSystem {
     }
   }
 
+  /**
+   * Inner implementation of {@link #getFileStatus(Path)}.
+   * Return a file status object that represents the path.
+   * @param f The path we want information from
+   * @return a FileStatus object
+   * @throws FileNotFoundException when the path does not exist
+   * @throws IOException Other failure
+   */
   private FileStatus getFileStatusInternal(Path f) throws FileNotFoundException, IOException {
 
     Path absolutePath = makeAbsolute(f);
@@ -3725,6 +3719,41 @@ public class NativeAzureFileSystem extends FileSystem {
     }
   }
 
+  @Override
+  public void access(Path path, FsAction mode) throws IOException {
+    if (azureAuthorization && authorizer != null) {
+      try {
+        // Required to check the existence of the path.
+        getFileStatus(path);
+        switch (mode) {
+        case READ:
+        case READ_EXECUTE:
+          performAuthCheck(path, WasbAuthorizationOperations.READ, "access", path);
+          break;
+        case WRITE:
+        case WRITE_EXECUTE:
+          performAuthCheck(path, WasbAuthorizationOperations.WRITE, "access",
+              path);
+          break;
+        case READ_WRITE:
+        case ALL:
+          performAuthCheck(path, WasbAuthorizationOperations.READ, "access", path);
+          performAuthCheck(path, WasbAuthorizationOperations.WRITE, "access",
+              path);
+          break;
+        case EXECUTE:
+        case NONE:
+        default:
+          break;
+        }
+      } catch (WasbAuthorizationException wae){
+        throw new AccessControlException(wae);
+      }
+    } else {
+      super.access(path, mode);
+    }
+  }
+
   /**
    * A handler that defines what to do with blobs whose upload was
    * interrupted.

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/security/Constants.java

@@ -34,6 +34,7 @@ public final class Constants {
    */
   public static final String AZURE_ENABLE_SPNEGO_TOKEN_CACHE =
       "fs.azure.enable.spnego.token.cache";
+
   /**
    * Parameter to be used for impersonation.
    */

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/TestNativeAzureFileSystemAuthorization.java

@@ -24,6 +24,7 @@ import java.io.IOException;
 import java.util.Arrays;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.Callable;
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataInputStream;
@@ -31,9 +32,12 @@ import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileStatus;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.contract.ContractTestUtils;
+import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
+import org.apache.hadoop.security.AccessControlException;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.test.GenericTestUtils;
+import org.apache.hadoop.test.LambdaTestUtils;
 import org.apache.hadoop.util.StringUtils;
 
 import org.junit.Assume;
@@ -148,7 +152,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -174,7 +177,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -202,7 +204,6 @@ public class TestNativeAzureFileSystemAuthorization
     setExpectedFailureMessage("create", testPath);
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -228,7 +229,6 @@ public class TestNativeAzureFileSystemAuthorization
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(testPath.toString(), WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -334,7 +334,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     /* for rename */
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -365,7 +364,6 @@ public class TestNativeAzureFileSystemAuthorization
     /* to create parent dir */
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, false);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -398,8 +396,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true); /* to create parent dir */
     authorizer.addAuthRuleForOwner(parentSrcDir.toString(), WRITE, true);
     authorizer.addAuthRuleForOwner(parentDstDir.toString(), WRITE, false);
-    authorizer.addAuthRuleForOwner(parentSrcDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(parentDstDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -429,8 +425,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true); /* to create parent dirs */
     authorizer.addAuthRuleForOwner(parentSrcDir.toString(), WRITE, true);
     authorizer.addAuthRuleForOwner(parentDstDir.toString(), WRITE, true);
-    authorizer.addAuthRuleForOwner(parentSrcDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(parentDstDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -608,9 +602,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner(parentSrcDir.toString(), WRITE, true);
     authorizer.addAuthRuleForOwner(parentDstDir.toString(), WRITE, true);
     /* Required for asserPathExists calls */
-    authorizer.addAuthRuleForOwner("/", READ, true);
-    authorizer.addAuthRuleForOwner(parentSrcDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(parentDstDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -651,9 +642,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner(parentSrcDir.toString(),
         WRITE, true);
     /* Required for asserPathExists calls */
-    authorizer.addAuthRuleForOwner("/", READ, true);
-    authorizer.addAuthRuleForOwner(parentSrcDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(parentDstDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -675,10 +663,6 @@ public class TestNativeAzureFileSystemAuthorization
             WRITE, getCurrentUserShortName(), true);
           authorizer.addAuthRule(parentDstDir.toString(),
             WRITE, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(parentSrcDir.toString(),
-            READ, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(parentDstDir.toString(),
-            READ, getCurrentUserShortName(), true);
 
           try {
             fs.rename(srcPath, dstPath);
@@ -714,9 +698,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner(parentSrcDir.toString(),
         WRITE, true);
     /* Required for asserPathExists calls */
-    authorizer.addAuthRuleForOwner("/", READ, true);
-    authorizer.addAuthRuleForOwner(parentSrcDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(parentDstDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -738,10 +719,6 @@ public class TestNativeAzureFileSystemAuthorization
             WRITE, getCurrentUserShortName(), true);
           authorizer.addAuthRule(parentDstDir.toString(),
             WRITE, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(parentSrcDir.toString(),
-            READ, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(parentDstDir.toString(),
-            READ, getCurrentUserShortName(), true);
           // should return false since srcPath does not exist.
           assertRenameOutcome(fs, srcPath, dstPath, false);
           assertPathDoesNotExist(fs, "destPath exists", dstPath);
@@ -843,7 +820,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
     try {
       fs.create(testPath);
@@ -868,7 +844,6 @@ public class TestNativeAzureFileSystemAuthorization
     setExpectedFailureMessage("delete", testPath);
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
     try {
       fs.create(testPath);
@@ -878,7 +853,6 @@ public class TestNativeAzureFileSystemAuthorization
       /* Remove permissions for delete to force failure */
       authorizer.deleteAllAuthRules();
       authorizer.addAuthRuleForOwner("/", WRITE, false);
-      authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
       fs.updateWasbAuthorizer(authorizer);
 
       fs.delete(testPath, false);
@@ -887,7 +861,6 @@ public class TestNativeAzureFileSystemAuthorization
       /* Restore permissions to force a successful delete */
       authorizer.deleteAllAuthRules();
       authorizer.addAuthRuleForOwner("/", WRITE, true);
-      authorizer.addAuthRuleForOwner("/", READ, true);
       fs.updateWasbAuthorizer(authorizer);
 
       fs.delete(testPath, false);
@@ -908,8 +881,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path(childPath, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true); // for create and delete
-    authorizer.addAuthRuleForOwner(childPath.toString(), READ, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     authorizer.addAuthRuleForOwner("/testDeleteIntermediateFolder*",
         WRITE, true); // for recursive delete
     fs.updateWasbAuthorizer(authorizer);
@@ -942,8 +913,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner("/testDeleteAuthCheckFailureLeavesFilesUndeleted*",
         WRITE, true);
-    authorizer.addAuthRuleForOwner(childPath1.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(childPath2.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -958,9 +927,6 @@ public class TestNativeAzureFileSystemAuthorization
       authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
       authorizer.addAuthRuleForOwner(childPath2.toString(), WRITE, true);
       authorizer.addAuthRuleForOwner(childPath1.toString(), WRITE, false);
-      authorizer.addAuthRuleForOwner(childPath1.toString(), READ, true);
-      authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
-      authorizer.addAuthRuleForOwner("/", READ, true);
 
       assertFalse(fs.delete(parentDir, true));
 
@@ -989,7 +955,6 @@ public class TestNativeAzureFileSystemAuthorization
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1026,7 +991,6 @@ public class TestNativeAzureFileSystemAuthorization
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1044,8 +1008,6 @@ public class TestNativeAzureFileSystemAuthorization
           try {
             authorizer.addAuthRule(parentDir.toString(), WRITE,
                 getCurrentUserShortName(), true);
-            authorizer.addAuthRule(parentDir.toString(), READ,
-                getCurrentUserShortName(), true);
             fs.delete(testPath, true);
             return null;
           }
@@ -1078,8 +1040,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner("/testRecursiveDeleteSucceedsWithStickybit*",
         WRITE, true);
-    authorizer.addAuthRuleForOwner(childDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1116,8 +1076,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner("/testRecursiveDeleteFailsWithStickybit*",
         WRITE, true);
-    authorizer.addAuthRuleForOwner(childDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1173,7 +1131,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner(
         "/testDeleteSucceedsForOnlyFilesOwnedByUserWithStickybitSet*",
         WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1188,11 +1145,8 @@ public class TestNativeAzureFileSystemAuthorization
         @Override
         public Void run() throws Exception {
           authorizer.addAuthRule("/", WRITE, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(parentDir.toString(), READ, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(testFolderPath.toString(), READ, getCurrentUserShortName(), true);
           authorizer.addAuthRule("/testDeleteSucceedsForOnlyFilesOwnedByUserWithStickybitSet*",
               WRITE, getCurrentUserShortName(), true);
-          authorizer.addAuthRule("/", READ, getCurrentUserShortName(), true);
 
           fs.create(testFolderPath); // the folder will have owner as dummyuser
           ContractTestUtils.assertPathExists(fs, "folder was not created", testFolderPath);
@@ -1231,8 +1185,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner(
         "/testDeleteSucceedsForParentDirectoryOwnerUserWithStickybit*",
         WRITE, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1247,7 +1199,6 @@ public class TestNativeAzureFileSystemAuthorization
         @Override
         public Void run() throws Exception {
           authorizer.addAuthRule(parentDir.toString(), WRITE, getCurrentUserShortName(), true);
-          authorizer.addAuthRule(parentDir.toString(), READ, getCurrentUserShortName(), true);
           fs.create(testFilePath);
           ContractTestUtils.assertPathExists(fs, "file was not created", testFilePath);
 
@@ -1285,10 +1236,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner("/testDeleteScenarioForRoot*",
             WRITE, true);
-    authorizer.addAuthRuleForOwner(childPath1.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(childPath2.toString(), READ, true);
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1332,7 +1279,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path("/testMkdirsAccessCheckPositive/1/2/3");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1355,7 +1301,6 @@ public class TestNativeAzureFileSystemAuthorization
     Path testPath = new Path("/testMkdirsWithExistingHierarchyCheckPositive1");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1389,10 +1334,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner(childPath1.toString(),
         WRITE, true);
 
-    authorizer.addAuthRuleForOwner(childPath1.getParent().toString(), READ, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
-    authorizer.addAuthRuleForOwner(childPath3.getParent().toString(), READ, true);
-
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1467,8 +1408,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
     // additional rule used for assertPathExists
-    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1546,7 +1485,6 @@ public class TestNativeAzureFileSystemAuthorization
     final Path testPath = new Path("/testSetOwnerNegative");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     final String owner;
@@ -1585,7 +1523,6 @@ public class TestNativeAzureFileSystemAuthorization
     final Path testPath = new Path("/testSetOwnerPositive");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     final String newOwner = "user2";
@@ -1629,7 +1566,6 @@ public class TestNativeAzureFileSystemAuthorization
     Configuration conf = fs.getConf();
     authorizer.init(conf);
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     final String newOwner = "newowner";
@@ -1673,7 +1609,6 @@ public class TestNativeAzureFileSystemAuthorization
     Configuration conf = fs.getConf();
     authorizer.init(conf);
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     UserGroupInformation user = UserGroupInformation.createUserForTesting(
@@ -1717,8 +1652,6 @@ public class TestNativeAzureFileSystemAuthorization
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     /* Remove nextline after fixing createInternal from FolderRenamePending */
     authorizer.addAuthRuleForOwner(testPath.toString(), WRITE, true);
-    authorizer.addAuthRuleForOwner(srcPath.getParent().toString(), READ, true);
-    authorizer.addAuthRuleForOwner(dstFilePath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -1836,6 +1769,197 @@ public class TestNativeAzureFileSystemAuthorization
 
   }
 
+  /**
+   * Test access when requested permissions match the existing permissions.
+   */
+  @Test
+  public void testAccessWhenPermissionsMatchForAllAndReadWrite() throws Throwable {
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessWhenPermissionsMatchForAllAndReadWrite");
+
+    // For All and Read-Write FsAction.
+    authorizer.init(conf);
+    authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), READ, true);
+    fs.updateWasbAuthorizer(authorizer);
+    try {
+      fs.mkdirs(testPath);
+      assertPathExists(fs, "test path does not exist", testPath);
+      fs.access(testPath, FsAction.ALL);
+      fs.access(testPath, FsAction.READ_WRITE);
+    } finally {
+      recursiveDelete(testPath);
+    }
+
+  }
+
+  /**
+   * Test access when Write and Write-Execute match the existing permissions.
+   * @throws Throwable
+   */
+  @Test
+  public void testAccessWhenPermissionsMatchForWriteAndWriteExecute() throws Throwable {
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessWhenPermissionsMatchForWriteAndWriteExecute");
+    // For Write and Write-Execute FsAction.
+    authorizer.init(conf);
+    authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), WRITE, true);
+    fs.updateWasbAuthorizer(authorizer);
+    try {
+      fs.mkdirs(testPath);
+      assertPathExists(fs, "test path does not exist", testPath);
+      fs.access(testPath, FsAction.WRITE);
+      fs.access(testPath, FsAction.WRITE_EXECUTE);
+    } finally {
+      recursiveDelete(testPath);
+    }
+  }
+
+  /**
+   * Test access when Read and Read-Execute match the existing permissions.
+   * @throws Throwable
+   */
+  @Test
+  public void testAccessWhenPermissionsMatchForReadAndReadExecute() throws Throwable {
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessWhenPermissionsMatchForReadAndReadExecute");
+    // For Read and Read-Execute FsAction.
+    authorizer.init(conf);
+    authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), READ, true);
+    fs.updateWasbAuthorizer(authorizer);
+    try {
+      fs.mkdirs(testPath);
+      assertPathExists(fs, "test path does not exist", testPath);
+      fs.access(testPath, FsAction.READ);
+      fs.access(testPath, FsAction.READ_EXECUTE);
+    } finally {
+      recursiveDelete(testPath);
+    }
+  }
+
+  /**
+   * Test access when Execute and None match the existing permissions.
+   * @throws Throwable
+   */
+  @Test
+  public void testAccessWhenPermissionsMatchForExecuteAndNone() throws Throwable {
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessWhenPermissionsMatchForExecuteAndNone");
+    // For Execute and None FsAction.
+    authorizer.init(conf);
+    authorizer.addAuthRuleForOwner("/", WRITE, true);
+    fs.updateWasbAuthorizer(authorizer);
+    try {
+      fs.mkdirs(testPath);
+      assertPathExists(fs, "test path does not exist", testPath);
+      fs.access(testPath, FsAction.EXECUTE);
+      fs.access(testPath, FsAction.NONE);
+    } finally {
+      recursiveDelete(testPath);
+    }
+  }
+
+  /**
+   * Test access when requested permissions do not match existing permissions.
+   */
+  @Test
+  public void testAccessWhenPermissionsDoNotMatch() throws Throwable{
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessWhenPermissionsDoNotMatch");
+
+    authorizer.init(conf);
+    authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), READ,  true);
+    fs.updateWasbAuthorizer(authorizer);
+    try {
+      fs.mkdirs(testPath);
+      assertPathExists(fs, "test path does not exist", testPath);
+      assertNoAccess(testPath, FsAction.ALL);
+      assertNoAccess(testPath, FsAction.WRITE);
+      assertNoAccess(testPath, FsAction.WRITE_EXECUTE);
+    } finally {
+      recursiveDelete(testPath);
+    }
+  }
+
+  /**
+   * Test access when file does not exist and permissions match.
+   */
+  @Test
+  public void testAccessFileDoesNotExist() throws Throwable{
+    expectedEx.expect(FileNotFoundException.class);
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessFileDoesNotExist");
+
+    authorizer.init(conf);
+    authorizer.addAuthRuleForOwner(testPath.toString(), READ,  true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), WRITE,  true);
+    fs.updateWasbAuthorizer(authorizer);
+    assertPathDoesNotExist(fs, "test path exists", testPath);
+    fs.access(testPath, FsAction.ALL);
+  }
+
+  /**
+   * Test access when file does not exist and permissions do not match.
+   */
+  @Test
+  public void testAccessFileDoesNotExistWhenNoAccessPermission() throws Throwable {
+    expectedEx.expect(FileNotFoundException.class);
+    Configuration conf = fs.getConf();
+    fs.setConf(conf);
+    final Path testPath = new Path("/testAccessFileDoesNotExistWhenNoAccessPermission");
+
+    authorizer.init(conf);
+    fs.updateWasbAuthorizer(authorizer);
+    assertPathDoesNotExist(fs, "test path exists", testPath);
+    fs.access(testPath, FsAction.ALL);
+  }
+
+  /**
+   * Test access for file and intermediate directory after creating
+   * file with intermediate directory.
+   */
+  @Test
+  public void testAccessForFileAndIntermediateDirectoryCreated() throws Throwable {
+    Path parentDir = new Path("/testAccessDirectory");
+    Path intermediateDir = new Path(parentDir, "intermediateDir");
+    Path testPath = new Path(intermediateDir, "test.dat");
+
+    authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
+
+    // Recursive owner Rule is to determine the permission on intermediate dirs.
+    authorizer.addAuthRuleForOwner(parentDir.toString()+"/*", WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString()+"/*", READ, true);
+
+    fs.updateWasbAuthorizer(authorizer);
+
+    try {
+      fs.mkdirs(parentDir);
+      fs.create(testPath);
+      assertPathExists(fs, "testPath was not created", testPath);
+      fs.access(parentDir, FsAction.WRITE);
+      fs.access(parentDir, FsAction.WRITE_EXECUTE);
+
+      fs.access(intermediateDir, FsAction.ALL);
+      fs.access(intermediateDir, FsAction.READ_WRITE);
+
+      fs.access(testPath, FsAction.ALL);
+      fs.access(testPath, FsAction.READ_WRITE);
+    } finally {
+      recursiveDelete(testPath);
+    }
+  }
+
   /**
    * Helper method to test setPermission scenarios. This method handles both positive
    * and negative scenarios of setPermission tests
@@ -1852,7 +1976,6 @@ public class TestNativeAzureFileSystemAuthorization
     final Path testPath = new Path(path);
     final FsPermission newPermission = new FsPermission(FULL_PERMISSION_WITH_STICKYBIT);
     authorizer.addAuthRule("/", WRITE, getCurrentUserShortName(), true);
-    authorizer.addAuthRule("/", READ, getCurrentUserShortName(), true);
     fs.updateWasbAuthorizer(authorizer);
 
     if (chmodAllowedUsers != null && !chmodAllowedUsers.isEmpty()) {
@@ -1900,7 +2023,6 @@ public class TestNativeAzureFileSystemAuthorization
       public Void run() throws Exception {
         try {
           //READ access required for getFileStatus
-          authorizer.addAuthRule("/", READ, getCurrentUserShortName(), true);
           fs.setPermission(testPath, newPermission);
           fail("Failing test because setPermission was expected to throw");
 
@@ -1968,4 +2090,16 @@ public class TestNativeAzureFileSystemAuthorization
     assertEquals("Wrong owner in " + status, owner, status.getOwner());
   }
 
+  private void assertNoAccess(final Path path, final FsAction action)
+      throws Exception {
+    LambdaTestUtils.intercept(AccessControlException.class,
+        new Callable<String>() {
+          @Override
+          public String call() throws Exception {
+            fs.access(path, action);
+            return "Access granted to " + path + " for action " + action;
+          }
+        }
+    );
+  }
 }