HADOOP-16315. ABFS: transform full UPN for named user in AclStatus

Contributed by Da Zhou

Change-Id: Ibc78322415fcbeff89c06c8586c53f5695550290

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

@@ -721,8 +721,8 @@ public class AzureBlobFileSystemStore {
             path.toString(),
             AclEntry.aclSpecToString(aclSpec));
 
-    final List<AclEntry> transformedAclEntries = identityTransformer.transformAclEntriesForSetRequest(aclSpec);
-    final Map<String, String> modifyAclEntries = AbfsAclHelper.deserializeAclSpec(AclEntry.aclSpecToString(transformedAclEntries));
+    identityTransformer.transformAclEntriesForSetRequest(aclSpec);
+    final Map<String, String> modifyAclEntries = AbfsAclHelper.deserializeAclSpec(AclEntry.aclSpecToString(aclSpec));
     boolean useUpn = AbfsAclHelper.isUpnFormatAclEntries(modifyAclEntries);
 
     final AbfsRestOperation op = client.getAclStatus(AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path, true), useUpn);
@@ -748,8 +748,8 @@ public class AzureBlobFileSystemStore {
             path.toString(),
             AclEntry.aclSpecToString(aclSpec));
 
-    final List<AclEntry> transformedAclEntries = identityTransformer.transformAclEntriesForSetRequest(aclSpec);
-    final Map<String, String> removeAclEntries = AbfsAclHelper.deserializeAclSpec(AclEntry.aclSpecToString(transformedAclEntries));
+    identityTransformer.transformAclEntriesForSetRequest(aclSpec);
+    final Map<String, String> removeAclEntries = AbfsAclHelper.deserializeAclSpec(AclEntry.aclSpecToString(aclSpec));
     boolean isUpnFormat = AbfsAclHelper.isUpnFormatAclEntries(removeAclEntries);
 
     final AbfsRestOperation op = client.getAclStatus(AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path, true), isUpnFormat);
@@ -827,8 +827,8 @@ public class AzureBlobFileSystemStore {
             path.toString(),
             AclEntry.aclSpecToString(aclSpec));
 
-    final List<AclEntry> transformedAclEntries = identityTransformer.transformAclEntriesForSetRequest(aclSpec);
-    final Map<String, String> aclEntries = AbfsAclHelper.deserializeAclSpec(AclEntry.aclSpecToString(transformedAclEntries));
+    identityTransformer.transformAclEntriesForSetRequest(aclSpec);
+    final Map<String, String> aclEntries = AbfsAclHelper.deserializeAclSpec(AclEntry.aclSpecToString(aclSpec));
     final boolean isUpnFormat = AbfsAclHelper.isUpnFormatAclEntries(aclEntries);
 
     final AbfsRestOperation op = client.getAclStatus(AbfsHttpConstants.FORWARD_SLASH + getRelativePath(path, true), isUpnFormat);
@@ -867,7 +867,8 @@ public class AzureBlobFileSystemStore {
     final String permissions = result.getResponseHeader(HttpHeaderConfigurations.X_MS_PERMISSIONS);
     final String aclSpecString = op.getResult().getResponseHeader(HttpHeaderConfigurations.X_MS_ACL);
 
-    final List<AclEntry> processedAclEntries = AclEntry.parseAclSpec(AbfsAclHelper.processAclString(aclSpecString), true);
+    final List<AclEntry> aclEntries = AclEntry.parseAclSpec(AbfsAclHelper.processAclString(aclSpecString), true);
+    identityTransformer.transformAclEntriesForGetRequest(aclEntries, userName, primaryUserGroup);
     final FsPermission fsPermission = permissions == null ? new AbfsPermission(FsAction.ALL, FsAction.ALL, FsAction.ALL)
             : AbfsPermission.valueOf(permissions);
 
@@ -877,7 +878,7 @@ public class AzureBlobFileSystemStore {
 
     aclStatusBuilder.setPermission(fsPermission);
     aclStatusBuilder.stickyBit(fsPermission.getStickyBit());
-    aclStatusBuilder.addEntries(processedAclEntries);
+    aclStatusBuilder.addEntries(aclEntries);
     return aclStatusBuilder.build();
   }
 

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/oauth2/IdentityTransformer.java

@@ -81,6 +81,7 @@ public class IdentityTransformer {
    * Perform identity transformation for the Get request results in AzureBlobFileSystemStore:
    * getFileStatus(), listStatus(), getAclStatus().
    * Input originalIdentity can be one of the following:
+   * <pre>
    * 1. $superuser:
    *     by default it will be transformed to local user/group, this can be disabled by setting
    *     "fs.azure.identity.transformer.skip.superuser.replacement" to true.
@@ -93,7 +94,7 @@ public class IdentityTransformer {
    * 3. User principal name (UPN):
    *     can be transformed to a short name(localIdentity) if originalIdentity is owner name, and
    *     "fs.azure.identity.transformer.enable.short.name" is enabled.
-   *
+   * </pre>
    * @param originalIdentity the original user or group in the get request results: FileStatus, AclStatus.
    * @param isUserName indicate whether the input originalIdentity is an owner name or owning group name.
    * @param localIdentity the local user or group, should be parsed from UserGroupInformation.
@@ -134,7 +135,7 @@ public class IdentityTransformer {
    * Perform Identity transformation when setting owner on a path.
    * There are four possible input:
    * 1.short name; 2.$superuser; 3.Fully qualified name; 4. principal id.
-   *
+   * <pre>
    * short name could be transformed to:
    *    - A service principal id or $superuser, if short name belongs a daemon service
    *      stated in substitution list AND "fs.azure.identity.transformer.service.principal.id"
@@ -142,7 +143,7 @@ public class IdentityTransformer {
    *    - Fully qualified name, if "fs.azure.identity.transformer.domain.name" is set in configuration.
    *
    * $superuser, fully qualified name and principalId should not be transformed.
-   *
+   * </pre>
    * @param userOrGroup the user or group to be set as owner.
    * @return user or group after transformation.
    * */
@@ -168,21 +169,21 @@ public class IdentityTransformer {
    * Perform Identity transformation when calling setAcl(),removeAclEntries() and modifyAclEntries()
    * If the AclEntry type is a user or group, and its name is one of the following:
    * 1.short name; 2.$superuser; 3.Fully qualified name; 4. principal id.
+   * <pre>
    * Short name could be transformed to:
    *    - A service principal id or $superuser, if short name belongs a daemon service
    *      stated in substitution list AND "fs.azure.identity.transformer.service.principal.id"
    *      is set with $superuser or a principal id.
    *    - A fully qualified name, if the AclEntry type is User AND if "fs.azure.identity.transformer.domain.name"
-   *    is set in configuration. This is to make the behavior consistent with HDI.
+   *      is set in configuration. This is to make the behavior consistent with HDI.
    *
    * $superuser, fully qualified name and principal id should not be transformed.
-   *
+   * </pre>
    * @param aclEntries list of AclEntry
-   * @return list of AclEntry after the identity transformation.
    * */
-  public List<AclEntry> transformAclEntriesForSetRequest(final List<AclEntry> aclEntries) {
+  public void transformAclEntriesForSetRequest(final List<AclEntry> aclEntries) {
     if (skipUserIdentityReplacement) {
-      return aclEntries;
+      return;
     }
 
     for (int i = 0; i < aclEntries.size(); i++) {
@@ -218,7 +219,63 @@ public class IdentityTransformer {
       // Replace the original AclEntry
       aclEntries.set(i, aclEntryBuilder.build());
     }
-    return aclEntries;
+  }
+
+  /**
+   * Perform Identity transformation when calling GetAclStatus()
+   * If the AclEntry type is a user or group, and its name is one of the following:
+   * <pre>
+   * 1. $superuser:
+   *     by default it will be transformed to local user/group, this can be disabled by setting
+   *     "fs.azure.identity.transformer.skip.superuser.replacement" to true.
+   *
+   * 2. User principal id:
+   *     can be transformed to localUser/localGroup, if this principal id matches the principal id set in
+   *     "fs.azure.identity.transformer.service.principal.id" and localIdentity is stated in
+   *     "fs.azure.identity.transformer.service.principal.substitution.list"
+   *
+   * 3. User principal name (UPN):
+   *     can be transformed to a short name(local identity) if originalIdentity is owner name, and
+   *     "fs.azure.identity.transformer.enable.short.name" is enabled.
+   * </pre>
+   * @param aclEntries list of AclEntry
+   * @param localUser local user name
+   * @param localGroup local primary group
+   * */
+  public void transformAclEntriesForGetRequest(final List<AclEntry> aclEntries, String localUser, String localGroup) {
+    if (skipUserIdentityReplacement) {
+      return;
+    }
+
+    for (int i = 0; i < aclEntries.size(); i++) {
+      AclEntry aclEntry = aclEntries.get(i);
+      String name = aclEntry.getName();
+      String transformedName = name;
+      if (name == null || name.isEmpty() || aclEntry.getType().equals(AclEntryType.OTHER) || aclEntry.getType().equals(AclEntryType.MASK)) {
+        continue;
+      }
+
+      // when type of aclEntry is user or group
+      if (aclEntry.getType().equals(AclEntryType.USER)) {
+        transformedName = transformIdentityForGetRequest(name, true, localUser);
+      } else if (aclEntry.getType().equals(AclEntryType.GROUP)) {
+        transformedName = transformIdentityForGetRequest(name, false, localGroup);
+      }
+
+      // Avoid unnecessary new AclEntry allocation
+      if (transformedName.equals(name)) {
+        continue;
+      }
+
+      AclEntry.Builder aclEntryBuilder = new AclEntry.Builder();
+      aclEntryBuilder.setType(aclEntry.getType());
+      aclEntryBuilder.setName(transformedName);
+      aclEntryBuilder.setScope(aclEntry.getScope());
+      aclEntryBuilder.setPermission(aclEntry.getPermission());
+
+      // Replace the original AclEntry
+      aclEntries.set(i, aclEntryBuilder.build());
+    }
   }
 
   /**

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

@@ -249,9 +249,13 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
             aclEntry(DEFAULT, MASK, ALL)               //         to make the behavior consistent with HDI.
     );
 
+    // make a copy
+    List<AclEntry> aclEntries = Lists.newArrayList(aclEntriesToBeTransformed);
+
     // Default config should not change the identities
     IdentityTransformer identityTransformer = getTransformerWithDefaultIdentityConfig(config);
-    checkAclEntriesList(aclEntriesToBeTransformed, identityTransformer.transformAclEntriesForSetRequest(aclEntriesToBeTransformed));
+    identityTransformer.transformAclEntriesForSetRequest(aclEntries);
+    checkAclEntriesList(aclEntriesToBeTransformed, aclEntries);
 
     resetIdentityConfig(config);
     // With config
@@ -261,6 +265,8 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
     config.set(FS_AZURE_OVERRIDE_OWNER_SP, SERVICE_PRINCIPAL_ID);
     identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
 
+    identityTransformer.transformAclEntriesForSetRequest(aclEntries);
+
     // expected acl entries
     List<AclEntry> expectedAclEntries = Lists.newArrayList(
             aclEntry(ACCESS, USER, SERVICE_PRINCIPAL_ID, ALL),
@@ -274,8 +280,56 @@ public class ITestAbfsIdentityTransformer extends AbstractAbfsScaleTest{
             aclEntry(DEFAULT, MASK, ALL)
     );
 
-    checkAclEntriesList(identityTransformer.transformAclEntriesForSetRequest(aclEntriesToBeTransformed), expectedAclEntries);
+    checkAclEntriesList(aclEntries, expectedAclEntries);
+  }
+
+  @Test
+  public void transformAclEntriesForGetRequest() throws IOException {
+    Configuration config = this.getRawConfiguration();
+    resetIdentityConfig(config);
+
+    List<AclEntry> aclEntriesToBeTransformed = Lists.newArrayList(
+            aclEntry(ACCESS, USER, FULLY_QUALIFIED_NAME, ALL),
+            aclEntry(DEFAULT, USER, SUPER_USER, ALL),
+            aclEntry(DEFAULT, USER, SERVICE_PRINCIPAL_ID, ALL),
+            aclEntry(DEFAULT, USER, SHORT_NAME, ALL),
+            aclEntry(DEFAULT, GROUP, SHORT_NAME, ALL),
+            aclEntry(DEFAULT, OTHER, ALL),
+            aclEntry(DEFAULT, MASK, ALL)
+    );
+
+    // make a copy
+    List<AclEntry> aclEntries = Lists.newArrayList(aclEntriesToBeTransformed);
+
+    // Default config should not change the identities
+    IdentityTransformer identityTransformer = getTransformerWithDefaultIdentityConfig(config);
+    identityTransformer.transformAclEntriesForGetRequest(aclEntries, localUser, localGroup);
+    checkAclEntriesList(aclEntriesToBeTransformed, aclEntries);
+
+    resetIdentityConfig(config);
+    // With config
+    config.set(FS_AZURE_OVERRIDE_OWNER_SP_LIST, localUser + ",a,b,c,d");
+    config.setBoolean(FS_AZURE_FILE_OWNER_ENABLE_SHORTNAME, true);
+    config.set(FS_AZURE_FILE_OWNER_DOMAINNAME, DOMAIN);
+    config.set(FS_AZURE_OVERRIDE_OWNER_SP, SERVICE_PRINCIPAL_ID);
+    identityTransformer = getTransformerWithCustomizedIdentityConfig(config);
+
+    // make a copy
+    aclEntries = Lists.newArrayList(aclEntriesToBeTransformed);
+    identityTransformer.transformAclEntriesForGetRequest(aclEntries, localUser, localGroup);
+
+    // expected acl entries
+    List<AclEntry> expectedAclEntries = Lists.newArrayList(
+            aclEntry(ACCESS, USER, SHORT_NAME, ALL), // Full UPN should be transformed to shortName
+            aclEntry(DEFAULT, USER, localUser, ALL), // $SuperUser should be transformed to shortName
+            aclEntry(DEFAULT, USER, localUser, ALL), // principal Id should be transformed to local user name
+            aclEntry(DEFAULT, USER, SHORT_NAME, ALL),
+            aclEntry(DEFAULT, GROUP, SHORT_NAME, ALL),
+            aclEntry(DEFAULT, OTHER, ALL),
+            aclEntry(DEFAULT, MASK, ALL)
+    );
 
+    checkAclEntriesList(aclEntries, expectedAclEntries);
   }
 
   private void resetIdentityConfig(Configuration config) {