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

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

@@ -25,6 +25,8 @@
 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 void testCheckAccess() throws Exception {
     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 void testRootPath() throws Exception {
     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 void testSignatureMaskOnExceptionMessage() throws Exception {
             .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 @@ private byte[] getUserDelegationKey(String accountName, String appID, String app
   @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 suoid = (operation == SASTokenProvider.CHECK_ACCESS_OPERATION) ? TEST_OWNER : null;
+    String saoid = 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 String getDelegationSAS(String accountName, String containerName, String
         break;
       case SASTokenProvider.SET_ACL_OPERATION:
       case SASTokenProvider.SET_PERMISSION_OPERATION:
-        sp = "op";
+        sp = "p";
         break;
       case SASTokenProvider.SET_OWNER_OPERATION:
         sp = "o";