YARN-5836. Malicious AM can kill containers of other apps running in any node its containers are running. Contributed by Botong Huang

(cherry picked from commit 59bfcbf357)

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/ContainerManagerImpl.java

@@ -1291,18 +1291,21 @@ protected void authorizeGetAndStopContainerRequest(ContainerId containerId,
     if ((!nmTokenAppId.equals(containerId.getApplicationAttemptId().getApplicationId()))
         || (container != null && !nmTokenAppId.equals(container
             .getContainerId().getApplicationAttemptId().getApplicationId()))) {
+      String msg;
       if (stopRequest) {
-        LOG.warn(identifier.getApplicationAttemptId()
+        msg = identifier.getApplicationAttemptId()
             + " attempted to stop non-application container : "
-            + container.getContainerId());
+            + containerId;
         NMAuditLogger.logFailure("UnknownUser", AuditConstants.STOP_CONTAINER,
             "ContainerManagerImpl", "Trying to stop unknown container!",
-          nmTokenAppId, container.getContainerId());
+            nmTokenAppId, containerId);
       } else {
-        LOG.warn(identifier.getApplicationAttemptId()
+        msg = identifier.getApplicationAttemptId()
             + " attempted to get status for non-application container : "
-            + container.getContainerId());
+            + containerId;
       }
+      LOG.warn(msg);
+      throw RPCUtil.getRemoteException(msg);
     }
   }
 

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/TestContainerManagerWithLCE.java

@@ -189,6 +189,17 @@ public void testMultipleContainersStopAndGetStatus() throws Exception {
     super.testMultipleContainersStopAndGetStatus();
   }
 
+  @Override
+  public void testUnauthorizedRequests() throws IOException, YarnException {
+    // Don't run the test if the binary is not available.
+    if (!shouldRunTest()) {
+      LOG.info("LCE binary path is not passed. Not running the test");
+      return;
+    }
+    LOG.info("Running testUnauthorizedRequests");
+    super.testUnauthorizedRequests();
+  }
+
   @Override
   public void testStartContainerFailureWithUnknownAuxService() throws Exception {
     // Don't run the test if the binary is not available.

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/BaseContainerManagerTest.java

@@ -417,10 +417,16 @@ public static Token createContainerToken(ContainerId cId, long rmIdentifier,
   }
 
   public static ContainerId createContainerId(int id) {
-    ApplicationId appId = ApplicationId.newInstance(0, 0);
+    // Use default appId = 0
+    return createContainerId(id, 0);
+  }
+
+  public static ContainerId createContainerId(int cId, int aId) {
+    ApplicationId appId = ApplicationId.newInstance(0, aId);
     ApplicationAttemptId appAttemptId =
         ApplicationAttemptId.newInstance(appId, 1);
-    ContainerId containerId = ContainerId.newContainerId(appAttemptId, id);
+    ContainerId containerId =
+        ContainerId.newContainerId(appAttemptId, cId);
     return containerId;
   }
 }

hadoop-yarn-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/TestContainerManager.java

@@ -94,10 +94,12 @@
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.TestAuxServices.ServiceA;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.application.ApplicationState;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container;
+import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.ContainerImpl;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.launcher.ContainerLaunch;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ContainerLocalizer;
 import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.ResourceLocalizationService;
 import org.apache.hadoop.yarn.server.nodemanager.executor.ContainerSignalContext;
+import org.apache.hadoop.yarn.server.utils.BuilderUtils;
 import org.junit.Assert;
 import org.junit.Before;
 import org.junit.Test;
@@ -143,14 +145,6 @@ protected UserGroupInformation getRemoteUgi() throws YarnException {
           .getKeyId()));
         return ugi;
       }
-
-      @Override
-      protected void authorizeGetAndStopContainerRequest(ContainerId containerId,
-          Container container, boolean stopRequest, NMTokenIdentifier identifier) throws YarnException {
-        if(container == null || container.getUser().equals("Fail")){
-          throw new YarnException("Reject this container");
-        }
-      }
     };
   }
 
@@ -1263,13 +1257,12 @@ public void testMultipleContainersStopAndGetStatus() throws Exception {
 
     List<ContainerId> containerIds = new ArrayList<>();
     for (int i = 0; i < 10; i++) {
-      ContainerId cId = createContainerId(i);
+      ContainerId cId;
-      String user = null;
       if ((i & 1) == 0) {
-        // container with even id fail
+        // Containers with even id belong to an unauthorized app
-        user = "Fail";
+        cId = createContainerId(i, 1);
       } else {
-        user = "Pass";
+        cId = createContainerId(i, 0);
       }
       Token containerToken =
           createContainerToken(cId, DUMMY_RM_IDENTIFIER, context.getNodeId(),
@@ -1302,7 +1295,7 @@ public void testMultipleContainersStopAndGetStatus() throws Exception {
       // Containers with even id should fail.
       Assert.assertEquals(0, entry.getKey().getContainerId() & 1);
       Assert.assertTrue(entry.getValue().getMessage()
-        .contains("Reject this container"));
+          .contains("attempted to get status for non-application container"));
     }
 
     // stop containers
@@ -1322,10 +1315,70 @@ public void testMultipleContainersStopAndGetStatus() throws Exception {
       // Containers with even id should fail.
       Assert.assertEquals(0, entry.getKey().getContainerId() & 1);
       Assert.assertTrue(entry.getValue().getMessage()
-        .contains("Reject this container"));
+          .contains("attempted to stop non-application container"));
     }
   }
 
+  @Test
+  public void testUnauthorizedRequests() throws IOException, YarnException {
+    containerManager.start();
+
+    // Create a containerId that belongs to an unauthorized appId
+    ContainerId cId = createContainerId(0, 1);
+
+    // startContainers()
+    ContainerLaunchContext containerLaunchContext =
+        recordFactory.newRecordInstance(ContainerLaunchContext.class);
+    StartContainerRequest scRequest =
+        StartContainerRequest.newInstance(containerLaunchContext,
+            createContainerToken(cId, DUMMY_RM_IDENTIFIER, context.getNodeId(),
+                user, context.getContainerTokenSecretManager()));
+    List<StartContainerRequest> list = new ArrayList<>();
+    list.add(scRequest);
+    StartContainersRequest allRequests =
+        StartContainersRequest.newInstance(list);
+    StartContainersResponse startResponse =
+        containerManager.startContainers(allRequests);
+
+    Assert.assertFalse("Should not be authorized to start container",
+        startResponse.getSuccessfullyStartedContainers().contains(cId));
+    Assert.assertTrue("Start container request should fail",
+        startResponse.getFailedRequests().containsKey(cId));
+
+    // Insert the containerId into context, make it as if it is running
+    ContainerTokenIdentifier containerTokenIdentifier =
+        BuilderUtils.newContainerTokenIdentifier(scRequest.getContainerToken());
+    Container container = new ContainerImpl(conf, null, containerLaunchContext,
+        null, metrics, containerTokenIdentifier, context);
+    context.getContainers().put(cId, container);
+
+    // stopContainers()
+    List<ContainerId> containerIds = new ArrayList<>();
+    containerIds.add(cId);
+    StopContainersRequest stopRequest =
+        StopContainersRequest.newInstance(containerIds);
+    StopContainersResponse stopResponse =
+        containerManager.stopContainers(stopRequest);
+
+    Assert.assertFalse("Should not be authorized to stop container",
+        stopResponse.getSuccessfullyStoppedContainers().contains(cId));
+    Assert.assertTrue("Stop container request should fail",
+        stopResponse.getFailedRequests().containsKey(cId));
+
+    // getContainerStatuses()
+    containerIds = new ArrayList<>();
+    containerIds.add(cId);
+    GetContainerStatusesRequest request =
+        GetContainerStatusesRequest.newInstance(containerIds);
+    GetContainerStatusesResponse response =
+        containerManager.getContainerStatuses(request);
+
+    Assert.assertEquals("Should not be authorized to get container status",
+        response.getContainerStatuses().size(), 0);
+    Assert.assertTrue("Get status request should fail",
+        response.getFailedRequests().containsKey(cId));
+  }
+
   @Test
   public void testStartContainerFailureWithUnknownAuxService() throws Exception {
     conf.setStrings(YarnConfiguration.NM_AUX_SERVICES,