HADOOP-14845. Azure wasb: getFileStatus not making any auth check.

Contributed by Sivaguru Sankaridurg

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

@@ -287,7 +287,7 @@ void deleteRenamePendingFile(FileSystem fs, Path redoFile)
      * @param fs file system on which a file is written.
      * @throws IOException Thrown when fail to write file.
      */
-    public void writeFile(FileSystem fs) throws IOException {
+    public void writeFile(NativeAzureFileSystem fs) throws IOException {
       Path path = getRenamePendingFilePath();
       LOG.debug("Preparing to write atomic rename state to {}", path.toString());
       OutputStream output = null;
@@ -296,7 +296,7 @@ public void writeFile(FileSystem fs) throws IOException {
 
       // Write file.
       try {
-        output = fs.create(path);
+        output = fs.createInternal(path, FsPermission.getFileDefault(), false, null);
         output.write(contents.getBytes(Charset.forName("UTF-8")));
       } catch (IOException e) {
         throw new IOException("Unable to write RenamePending file for folder rename from "
@@ -561,7 +561,7 @@ public void redo() throws IOException {
       if (!sourceFolderGone) {
         // Make sure the target folder exists.
         Path dst = fullPath(dstKey);
-        if (!fs.exists(dst)) {
+        if (!fs.existsInternal(dst)) {
           fs.mkdirs(dst);
         }
 
@@ -1655,6 +1655,10 @@ public FSDataOutputStream createNonRecursive(Path f, FsPermission permission,
     // At this point, we have exclusive access to the source folder
     // via the lease, so we will not conflict with an active folder
     // rename operation.
+    //
+    // In the secure case, the call to exists will happen in the context
+    // of the user that initiated the operation. In this case, we should
+    // do the auth-check against ranger for the path.
     if (!exists(parent)) {
       try {
 
@@ -1750,6 +1754,30 @@ private FSDataOutputStream create(Path f, FsPermission permission,
 
     performAuthCheck(ancestor, WasbAuthorizationOperations.WRITE, "create", absolutePath);
 
+    return createInternal(f, permission, overwrite, parentFolderLease);
+  }
+
+
+  /**
+   * This is the version of the create call that is meant for internal usage.
+   * This version is not public facing and does not perform authorization checks.
+   * It is used by the public facing create call and by FolderRenamePending to
+   * create the internal -RenamePending.json file.
+   * @param f the path to a file to be created.
+   * @param permission for the newly created file.
+   * @param overwrite specifies if the file should be overwritten.
+   * @param parentFolderLease lease on the parent folder.
+   * @return the output stream used to write data into the newly created file .
+   * @throws IOException if an IO error occurs while attempting to delete the
+   * path.
+   *
+   */
+  protected FSDataOutputStream createInternal(Path f, FsPermission permission,
+                                    boolean overwrite,
+                                    SelfRenewingLease parentFolderLease)
+      throws FileAlreadyExistsException, IOException {
+
+    Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
 
     FileMetadata existingMetadata = store.retrieveMetadata(key);
@@ -2589,6 +2617,49 @@ public FileStatus getFileStatus(Path f) throws FileNotFoundException, IOExceptio
 
     // 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);
+  }
+
+  /**
+   * Checks if a given path exists in the filesystem.
+   * Calls getFileStatusInternal and has the same costs
+   * as the public facing exists call.
+   * This internal version of the exists call does not perform
+   * authorization checks, and is used internally by various filesystem
+   * operations that need to check if the parent/ancestor/path exist.
+   * The idea is to avoid having to configure authorization policies for
+   * these internal calls.
+   * @param f the path to a file or directory.
+   * @return true if path exists; otherwise false.
+   * @throws IOException if an IO error occurs while attempting to check
+   * for existence of the path.
+   *
+   */
+  protected boolean existsInternal(Path f) throws IOException {
+    try {
+      this.getFileStatusInternal(f);
+      return true;
+    } catch (FileNotFoundException fnfe) {
+      return false;
+    }
+  }
+
+  protected FileStatus getFileStatusInternal(Path f) throws FileNotFoundException, IOException {
+
+    Path absolutePath = makeAbsolute(f);
     String key = pathToKey(absolutePath);
     if (key.length() == 0) { // root always exists
       return newDirectory(null, absolutePath);
@@ -2654,7 +2725,7 @@ private boolean conditionalRedoFolderRename(Path f) throws IOException {
     // Check if there is a -RenamePending.json file for this folder, and if so,
     // redo the rename.
     Path absoluteRenamePendingFile = renamePendingFilePath(f);
-    if (exists(absoluteRenamePendingFile)) {
+    if (existsInternal(absoluteRenamePendingFile)) {
       FolderRenamePending pending =
           new FolderRenamePending(absoluteRenamePendingFile, this);
       pending.redo();

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

@@ -357,7 +357,7 @@ class HttpGetForServiceLocal extends ArgumentMatcher<HttpGet>{
 
     performop(mockHttpClient);
 
-    int expectedNumberOfInvocations = isAuthorizationCachingEnabled ? 1 : 2;
+    int expectedNumberOfInvocations = isAuthorizationCachingEnabled ? 2 : 3;
     Mockito.verify(mockHttpClient, times(expectedNumberOfInvocations)).execute(Mockito.argThat(new HttpGetForServiceLocal()));
     Mockito.verify(mockHttpClient, times(expectedNumberOfInvocations)).execute(Mockito.argThat(new HttpGetForService2()));
   }

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

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.fs.azure;
 
+import java.io.FileNotFoundException;
 import java.security.PrivilegedExceptionAction;
 import java.io.IOException;
 
@@ -28,6 +29,7 @@
 import org.apache.hadoop.fs.contract.ContractTestUtils;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.test.GenericTestUtils;
 import org.apache.hadoop.util.StringUtils;
 
 import org.junit.Assume;
@@ -38,6 +40,7 @@
 import com.google.common.annotations.VisibleForTesting;
 
 import static org.apache.hadoop.fs.azure.AzureNativeFileSystemStore.KEY_USE_SECURE_MODE;
+import static org.apache.hadoop.fs.azure.CachingAuthorizer.KEY_AUTH_SERVICE_CACHING_ENABLE;
 import static org.junit.Assert.assertEquals;
 
 /**
@@ -62,6 +65,7 @@ public Configuration createConfiguration() {
     conf.set(NativeAzureFileSystem.KEY_AZURE_AUTHORIZATION, "true");
     conf.set(RemoteWasbAuthorizerImpl.KEY_REMOTE_AUTH_SERVICE_URLS, "http://localhost/");
     conf.set(NativeAzureFileSystem.AZURE_CHOWN_USERLIST_PROPERTY_NAME, "user1 , user2");
+    conf.set(KEY_AUTH_SERVICE_CACHING_ENABLE, "false");
     return conf;
   }
 
@@ -131,6 +135,7 @@ public void testCreateAccessWithoutCreateIntermediateFoldersCheckPositive() thro
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -156,6 +161,7 @@ public void testCreateAccessWithCreateIntermediateFoldersCheckPositive() throws
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -183,6 +189,7 @@ public void testCreateAccessWithOverwriteCheckNegative() throws Throwable {
     setExpectedFailureMessage("create", testPath);
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -208,6 +215,7 @@ public void testCreateAccessWithOverwriteCheckPositive() throws Throwable {
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(testPath.toString(), WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -313,6 +321,7 @@ public void testRenameAccessCheckPositive() throws Throwable {
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     /* for rename */
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -344,6 +353,7 @@ public void testRenameAccessCheckNegative() throws Throwable {
     /* to create parent dir */
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, false);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -376,6 +386,8 @@ public void testRenameAccessCheckNegativeOnDstFolder() throws Throwable {
     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 {
@@ -405,6 +417,8 @@ public void testRenameAccessCheckPositiveOnDstFolder() throws Throwable {
     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 {
@@ -512,6 +526,7 @@ public void testFileDeleteAccessCheckPositive() throws Throwable {
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
     try {
       fs.create(testPath);
@@ -536,6 +551,7 @@ public void testFileDeleteAccessCheckNegative() throws Throwable {
     setExpectedFailureMessage("delete", testPath);
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
     try {
       fs.create(testPath);
@@ -545,6 +561,7 @@ public void testFileDeleteAccessCheckNegative() throws Throwable {
       /* 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);
@@ -553,6 +570,7 @@ public void testFileDeleteAccessCheckNegative() throws Throwable {
       /* Restore permissions to force a successful delete */
       authorizer.deleteAllAuthRules();
       authorizer.addAuthRuleForOwner("/", WRITE, true);
+      authorizer.addAuthRuleForOwner("/", READ, true);
       fs.updateWasbAuthorizer(authorizer);
 
       fs.delete(testPath, false);
@@ -569,9 +587,12 @@ public void testFileDeleteAccessCheckNegative() throws Throwable {
   public void testFileDeleteAccessWithIntermediateFoldersCheckPositive() throws Throwable {
 
     Path parentDir = new Path("/testDeleteIntermediateFolder");
-    Path testPath = new Path(parentDir, "1/2/test.dat");
+    Path childPath = new Path(parentDir, "1/2");
+    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);
@@ -596,12 +617,16 @@ public void testFileDeleteAccessWithIntermediateFoldersCheckPositive() throws Th
   public void testDeleteAuthCheckFailureLeavesFilesUndeleted() throws Throwable {
 
     Path parentDir = new Path("/testDeleteAuthCheckFailureLeavesFilesUndeleted");
-    Path testPath1 = new Path(parentDir, "child1/test.dat");
+    Path childPath1 = new Path(parentDir, "child1");
-    Path testPath2 = new Path(parentDir, "child2/test.dat");
+    Path childPath2 = new Path(parentDir, "child2");
+    Path testPath1 = new Path(childPath1, "test.dat");
+    Path testPath2 = new Path(childPath2, "test.dat");
 
     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 {
@@ -613,20 +638,19 @@ public void testDeleteAuthCheckFailureLeavesFilesUndeleted() throws Throwable {
       // revoke write on one of the child folders
       authorizer.deleteAllAuthRules();
       authorizer.addAuthRuleForOwner("/", WRITE, true);
-      authorizer.addAuthRuleForOwner("/testDeleteAuthCheckFailureLeavesFilesUndeleted",
+      authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
-        WRITE, true);
+      authorizer.addAuthRuleForOwner(childPath2.toString(), WRITE, true);
-      authorizer.addAuthRuleForOwner("/testDeleteAuthCheckFailureLeavesFilesUndeleted/child2",
+      authorizer.addAuthRuleForOwner(childPath1.toString(), WRITE, false);
-        WRITE, true);
+      authorizer.addAuthRuleForOwner(childPath1.toString(), READ, true);
-      authorizer.addAuthRuleForOwner("/testDeleteAuthCheckFailureLeavesFilesUndeleted/child1",
+      authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
-          WRITE, false);
+      authorizer.addAuthRuleForOwner("/", READ, true);
 
       assertFalse(fs.delete(parentDir, true));
 
       // Assert that only child2 contents are deleted
       ContractTestUtils.assertPathExists(fs, "child1 is deleted!", testPath1);
       ContractTestUtils.assertPathDoesNotExist(fs, "child2 exists after deletion!", testPath2);
-      ContractTestUtils.assertPathDoesNotExist(fs, "child2 exists after deletion!",
+      ContractTestUtils.assertPathDoesNotExist(fs, "child2 exists after deletion!", childPath2);
-          new Path("/testDeleteAuthCheckFailureLeavesFilesUndeleted/childPath2"));
       ContractTestUtils.assertPathExists(fs, "parentDir is deleted!", parentDir);
 
     }
@@ -647,8 +671,8 @@ public void testSingleFileDeleteWithStickyBitPositive() throws Throwable {
     Path testPath = new Path(parentDir, "test.dat");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner("/testSingleFileDeleteWithStickyBitPositive",
+    authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
-        WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -684,8 +708,8 @@ public void testSingleFileDeleteWithStickyBitNegative() throws Throwable {
         parentDir.toString(), testPath.toString()));
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner("/testSingleFileDeleteWithStickyBitNegative",
+    authorizer.addAuthRuleForOwner(parentDir.toString(), WRITE, true);
-        WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -700,16 +724,22 @@ public void testSingleFileDeleteWithStickyBitNegative() throws Throwable {
       dummyUser.doAs(new PrivilegedExceptionAction<Void>() {
         @Override
         public Void run() throws Exception {
-          authorizer.addAuthRule(parentDir.toString(), WRITE,
+          try {
-              getCurrentUserShortName(), true);
+            authorizer.addAuthRule(parentDir.toString(), WRITE,
-          fs.delete(testPath, true);
+                getCurrentUserShortName(), true);
-          return null;
+            authorizer.addAuthRule(parentDir.toString(), READ,
+                getCurrentUserShortName(), true);
+            fs.delete(testPath, true);
+            return null;
+          }
+          catch (WasbAuthorizationException wae) {
+            ContractTestUtils.assertPathExists(fs, "testPath should not be deleted!", testPath);
+            throw wae;
+          }
         }
       });
     }
     finally {
-      ContractTestUtils.assertPathExists(fs, "testPath should not be deleted!", testPath);
-
       allowRecursiveDelete(fs, parentDir.toString());
       fs.delete(parentDir, true);
     }
@@ -724,12 +754,15 @@ public Void run() throws Exception {
   public void testRecursiveDeleteSucceedsWithStickybit() throws Throwable {
 
     Path parentDir = new Path("/testRecursiveDeleteSucceedsWithStickybit");
-    Path testFilePath = new Path(parentDir, "child/test.dat");
+    Path childDir = new Path(parentDir, "child");
-    Path testFolderPath = new Path(parentDir, "child/testDirectory");
+    Path testFilePath = new Path(childDir, "test.dat");
+    Path testFolderPath = new Path(childDir, "testDirectory");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner("/testRecursiveDeleteSucceedsWithStickybit*",
         WRITE, true);
+    authorizer.addAuthRuleForOwner(childDir.toString(), READ, true);
+    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -759,12 +792,15 @@ public void testRecursiveDeleteSucceedsWithStickybit() throws Throwable {
   public void testRecursiveDeleteFailsWithStickybit() throws Throwable {
 
     Path parentDir = new Path("/testRecursiveDeleteFailsWithStickybit");
-    Path testFilePath = new Path(parentDir, "child/test.dat");
+    Path childDir = new Path(parentDir, "child");
-    Path testFolderPath = new Path(parentDir, "child/testDirectory");
+    Path testFilePath = new Path(childDir, "test.dat");
+    Path testFolderPath = new Path(childDir, "testDirectory");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
     authorizer.addAuthRuleForOwner("/testRecursiveDeleteFailsWithStickybit*",
         WRITE, true);
+    authorizer.addAuthRuleForOwner(childDir.toString(), READ, true);
+    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -784,8 +820,7 @@ public void testRecursiveDeleteFailsWithStickybit() throws Throwable {
         @Override
         public Void run() throws Exception {
           // Add auth rules for dummyuser
-          authorizer.addAuthRule("/", WRITE,
+          authorizer.addAuthRule("/", WRITE, getCurrentUserShortName(), true);
-              getCurrentUserShortName(), true);
           authorizer.addAuthRule("/testRecursiveDeleteFailsWithStickybit*",
               WRITE, getCurrentUserShortName(), true);
 
@@ -821,6 +856,7 @@ public void testDeleteSucceedsForOnlyFilesOwnedByUserWithStickybitSet()
     authorizer.addAuthRuleForOwner(
         "/testDeleteSucceedsForOnlyFilesOwnedByUserWithStickybitSet*",
         WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -834,10 +870,12 @@ public void testDeleteSucceedsForOnlyFilesOwnedByUserWithStickybitSet()
       dummyUser.doAs(new PrivilegedExceptionAction<Void>() {
         @Override
         public Void run() throws Exception {
-          authorizer.addAuthRule("/", WRITE,
+          authorizer.addAuthRule("/", WRITE, getCurrentUserShortName(), true);
-              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);
@@ -876,6 +914,8 @@ public void testDeleteSucceedsForParentDirectoryOwnerUserWithStickybit() throws
     authorizer.addAuthRuleForOwner(
         "/testDeleteSucceedsForParentDirectoryOwnerUserWithStickybit*",
         WRITE, true);
+    authorizer.addAuthRuleForOwner(parentDir.toString(), READ, true);
+    authorizer.addAuthRuleForOwner("/", READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -889,8 +929,8 @@ public void testDeleteSucceedsForParentDirectoryOwnerUserWithStickybit() throws
       dummyUser.doAs(new PrivilegedExceptionAction<Void>() {
         @Override
         public Void run() throws Exception {
-          authorizer.addAuthRule("/testDeleteSucceedsForParentDirectoryOwnerUserWithStickybit",
+          authorizer.addAuthRule(parentDir.toString(), WRITE, getCurrentUserShortName(), true);
-              WRITE, getCurrentUserShortName(), true);
+          authorizer.addAuthRule(parentDir.toString(), READ, getCurrentUserShortName(), true);
           fs.create(testFilePath);
           ContractTestUtils.assertPathExists(fs, "file was not created", testFilePath);
 
@@ -920,12 +960,18 @@ public Void run() throws Exception {
   public void testDeleteScenarioForRoot() throws Throwable {
     Path rootPath = new Path("/");
     Path parentDir = new Path("/testDeleteScenarioForRoot");
-    Path testPath1 = new Path(parentDir, "child1/test.dat");
+    Path childPath1 = new Path(parentDir, "child1");
-    Path testPath2 = new Path(parentDir, "child2/testFolder");
+    Path childPath2 = new Path(parentDir, "child2");
+    Path testPath1 = new Path(childPath1, "test.dat");
+    Path testPath2 = new Path(childPath2, "testFolder");
 
     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 {
@@ -948,13 +994,14 @@ public void testDeleteScenarioForRoot() throws Throwable {
   }
 
   /**
-   * Positive test for getFileStatus. No permissions are required for getting filestatus.
+   * Positive test for getFileStatus.
    * @throws Throwable
    */
   @Test
   public void testGetFileStatusPositive() throws Throwable {
 
     Path testPath = new Path("/");
+    authorizer.addAuthRuleForOwner("/", READ, true);
     ContractTestUtils.assertIsDirectory(fs, testPath);
   }
 
@@ -968,6 +1015,7 @@ public void testMkdirsCheckPositive() throws Throwable {
     Path testPath = new Path("/testMkdirsAccessCheckPositive/1/2/3");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -990,6 +1038,7 @@ public void testMkdirsWithExistingHierarchyCheckPositive1() throws Throwable {
     Path testPath = new Path("/testMkdirsWithExistingHierarchyCheckPositive1");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     try {
@@ -998,6 +1047,7 @@ public void testMkdirsWithExistingHierarchyCheckPositive1() throws Throwable {
 
       /* Don't need permissions to create a directory that already exists */
       authorizer.deleteAllAuthRules();
+      authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true); // for assert
 
       fs.mkdirs(testPath);
       ContractTestUtils.assertIsDirectory(fs, testPath);
@@ -1022,6 +1072,10 @@ public void testMkdirsWithExistingHierarchyCheckPositive2() throws Throwable {
     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 {
@@ -1094,10 +1148,10 @@ public void testOwnerPermissionPositive() throws Throwable {
     Path testPath = new Path(parentDir, "test.data");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
-    authorizer.addAuthRuleForOwner(testPath.toString(), READ, 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 {
@@ -1178,6 +1232,7 @@ public void testSetOwnerThrowsForUnauthorisedUsers() throws Throwable {
     Path testPath = new Path("/testSetOwnerNegative");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     String owner = null;
@@ -1212,9 +1267,10 @@ public void testSetOwnerSucceedsForAuthorisedUsers() throws Throwable {
     Path testPath = new Path("/testSetOwnerPositive");
 
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
-    String newOwner = "newowner";
+    String newOwner = "user2";
     String newGroup = "newgroup";
 
     UserGroupInformation authorisedUser = UserGroupInformation.createUserForTesting(
@@ -1257,6 +1313,7 @@ public void testSetOwnerSucceedsForAnyUserWhenWildCardIsSpecified() throws Throw
 
     authorizer.init(conf);
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     String newOwner = "newowner";
@@ -1303,6 +1360,7 @@ public void testSetOwnerFailsForIllegalSetup() throws Throwable {
 
     authorizer.init(conf);
     authorizer.addAuthRuleForOwner("/", WRITE, true);
+    authorizer.addAuthRuleForOwner(testPath.getParent().toString(), READ, true);
     fs.updateWasbAuthorizer(authorizer);
 
     String owner = null;
@@ -1328,4 +1386,51 @@ public Void run() throws Exception {
       fs.delete(testPath, false);
     }
   }
+
+  /** Test to ensure that the internal RenamePending mechanism
+   * does not make authorization calls.
+   */
+  @Test
+  public void testRenamePendingAuthorizationCalls() throws Throwable {
+    Path testPath = new Path("/testRenamePendingAuthorizationCalls");
+    Path srcPath = new Path(testPath, "srcPath");
+    Path dstPath = new Path(testPath, "dstPath");
+    Path srcFilePath = new Path(srcPath, "file.txt");
+    Path dstFilePath = new Path(dstPath, "file.txt");
+
+    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 {
+      fs.create(srcFilePath);
+
+      String srcKey = fs.pathToKey(srcPath);
+      String dstKey = fs.pathToKey(dstPath);
+
+      // Create a -RenamePendingFile
+      NativeAzureFileSystem.FolderRenamePending renamePending =
+          new NativeAzureFileSystem.FolderRenamePending(srcKey, dstKey, null, fs);
+      renamePending.writeFile(fs);
+
+      // Initiate the pending-rename
+      fs.getFileStatus(srcPath);
+    } catch (FileNotFoundException fnfe) {
+      // This is expected because getFileStatus would complete the pending "rename"
+      // represented by the -RenamePending file.
+      GenericTestUtils.assertExceptionContains(
+          srcPath.toString() + ": No such file or directory.", fnfe
+      );
+
+      // The pending rename should have completed
+      ContractTestUtils.assertPathExists(fs,
+          "dstFilePath does not exist -- pending rename failed", dstFilePath);
+    } finally {
+      allowRecursiveDelete(fs, testPath.toString());
+      fs.delete(testPath, true);
+    }
+  }
 }

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

@@ -343,9 +343,9 @@ public void testMetricsOnFileExistsDelete() throws Exception {
     assertFalse(getFileSystem().exists(filePath));
     // At the time of writing this code it takes 2 requests/responses to
     // check existence, which seems excessive, plus initial request for
-    // container check.
+    // container check, plus 2 ancestor checks only in the secure case.
     logOpResponseCount("Checking file existence for non-existent file", base);
-    base = assertWebResponsesInRange(base, 1, 3);
+    base = assertWebResponsesInRange(base, 1, 5);
 
     // Create an empty file
     assertTrue(getFileSystem().createNewFile(filePath));
@@ -354,7 +354,7 @@ public void testMetricsOnFileExistsDelete() throws Exception {
     // Check existence again
     assertTrue(getFileSystem().exists(filePath));
     logOpResponseCount("Checking file existence for existent file", base);
-    base = assertWebResponsesInRange(base, 1, 2);
+    base = assertWebResponsesInRange(base, 1, 4);
 
     // Delete the file
     assertEquals(0, AzureMetricsTestUtil.getLongCounterValue(getInstrumentation(), WASB_FILES_DELETED));