TEZ-4739: Improve ACLManager

tez-api/src/main/java/org/apache/tez/common/security/ACLManager.java

@@ -22,6 +22,7 @@
 import java.util.Collection;
 import java.util.EnumSet;
 import java.util.HashMap;
+import java.util.LinkedHashSet;
 import java.util.Map;
 import java.util.Map.Entry;
 import java.util.Set;
@@ -57,8 +58,8 @@ public ACLManager(String amUser) {
   public ACLManager(String amUser, Configuration conf) {
     this.amUser = amUser;
     this.dagUser = null;
-    this.users = new HashMap<ACLType, Set<String>>();
-    this.groups = new HashMap<ACLType, Set<String>>();
+    this.users = new HashMap<>();
+    this.groups = new HashMap<>();
     aclsEnabled = conf.getBoolean(TezConfiguration.TEZ_AM_ACLS_ENABLED,
         TezConfiguration.TEZ_AM_ACLS_ENABLED_DEFAULT);
     if (!aclsEnabled) {
@@ -76,10 +77,14 @@ public ACLManager(String amUser, Configuration conf) {
   public ACLManager(ACLManager amACLManager, String dagUser, ACLInfo aclInfo) {
     this.amUser = amACLManager.amUser;
     this.dagUser = dagUser;
-    this.users = amACLManager.users;
-    this.groups = amACLManager.groups;
+    // Deep-copy the AM-level maps so per-DAG entries stay scoped to this DAG.
+    // A shallow copy of the outer map would leave the inner Sets shared with
+    // the AM manager, so a future set.add(...) on any entry would silently
+    // mutate the AM's global ACLs and every sibling DAG's view of them.
+    this.users = deepCopy(amACLManager.users);
+    this.groups = deepCopy(amACLManager.groups);
     this.aclsEnabled = amACLManager.aclsEnabled;
-    if (!aclsEnabled) {
+    if (!aclsEnabled || aclInfo == null) {
       return;
     }
     if (aclInfo.getUsersWithViewAccessCount() > 0) {
@@ -233,6 +238,14 @@ public Map<ApplicationAccessType, String> toYARNACls() {
     return acls;
   }
 
+  private static Map<ACLType, Set<String>> deepCopy(Map<ACLType, Set<String>> source) {
+    Map<ACLType, Set<String>> copy = new HashMap<>(source.size());
+    for (Entry<ACLType, Set<String>> entry : source.entrySet()) {
+      copy.put(entry.getKey(), new LinkedHashSet<>(entry.getValue()));
+    }
+    return copy;
+  }
+
   public static String toCommaSeparatedString(Collection<String> collection) {
     StringBuilder sb = new StringBuilder();
     boolean first = true;

tez-api/src/test/java/org/apache/tez/common/security/TestACLManager.java

@@ -23,7 +23,9 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.util.Map;
+import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
 import org.apache.hadoop.conf.Configuration;
@@ -431,4 +433,107 @@ public void testConvertToYARNACLs() {
 
   }
 
+  /**
+   * Per-DAG ACLs should stay scoped to their DAG and not carry over to
+   * sibling DAGs sharing the same AM ACLManager.
+   */
+  @Test
+  @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+  public void testDAGACLsDoNotLeakAcrossDAGs() throws IOException {
+    UserGroupInformation amUser = UserGroupInformation.createUserForTesting("amUser", noGroups);
+    UserGroupInformation dag1User = UserGroupInformation.createUserForTesting("dag1User", noGroups);
+    UserGroupInformation dag2User = UserGroupInformation.createUserForTesting("dag2User", noGroups);
+    UserGroupInformation user1 = UserGroupInformation.createUserForTesting("user1", noGroups);
+    String[] grp1 = new String[] {"grp1"};
+    UserGroupInformation user2 = UserGroupInformation.createUserForTesting("user2", grp1);
+
+    Configuration conf = new Configuration(false);
+    ACLManager amAclManager = new ACLManager(amUser.getShortUserName(), conf);
+
+    ACLInfo.Builder dag1Builder = ACLInfo.newBuilder();
+    dag1Builder.addUsersWithViewAccess(user1.getShortUserName());
+    dag1Builder.addUsersWithModifyAccess(user1.getShortUserName());
+    dag1Builder.addGroupsWithViewAccess("grp1");
+    dag1Builder.addGroupsWithModifyAccess("grp1");
+    ACLManager dag1AclManager =
+        new ACLManager(amAclManager, dag1User.getShortUserName(), dag1Builder.build());
+
+    assertTrue(dag1AclManager.checkDAGModifyAccess(user1));
+    assertTrue(dag1AclManager.checkDAGViewAccess(user1));
+    assertTrue(dag1AclManager.checkDAGModifyAccess(user2));
+    assertTrue(dag1AclManager.checkDAGViewAccess(user2));
+
+    // DAG 2 in the same session has an empty ACLInfo — DAG 1's grants must
+    // not carry over.
+    ACLManager dag2AclManager = new ACLManager(amAclManager, dag2User.getShortUserName(),
+        ACLInfo.getDefaultInstance());
+
+    assertFalse(dag2AclManager.checkDAGModifyAccess(user1));
+    assertFalse(dag2AclManager.checkDAGViewAccess(user1));
+    assertFalse(dag2AclManager.checkDAGModifyAccess(user2));
+    assertFalse(dag2AclManager.checkDAGViewAccess(user2));
+
+    // AM-level manager should also be untouched by DAG 1's grants.
+    assertFalse(amAclManager.checkAccess(user1, ACLType.DAG_MODIFY_ACL));
+    assertFalse(amAclManager.checkAccess(user2, ACLType.DAG_MODIFY_ACL));
+  }
+
+  /**
+   * The per-DAG constructor must deep-copy the AM's ACL sets. Otherwise a
+   * later {@code users.get(ACLType.AM_VIEW_ACL).add(...)} inside the DAG-scoped
+   * manager would silently mutate the AM's global ACLs (and every sibling
+   * DAG's view of them). This test simulates that mutation via reflection so
+   * it catches the shallow-copy regression even though today's code paths
+   * only replace entries via {@code put}.
+   */
+  @Test
+  @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+  public void testDAGConstructorDeepCopiesAMACLs() throws Exception {
+    UserGroupInformation amUser = UserGroupInformation.createUserForTesting("amUser", noGroups);
+    UserGroupInformation dagUser = UserGroupInformation.createUserForTesting("dagUser", noGroups);
+    UserGroupInformation user1 = UserGroupInformation.createUserForTesting("user1", noGroups);
+    UserGroupInformation leakedUser = UserGroupInformation.createUserForTesting("leakedUser", noGroups);
+
+    Configuration conf = new Configuration(false);
+    conf.set(TezConfiguration.TEZ_AM_VIEW_ACLS, user1.getShortUserName());
+    ACLManager amAclManager = new ACLManager(amUser.getShortUserName(), conf);
+
+    ACLManager dagAclManager =
+        new ACLManager(amAclManager, dagUser.getShortUserName(), ACLInfo.getDefaultInstance());
+
+    // Reach into the DAG manager and mutate the AM_VIEW_ACL set in place.
+    // With a deep copy this only affects the DAG manager; with a shallow
+    // copy the AM manager would see the change too.
+    Field usersField = ACLManager.class.getDeclaredField("users");
+    usersField.setAccessible(true);
+    @SuppressWarnings("unchecked")
+    Map<ACLType, Set<String>> dagUsers = (Map<ACLType, Set<String>>) usersField.get(dagAclManager);
+    dagUsers.get(ACLType.AM_VIEW_ACL).add(leakedUser.getShortUserName());
+
+    // The AM manager must NOT have picked up the new user.
+    assertFalse(amAclManager.checkAMViewAccess(leakedUser));
+  }
+
+  /**
+   * The per-DAG constructor must tolerate a null ACLInfo (the disabled-ACLs
+   * test above already relies on this via a different code path, but with
+   * ACLs enabled the null must still not NPE).
+   */
+  @Test
+  @Timeout(value = 5000, unit = TimeUnit.MILLISECONDS)
+  public void testDAGConstructorAcceptsNullACLInfo() {
+    UserGroupInformation amUser = UserGroupInformation.createUserForTesting("amUser", noGroups);
+    UserGroupInformation dagUser = UserGroupInformation.createUserForTesting("dagUser", noGroups);
+    UserGroupInformation other = UserGroupInformation.createUserForTesting("other", noGroups);
+
+    Configuration conf = new Configuration(false);
+    ACLManager amAclManager = new ACLManager(amUser.getShortUserName(), conf);
+    ACLManager dagAclManager =
+        new ACLManager(amAclManager, dagUser.getShortUserName(), null);
+
+    assertTrue(dagAclManager.checkDAGViewAccess(dagUser));
+    assertTrue(dagAclManager.checkDAGModifyAccess(dagUser));
+    assertFalse(dagAclManager.checkDAGViewAccess(other));
+    assertFalse(dagAclManager.checkDAGModifyAccess(other));
+  }
 }