HADOOP-17397: ABFS: SAS Test updates for version and permission update

DETAILS:

    The previous commit for HADOOP-17397 was not the correct fix.  DelegationSASGenerator.getDelegationSAS
    should return sp=p for the set-permission and set-acl operations.  The tests have also been updated as
    follows:

    1. When saoid and suoid are not specified, skoid must have an RBAC role assignment which grants
       Microsoft.Storage/storageAccounts/blobServices/containers/blobs/modifyPermissions/action and sp=p
       to set permissions or set ACL.

    2. When saoid or suiod is specified, same as 1) but furthermore the saoid or suoid must be an owner of
       the file or directory in order for the operation to succeed.

    3. When saoid or suiod is specified, the ownership check is bypassed by also including 'o' (ownership)
       in the SAS permission (for example, sp=op).  Note that 'o' grants the saoid or suoid the ability to
       change the file or directory owner to themself, and they can also change the owning group. Generally
       speaking, if a trusted authorizer would like to give a user the ability to change the permissions or
       ACL, then that user should be the file or directory owner.

TEST RESULTS:

    namespace.enabled=true
    auth.type=SharedKey
    -------------------
    $mvn -T 1C -Dparallel-tests=abfs -Dscale -DtestsThreadCount=8 clean verify
    Tests run: 89, Failures: 0, Errors: 0, Skipped: 0
    Tests run: 461, Failures: 0, Errors: 0, Skipped: 24
    Tests run: 208, Failures: 0, Errors: 0, Skipped: 24

    namespace.enabled=true
    auth.type=OAuth
    -------------------
    $mvn -T 1C -Dparallel-tests=abfs -Dscale -DtestsThreadCount=8 clean verify
    Tests run: 89, Failures: 0, Errors: 0, Skipped: 0
    Tests run: 461, Failures: 0, Errors: 0, Skipped: 70
    Tests run: 208, Failures: 0, Errors: 0, Skipped: 141

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

@@ -25,6 +25,8 @@ import java.util.Arrays;
 import java.util.List;
 import java.util.UUID;
 
+import org.apache.hadoop.fs.azurebfs.contracts.exceptions.AbfsRestOperationException;
+import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
 import org.assertj.core.api.Assertions;
 import org.junit.Assume;
 import org.junit.Test;
@@ -94,13 +96,16 @@ public class ITestAzureBlobFileSystemDelegationSAS extends AbstractAbfsIntegrati
     final AzureBlobFileSystem fs = getFileSystem();
 
     Path rootPath = new Path("/");
+    fs.setOwner(rootPath, MockDelegationSASTokenProvider.TEST_OWNER, null);
     fs.setPermission(rootPath, new FsPermission(FsAction.ALL, FsAction.READ_EXECUTE, FsAction.EXECUTE));
     FileStatus rootStatus = fs.getFileStatus(rootPath);
     assertEquals("The directory permissions are not expected.", "rwxr-x--x", rootStatus.getPermission().toString());
+    assertEquals("The directory owner is not expected.",
+        MockDelegationSASTokenProvider.TEST_OWNER,
+        rootStatus.getOwner());
 
     Path dirPath = new Path(UUID.randomUUID().toString());
     fs.mkdirs(dirPath);
-    fs.setOwner(dirPath, MockDelegationSASTokenProvider.TEST_OWNER, null);
 
     Path filePath = new Path(dirPath, "file1");
     fs.create(filePath).close();
@@ -324,8 +329,10 @@ public class ITestAzureBlobFileSystemDelegationSAS extends AbstractAbfsIntegrati
     final AzureBlobFileSystem fs = getFileSystem();
     Path rootPath = new Path(AbfsHttpConstants.ROOT_PATH);
 
+    fs.setOwner(rootPath, MockDelegationSASTokenProvider.TEST_OWNER, null);
     FileStatus status = fs.getFileStatus(rootPath);
     assertEquals("rwxr-x---", status.getPermission().toString());
+    assertEquals(MockDelegationSASTokenProvider.TEST_OWNER, status.getOwner());
     assertTrue(status.isDirectory());
 
     AclStatus acl = fs.getAclStatus(rootPath);
@@ -410,4 +417,64 @@ public class ITestAzureBlobFileSystemDelegationSAS extends AbstractAbfsIntegrati
             .renamePath("testABC/test.xt", "testABC/abc.txt", null));
   }
 
+  @Test
+  // SetPermission should fail when saoid is not the owner and succeed when it is.
+  public void testSetPermissionForNonOwner() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+
+    Path rootPath = new Path("/");
+    FileStatus rootStatus = fs.getFileStatus(rootPath);
+    assertEquals("The permissions are not expected.",
+        "rwxr-x---",
+        rootStatus.getPermission().toString());
+    assertNotEquals("The owner is not expected.",
+        MockDelegationSASTokenProvider.TEST_OWNER,
+        rootStatus.getOwner());
+
+    // Attempt to set permission without being the owner.
+    try {
+      fs.setPermission(rootPath, new FsPermission(FsAction.ALL,
+          FsAction.READ_EXECUTE, FsAction.EXECUTE));
+      assertTrue("Set permission should fail because saoid is not the owner.", false);
+    } catch (AbfsRestOperationException ex) {
+      // Should fail with permission mismatch
+      assertEquals(AzureServiceErrorCode.AUTHORIZATION_PERMISSION_MISS_MATCH,
+          ex.getErrorCode());
+    }
+
+    // Attempt to set permission as the owner.
+    fs.setOwner(rootPath, MockDelegationSASTokenProvider.TEST_OWNER, null);
+    fs.setPermission(rootPath, new FsPermission(FsAction.ALL,
+        FsAction.READ_EXECUTE, FsAction.EXECUTE));
+    rootStatus = fs.getFileStatus(rootPath);
+    assertEquals("The permissions are not expected.",
+        "rwxr-x--x",
+        rootStatus.getPermission().toString());
+    assertEquals("The directory owner is not expected.",
+        MockDelegationSASTokenProvider.TEST_OWNER,
+        rootStatus.getOwner());
+  }
+
+  @Test
+  // Without saoid or suoid, setPermission should succeed with sp=p for a non-owner.
+  public void testSetPermissionWithoutAgentForNonOwner() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    Path path = new Path(MockDelegationSASTokenProvider.NO_AGENT_PATH);
+    fs.create(path).close();
+
+    FileStatus status = fs.getFileStatus(path);
+    assertEquals("The permissions are not expected.",
+        "rw-r--r--",
+        status.getPermission().toString());
+    assertNotEquals("The owner is not expected.",
+        TestConfigurationKeys.FS_AZURE_TEST_APP_SERVICE_PRINCIPAL_OBJECT_ID,
+        status.getOwner());
+
+    fs.setPermission(path, new FsPermission(FsAction.READ, FsAction.READ, FsAction.NONE));
+
+    FileStatus fileStatus = fs.getFileStatus(path);
+    assertEquals("The permissions are not expected.",
+        "r--r-----",
+        fileStatus.getPermission().toString());
+  }
 }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/extensions/MockDelegationSASTokenProvider.java

@@ -48,6 +48,7 @@ public class MockDelegationSASTokenProvider implements SASTokenProvider {
 
   public static final String TEST_OWNER = "325f1619-4205-432f-9fce-3fd594325ce5";
   public static final String CORRELATION_ID = "66ff4ffc-ff17-417e-a2a9-45db8c5b0b5c";
+  public static final String NO_AGENT_PATH = "NoAgentPath";
 
   @Override
   public void initialize(Configuration configuration, String accountName) throws IOException {
@@ -131,11 +132,16 @@ public class MockDelegationSASTokenProvider implements SASTokenProvider {
   @Override
   public String getSASToken(String accountName, String fileSystem, String path,
                      String operation) throws IOException, AccessControlException {
-    // The user for these tests is always TEST_OWNER.  The check access operation
+    // Except for the special case where we test without an agent,
+    // the user for these tests is always TEST_OWNER.  The check access operation
     // requires suoid to check permissions for the user and will throw if the
     // user does not have access and otherwise succeed.
-    String saoid = (operation == SASTokenProvider.CHECK_ACCESS_OPERATION) ? null : TEST_OWNER;
+    String saoid = null;
-    String suoid = (operation == SASTokenProvider.CHECK_ACCESS_OPERATION) ? TEST_OWNER : null;
+    String suoid = null;
+    if (path == null || !path.endsWith(NO_AGENT_PATH)) {
+      saoid = (operation == SASTokenProvider.CHECK_ACCESS_OPERATION) ? null : TEST_OWNER;
+      suoid = (operation == SASTokenProvider.CHECK_ACCESS_OPERATION) ? TEST_OWNER : null;
+    }
     return generator.getDelegationSAS(accountName, fileSystem, path, operation,
         saoid, suoid, CORRELATION_ID);
   }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/utils/DelegationSASGenerator.java

@@ -88,7 +88,7 @@ public class DelegationSASGenerator extends SASGenerator {
         break;
       case SASTokenProvider.SET_ACL_OPERATION:
       case SASTokenProvider.SET_PERMISSION_OPERATION:
-        sp = "op";
+        sp = "p";
         break;
       case SASTokenProvider.SET_OWNER_OPERATION:
         sp = "o";