YARN-5554. MoveApplicationAcrossQueues does not check user permission on the target queue

(Contributed by Wilfred Spiegelenburg via Daniel Templeton)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/ClientRMService.java

@@ -1186,20 +1186,35 @@ public class ClientRMService extends AbstractService implements
           + callerUGI.getShortUserName() + " cannot perform operation "
           + ApplicationAccessType.MODIFY_APP.name() + " on " + applicationId));
     }
-    
+
+    String targetQueue = request.getTargetQueue();
+    if (!accessToTargetQueueAllowed(callerUGI, application, targetQueue)) {
+      RMAuditLogger.logFailure(callerUGI.getShortUserName(),
+          AuditConstants.MOVE_APP_REQUEST, "Target queue doesn't exist or user"
+              + " doesn't have permissions to submit to target queue: "
+              + targetQueue, "ClientRMService",
+          AuditConstants.UNAUTHORIZED_USER, applicationId);
+      throw RPCUtil.getRemoteException(new AccessControlException("User "
+          + callerUGI.getShortUserName() + " cannot submit applications to"
+          + " target queue or the target queue doesn't exist: "
+          + targetQueue + " while moving " + applicationId));
+    }
+
     // Moves only allowed when app is in a state that means it is tracked by
     // the scheduler. Introducing SUBMITTED state also to this list as there
     // could be a corner scenario that app may not be in Scheduler in SUBMITTED
     // state.
     if (!ACTIVE_APP_STATES.contains(application.getState())) {
-      String msg = "App in " + application.getState() + " state cannot be moved.";
+      String msg = "App in " + application.getState() +
+          " state cannot be moved.";
       RMAuditLogger.logFailure(callerUGI.getShortUserName(),
           AuditConstants.MOVE_APP_REQUEST, "UNKNOWN", "ClientRMService", msg);
       throw new YarnException(msg);
     }
 
     try {
-      this.rmAppManager.moveApplicationAcrossQueue(applicationId, request.getTargetQueue());
+      this.rmAppManager.moveApplicationAcrossQueue(applicationId,
+          request.getTargetQueue());
     } catch (YarnException ex) {
       RMAuditLogger.logFailure(callerUGI.getShortUserName(),
           AuditConstants.MOVE_APP_REQUEST, "UNKNOWN", "ClientRMService",
@@ -1214,6 +1229,24 @@ public class ClientRMService extends AbstractService implements
     return response;
   }
 
+  /**
+   * Check if the submission of an application to the target queue is allowed.
+   * @param callerUGI the caller UGI
+   * @param application the application to move
+   * @param targetQueue the queue to move the application to
+   * @return true if submission is allowed, false otherwise
+   */
+  private boolean accessToTargetQueueAllowed(UserGroupInformation callerUGI,
+      RMApp application, String targetQueue) {
+    return
+        queueACLsManager.checkAccess(callerUGI,
+            QueueACL.SUBMIT_APPLICATIONS, application,
+            Server.getRemoteAddress(), null, targetQueue) ||
+        queueACLsManager.checkAccess(callerUGI,
+            QueueACL.ADMINISTER_QUEUE, application,
+            Server.getRemoteAddress(), null, targetQueue);
+  }
+
   private String getRenewerForToken(Token<RMDelegationTokenIdentifier> token)
       throws IOException {
     UserGroupInformation user = UserGroupInformation.getCurrentUser();

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/main/java/org/apache/hadoop/yarn/server/resourcemanager/security/QueueACLsManager.java

@@ -32,6 +32,8 @@ import org.apache.hadoop.yarn.server.resourcemanager.scheduler.ResourceScheduler
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.SchedulerUtils;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CSQueue;
 import org.apache.hadoop.yarn.server.resourcemanager.scheduler.capacity.CapacityScheduler;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FSQueue;
+import org.apache.hadoop.yarn.server.resourcemanager.scheduler.fair.FairScheduler;
 
 import java.util.List;
 
@@ -64,9 +66,11 @@ public class QueueACLsManager {
     if (scheduler instanceof CapacityScheduler) {
       CSQueue queue = ((CapacityScheduler) scheduler).getQueue(app.getQueue());
       if (queue == null) {
-        // Application exists but the associated queue does not exist.
+        // The application exists but the associated queue does not exist.
-        // This may be due to queue is removed after RM restarts. Here, we choose
+        // This may be due to a queue that is not defined when the RM restarts.
-        // to allow users to be able to view the apps for removed queue.
+        // At this point we choose to log the fact and allow users to access
+        // and view the apps in a removed queue. This should only happen on
+        // application recovery.
         LOG.error("Queue " + app.getQueue() + " does not exist for " + app
             .getApplicationId());
         return true;
@@ -80,4 +84,63 @@ public class QueueACLsManager {
       return scheduler.checkAccess(callerUGI, acl, app.getQueue());
     }
   }
+
+  /**
+   * Check access to a targetQueue in the case of a move of an application.
+   * The application cannot contain the destination queue since it has not
+   * been moved yet, thus need to pass it in separately.
+   *
+   * @param callerUGI the caller UGI
+   * @param acl the acl for the Queue to check
+   * @param app the application to move
+   * @param remoteAddress server ip address
+   * @param forwardedAddresses forwarded adresses
+   * @param targetQueue the name of the queue to move the application to
+   * @return true: if submission is allowed and queue exists,
+   *         false: in all other cases (also non existing target queue)
+   */
+  public boolean checkAccess(UserGroupInformation callerUGI, QueueACL acl,
+      RMApp app, String remoteAddress, List<String> forwardedAddresses,
+      String targetQueue) {
+    if (!isACLsEnable) {
+      return true;
+    }
+
+    // Based on the discussion in YARN-5554 detail on why there are two
+    // versions:
+    // The access check inside these calls is currently scheduler dependent.
+    // This is due to the extra parameters needed for the CS case which are not
+    // in the version defined in the YarnScheduler interface. The second
+    // version is added for the moving the application case. The check has
+    // extra logging to distinguish between the queue not existing in the
+    // application move request case and the real access denied case.
+
+    if (scheduler instanceof CapacityScheduler) {
+      CSQueue queue = ((CapacityScheduler) scheduler).getQueue(targetQueue);
+      if (queue == null) {
+        LOG.warn("Target queue " + targetQueue
+            + " does not exist while trying to move "
+            + app.getApplicationId());
+        return false;
+      }
+      return authorizer.checkPermission(
+          new AccessRequest(queue.getPrivilegedEntity(), callerUGI,
+              SchedulerUtils.toAccessType(acl),
+              app.getApplicationId().toString(), app.getName(),
+              remoteAddress, forwardedAddresses));
+    } else if (scheduler instanceof FairScheduler) {
+      FSQueue queue = ((FairScheduler) scheduler).getQueueManager().
+          getQueue(targetQueue);
+      if (queue == null) {
+        LOG.warn("Target queue " + targetQueue
+            + " does not exist while trying to move "
+            + app.getApplicationId());
+        return false;
+      }
+      return scheduler.checkAccess(callerUGI, acl, targetQueue);
+    } else {
+      // Any other scheduler just try
+      return scheduler.checkAccess(callerUGI, acl, targetQueue);
+    }
+  }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-resourcemanager/src/test/java/org/apache/hadoop/yarn/server/resourcemanager/TestClientRMService.java

@@ -22,6 +22,7 @@ import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertTrue;
 import static org.mockito.Matchers.any;
 import static org.mockito.Matchers.anyBoolean;
+import static org.mockito.Matchers.anyListOf;
 import static org.mockito.Matchers.anyString;
 import static org.mockito.Matchers.eq;
 import static org.mockito.Mockito.doReturn;
@@ -33,6 +34,8 @@ import java.io.File;
 import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.InetSocketAddress;
+import java.security.AccessControlException;
+import java.security.PrivilegedExceptionAction;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.EnumSet;
@@ -155,6 +158,8 @@ import org.junit.Test;
 
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.Sets;
+import org.mockito.invocation.InvocationOnMock;
+import org.mockito.stubbing.Answer;
 
 public class TestClientRMService {
 
@@ -572,10 +577,261 @@ public class TestClientRMService {
     ApplicationId applicationId =
         BuilderUtils.newApplicationId(System.currentTimeMillis(), 0);
     MoveApplicationAcrossQueuesRequest request =
-        MoveApplicationAcrossQueuesRequest.newInstance(applicationId, "newqueue");
+        MoveApplicationAcrossQueuesRequest.newInstance(applicationId,
+            "newqueue");
     rmService.moveApplicationAcrossQueues(request);
   }
 
+  @Test
+  public void testMoveApplicationSubmitTargetQueue() throws Exception {
+    // move the application as the owner
+    ApplicationId applicationId = getApplicationId(1);
+    UserGroupInformation aclUGI = UserGroupInformation.getCurrentUser();
+    QueueACLsManager queueACLsManager = getQueueAclManager("allowed_queue",
+        QueueACL.SUBMIT_APPLICATIONS, aclUGI);
+    ApplicationACLsManager appAclsManager = getAppAclManager();
+
+    ClientRMService rmService = createClientRMServiceForMoveApplicationRequest(
+        applicationId, aclUGI.getShortUserName(), appAclsManager,
+        queueACLsManager);
+
+    // move as the owner queue in the acl
+    MoveApplicationAcrossQueuesRequest moveAppRequest =
+        MoveApplicationAcrossQueuesRequest.
+        newInstance(applicationId, "allowed_queue");
+    rmService.moveApplicationAcrossQueues(moveAppRequest);
+
+    // move as the owner queue not in the acl
+    moveAppRequest = MoveApplicationAcrossQueuesRequest.newInstance(
+        applicationId, "not_allowed");
+
+    try {
+      rmService.moveApplicationAcrossQueues(moveAppRequest);
+      Assert.fail("The request should fail with an AccessControlException");
+    } catch (YarnException rex) {
+      Assert.assertTrue("AccessControlException is expected",
+          rex.getCause() instanceof AccessControlException);
+    }
+
+    // ACL is owned by "moveuser", move is performed as a different user
+    aclUGI = UserGroupInformation.createUserForTesting("moveuser",
+        new String[]{});
+    queueACLsManager = getQueueAclManager("move_queue",
+        QueueACL.SUBMIT_APPLICATIONS, aclUGI);
+    appAclsManager = getAppAclManager();
+    ClientRMService rmService2 =
+        createClientRMServiceForMoveApplicationRequest(applicationId,
+            aclUGI.getShortUserName(), appAclsManager, queueACLsManager);
+
+    // access to the queue not OK: user not allowed in this queue
+    MoveApplicationAcrossQueuesRequest moveAppRequest2 =
+        MoveApplicationAcrossQueuesRequest.
+            newInstance(applicationId, "move_queue");
+    try {
+      rmService2.moveApplicationAcrossQueues(moveAppRequest2);
+      Assert.fail("The request should fail with an AccessControlException");
+    } catch (YarnException rex) {
+      Assert.assertTrue("AccessControlException is expected",
+          rex.getCause() instanceof AccessControlException);
+    }
+
+    // execute the move as the acl owner
+    // access to the queue OK: user allowed in this queue
+    aclUGI.doAs(new PrivilegedExceptionAction<Object>() {
+      @Override
+      public Object run() throws Exception {
+        return rmService2.moveApplicationAcrossQueues(moveAppRequest2);
+      }
+    });
+  }
+
+  @Test
+  public void testMoveApplicationAdminTargetQueue() throws Exception {
+    ApplicationId applicationId = getApplicationId(1);
+    UserGroupInformation aclUGI = UserGroupInformation.getCurrentUser();
+    QueueACLsManager queueAclsManager = getQueueAclManager("allowed_queue",
+        QueueACL.ADMINISTER_QUEUE, aclUGI);
+    ApplicationACLsManager appAclsManager = getAppAclManager();
+    ClientRMService rmService =
+        createClientRMServiceForMoveApplicationRequest(applicationId,
+            aclUGI.getShortUserName(), appAclsManager, queueAclsManager);
+
+    // user is admin move to queue in acl
+    MoveApplicationAcrossQueuesRequest moveAppRequest =
+        MoveApplicationAcrossQueuesRequest.newInstance(applicationId,
+            "allowed_queue");
+    rmService.moveApplicationAcrossQueues(moveAppRequest);
+
+    // user is admin move to queue not in acl
+    moveAppRequest = MoveApplicationAcrossQueuesRequest.newInstance(
+        applicationId, "not_allowed");
+
+    try {
+      rmService.moveApplicationAcrossQueues(moveAppRequest);
+      Assert.fail("The request should fail with an AccessControlException");
+    } catch (YarnException rex) {
+      Assert.assertTrue("AccessControlException is expected",
+          rex.getCause() instanceof AccessControlException);
+    }
+
+    // ACL is owned by "moveuser", move is performed as a different user
+    aclUGI = UserGroupInformation.createUserForTesting("moveuser",
+        new String[]{});
+    queueAclsManager = getQueueAclManager("move_queue",
+        QueueACL.ADMINISTER_QUEUE, aclUGI);
+    appAclsManager = getAppAclManager();
+    ClientRMService rmService2 =
+        createClientRMServiceForMoveApplicationRequest(applicationId,
+            aclUGI.getShortUserName(), appAclsManager, queueAclsManager);
+
+    // no access to this queue
+    MoveApplicationAcrossQueuesRequest moveAppRequest2 =
+        MoveApplicationAcrossQueuesRequest.
+            newInstance(applicationId, "move_queue");
+
+    try {
+      rmService2.moveApplicationAcrossQueues(moveAppRequest2);
+      Assert.fail("The request should fail with an AccessControlException");
+    } catch (YarnException rex) {
+      Assert.assertTrue("AccessControlException is expected",
+          rex.getCause() instanceof AccessControlException);
+    }
+
+    // execute the move as the acl owner
+    // access to the queue OK: user allowed in this queue
+    aclUGI.doAs(new PrivilegedExceptionAction<Object>() {
+      @Override
+      public Object run() throws Exception {
+        return rmService2.moveApplicationAcrossQueues(moveAppRequest2);
+      }
+    });
+  }
+
+  @Test (expected = YarnException.class)
+  public void testNonExistingQueue() throws Exception {
+    ApplicationId applicationId = getApplicationId(1);
+    UserGroupInformation aclUGI = UserGroupInformation.getCurrentUser();
+    QueueACLsManager queueAclsManager = getQueueAclManager();
+    ApplicationACLsManager appAclsManager = getAppAclManager();
+    ClientRMService rmService =
+        createClientRMServiceForMoveApplicationRequest(applicationId,
+            aclUGI.getShortUserName(), appAclsManager, queueAclsManager);
+
+    MoveApplicationAcrossQueuesRequest moveAppRequest =
+        MoveApplicationAcrossQueuesRequest.newInstance(applicationId,
+            "unknown_queue");
+    rmService.moveApplicationAcrossQueues(moveAppRequest);
+  }
+
+  /**
+   * Create an instance of ClientRMService for testing
+   * moveApplicationAcrossQueues requests.
+   * @param applicationId the application
+   * @return ClientRMService
+   */
+  private ClientRMService createClientRMServiceForMoveApplicationRequest(
+      ApplicationId applicationId, String appOwner,
+      ApplicationACLsManager appAclsManager, QueueACLsManager queueAclsManager)
+      throws IOException {
+    RMApp app = mock(RMApp.class);
+    when(app.getUser()).thenReturn(appOwner);
+    when(app.getState()).thenReturn(RMAppState.RUNNING);
+    ConcurrentHashMap<ApplicationId, RMApp> apps = new ConcurrentHashMap<>();
+    apps.put(applicationId, app);
+
+    RMContext rmContext = mock(RMContext.class);
+    when(rmContext.getRMApps()).thenReturn(apps);
+
+    RMAppManager rmAppManager = mock(RMAppManager.class);
+    return new ClientRMService(rmContext, null, rmAppManager, appAclsManager,
+        queueAclsManager, null);
+  }
+
+  /**
+   * Plain application acl manager that always returns true.
+   * @return ApplicationACLsManager
+   */
+  private ApplicationACLsManager getAppAclManager() {
+    ApplicationACLsManager aclsManager = mock(ApplicationACLsManager.class);
+    when(aclsManager.checkAccess(
+        any(UserGroupInformation.class),
+        any(ApplicationAccessType.class),
+        any(String.class),
+        any(ApplicationId.class))).thenReturn(true);
+    return aclsManager;
+  }
+
+  /**
+   * Generate the Queue acl.
+   * @param allowedQueue the queue to allow the move to
+   * @param queueACL the acl to check: submit app or queue admin
+   * @param aclUser the user to check
+   * @return QueueACLsManager
+   */
+  private QueueACLsManager getQueueAclManager(String allowedQueue,
+      QueueACL queueACL, UserGroupInformation aclUser) throws IOException {
+    // ACL that checks the queue is allowed
+    QueueACLsManager queueACLsManager = mock(QueueACLsManager.class);
+    when(queueACLsManager.checkAccess(
+        any(UserGroupInformation.class),
+        any(QueueACL.class),
+        any(RMApp.class),
+        any(String.class),
+        anyListOf(String.class))).thenAnswer(new Answer<Boolean>() {
+            @Override
+            public Boolean answer(InvocationOnMock invocationOnMock) {
+              final UserGroupInformation user =
+                  (UserGroupInformation) invocationOnMock.getArguments()[0];
+              final QueueACL acl =
+                  (QueueACL) invocationOnMock.getArguments()[1];
+              return (queueACL.equals(acl) &&
+                  aclUser.getShortUserName().equals(user.getShortUserName()));
+            }
+        });
+
+    when(queueACLsManager.checkAccess(
+        any(UserGroupInformation.class),
+        any(QueueACL.class),
+        any(RMApp.class),
+        any(String.class),
+        anyListOf(String.class),
+        any(String.class))).thenAnswer(new Answer<Boolean>() {
+          @Override
+          public Boolean answer(InvocationOnMock invocationOnMock) {
+            final UserGroupInformation user =
+                (UserGroupInformation) invocationOnMock.getArguments()[0];
+            final QueueACL acl = (QueueACL) invocationOnMock.getArguments()[1];
+            final String queue = (String) invocationOnMock.getArguments()[5];
+            return (allowedQueue.equals(queue) && queueACL.equals(acl) &&
+                aclUser.getShortUserName().equals(user.getShortUserName()));
+          }
+        });
+    return queueACLsManager;
+  }
+
+  /**
+   * QueueACLsManager that always returns false when a target queue is passed
+   * in and true for other checks to simulate a missing queue.
+   * @return QueueACLsManager
+   */
+  private QueueACLsManager getQueueAclManager() {
+    QueueACLsManager queueACLsManager = mock(QueueACLsManager.class);
+    when(queueACLsManager.checkAccess(
+        any(UserGroupInformation.class),
+        any(QueueACL.class),
+        any(RMApp.class),
+        any(String.class),
+        anyListOf(String.class),
+        any(String.class))).thenReturn(false);
+    when(queueACLsManager.checkAccess(
+        any(UserGroupInformation.class),
+        any(QueueACL.class),
+        any(RMApp.class),
+        any(String.class),
+        anyListOf(String.class))).thenReturn(true);
+    return queueACLsManager;
+  }
+
   @Test
   public void testGetQueueInfo() throws Exception {
     YarnScheduler yarnScheduler = mock(YarnScheduler.class);