| /** |
| * Licensed to the Apache Software Foundation (ASF) under one |
| * or more contributor license agreements. See the NOTICE file |
| * distributed with this work for additional information |
| * regarding copyright ownership. The ASF licenses this file |
| * to you under the Apache License, Version 2.0 (the |
| * "License"); you may not use this file except in compliance |
| * with the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.hadoop.yarn.server; |
| |
| import static org.junit.Assert.fail; |
| |
| import java.io.File; |
| import java.io.IOException; |
| import java.net.InetSocketAddress; |
| import java.util.ArrayList; |
| import java.util.Arrays; |
| import java.util.Collection; |
| import java.util.LinkedList; |
| import java.util.List; |
| |
| import org.apache.commons.logging.Log; |
| import org.apache.commons.logging.LogFactory; |
| import org.apache.hadoop.conf.Configuration; |
| import org.apache.hadoop.fs.CommonConfigurationKeysPublic; |
| import org.apache.hadoop.io.DataInputBuffer; |
| import org.apache.hadoop.minikdc.KerberosSecurityTestcase; |
| import org.apache.hadoop.net.NetUtils; |
| import org.apache.hadoop.security.UserGroupInformation; |
| import org.apache.hadoop.security.token.SecretManager.InvalidToken; |
| import org.apache.hadoop.yarn.api.ContainerManagementProtocol; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.GetContainerStatusesResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.StartContainerRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.StartContainersRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.StartContainersResponse; |
| import org.apache.hadoop.yarn.api.protocolrecords.StopContainersRequest; |
| import org.apache.hadoop.yarn.api.protocolrecords.StopContainersResponse; |
| import org.apache.hadoop.yarn.api.records.ApplicationAttemptId; |
| import org.apache.hadoop.yarn.api.records.ApplicationId; |
| import org.apache.hadoop.yarn.api.records.ContainerId; |
| import org.apache.hadoop.yarn.api.records.ContainerLaunchContext; |
| import org.apache.hadoop.yarn.api.records.ContainerState; |
| import org.apache.hadoop.yarn.api.records.NodeId; |
| import org.apache.hadoop.yarn.api.records.Priority; |
| import org.apache.hadoop.yarn.api.records.Resource; |
| import org.apache.hadoop.yarn.api.records.SerializedException; |
| import org.apache.hadoop.yarn.api.records.Token; |
| import org.apache.hadoop.yarn.client.NMProxy; |
| import org.apache.hadoop.yarn.conf.YarnConfiguration; |
| import org.apache.hadoop.yarn.exceptions.YarnException; |
| import org.apache.hadoop.yarn.factories.RecordFactory; |
| import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider; |
| import org.apache.hadoop.yarn.ipc.YarnRPC; |
| import org.apache.hadoop.yarn.security.ContainerTokenIdentifier; |
| import org.apache.hadoop.yarn.security.NMTokenIdentifier; |
| import org.apache.hadoop.yarn.server.nodemanager.Context; |
| import org.apache.hadoop.yarn.server.nodemanager.NodeManager; |
| import org.apache.hadoop.yarn.server.nodemanager.security.NMTokenSecretManagerInNM; |
| import org.apache.hadoop.yarn.server.nodemanager.containermanager.container.Container; |
| import org.apache.hadoop.yarn.server.resourcemanager.rmapp.MockRMApp; |
| import org.apache.hadoop.yarn.server.resourcemanager.rmapp.RMAppState; |
| import org.apache.hadoop.yarn.server.resourcemanager.security.NMTokenSecretManagerInRM; |
| import org.apache.hadoop.yarn.server.resourcemanager.security.RMContainerTokenSecretManager; |
| import org.apache.hadoop.yarn.server.security.BaseNMTokenSecretManager; |
| import org.apache.hadoop.yarn.server.utils.BuilderUtils; |
| import org.apache.hadoop.yarn.util.ConverterUtils; |
| import org.apache.hadoop.yarn.util.Records; |
| import org.junit.After; |
| import org.junit.Assert; |
| import org.junit.Before; |
| import org.junit.Test; |
| import org.junit.runner.RunWith; |
| import org.junit.runners.Parameterized; |
| import org.junit.runners.Parameterized.Parameters; |
| |
| |
| @RunWith(Parameterized.class) |
| public class TestContainerManagerSecurity extends KerberosSecurityTestcase { |
| |
| static Log LOG = LogFactory.getLog(TestContainerManagerSecurity.class); |
| static final RecordFactory recordFactory = RecordFactoryProvider |
| .getRecordFactory(null); |
| private static MiniYARNCluster yarnCluster; |
| private static final File testRootDir = new File("target", |
| TestContainerManagerSecurity.class.getName() + "-root"); |
| private static File httpSpnegoKeytabFile = new File(testRootDir, |
| "httpSpnegoKeytabFile.keytab"); |
| private static String httpSpnegoPrincipal = "HTTP/localhost@EXAMPLE.COM"; |
| |
| private Configuration conf; |
| |
| @Before |
| public void setUp() throws Exception { |
| testRootDir.mkdirs(); |
| httpSpnegoKeytabFile.deleteOnExit(); |
| getKdc().createPrincipal(httpSpnegoKeytabFile, httpSpnegoPrincipal); |
| UserGroupInformation.setConfiguration(conf); |
| |
| yarnCluster = |
| new MiniYARNCluster(TestContainerManagerSecurity.class.getName(), 1, 1, |
| 1); |
| yarnCluster.init(conf); |
| yarnCluster.start(); |
| } |
| |
| @After |
| public void tearDown() { |
| if (yarnCluster != null) { |
| yarnCluster.stop(); |
| yarnCluster = null; |
| } |
| testRootDir.delete(); |
| } |
| |
| /* |
| * Run two tests: one with no security ("simple") and one with "Secure" |
| * The first parameter is just the test name to make it easier to debug |
| * and to give details in say an IDE. The second is the configuraiton |
| * object to use. |
| */ |
| @Parameters(name = "{0}") |
| public static Collection<Object[]> configs() { |
| Configuration configurationWithoutSecurity = new Configuration(); |
| configurationWithoutSecurity.set( |
| CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, "simple"); |
| |
| Configuration configurationWithSecurity = new Configuration(); |
| configurationWithSecurity.set( |
| CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION, "kerberos"); |
| configurationWithSecurity.set( |
| YarnConfiguration.RM_WEBAPP_SPNEGO_USER_NAME_KEY, httpSpnegoPrincipal); |
| configurationWithSecurity.set( |
| YarnConfiguration.RM_WEBAPP_SPNEGO_KEYTAB_FILE_KEY, |
| httpSpnegoKeytabFile.getAbsolutePath()); |
| configurationWithSecurity.set( |
| YarnConfiguration.NM_WEBAPP_SPNEGO_USER_NAME_KEY, httpSpnegoPrincipal); |
| configurationWithSecurity.set( |
| YarnConfiguration.NM_WEBAPP_SPNEGO_KEYTAB_FILE_KEY, |
| httpSpnegoKeytabFile.getAbsolutePath()); |
| |
| return Arrays.asList(new Object[][] { |
| {"Simple", configurationWithoutSecurity}, |
| {"Secure", configurationWithSecurity}}); |
| } |
| |
| public TestContainerManagerSecurity(String name, Configuration conf) { |
| LOG.info("RUNNING TEST " + name); |
| conf.setLong(YarnConfiguration.RM_AM_EXPIRY_INTERVAL_MS, 100000L); |
| this.conf = conf; |
| } |
| |
| @Test |
| public void testContainerManager() throws Exception { |
| |
| // TestNMTokens. |
| testNMTokens(conf); |
| |
| // Testing for container token tampering |
| testContainerToken(conf); |
| |
| // Testing for container token tampering with epoch |
| testContainerTokenWithEpoch(conf); |
| |
| } |
| |
| /** |
| * Run a series of tests using different NMTokens. A configuration is |
| * provided for managing creating of the tokens and rpc. |
| */ |
| private void testNMTokens(Configuration testConf) throws Exception { |
| NMTokenSecretManagerInRM nmTokenSecretManagerRM = |
| yarnCluster.getResourceManager().getRMContext() |
| .getNMTokenSecretManager(); |
| NMTokenSecretManagerInNM nmTokenSecretManagerNM = |
| yarnCluster.getNodeManager(0).getNMContext().getNMTokenSecretManager(); |
| RMContainerTokenSecretManager containerTokenSecretManager = |
| yarnCluster.getResourceManager().getRMContext(). |
| getContainerTokenSecretManager(); |
| |
| NodeManager nm = yarnCluster.getNodeManager(0); |
| |
| waitForNMToReceiveNMTokenKey(nmTokenSecretManagerNM); |
| |
| // Both id should be equal. |
| Assert.assertEquals(nmTokenSecretManagerNM.getCurrentKey().getKeyId(), |
| nmTokenSecretManagerRM.getCurrentKey().getKeyId()); |
| |
| /* |
| * Below cases should be tested. |
| * 1) If Invalid NMToken is used then it should be rejected. |
| * 2) If valid NMToken but belonging to another Node is used then that |
| * too should be rejected. |
| * 3) NMToken for say appAttempt-1 is used for starting/stopping/retrieving |
| * status for container with containerId for say appAttempt-2 should |
| * be rejected. |
| * 4) After start container call is successful nmtoken should have been |
| * saved in NMTokenSecretManagerInNM. |
| * 5) If start container call was successful (no matter if container is |
| * still running or not), appAttempt->NMToken should be present in |
| * NMTokenSecretManagerInNM's cache. Any future getContainerStatus call |
| * for containerId belonging to that application attempt using |
| * applicationAttempt's older nmToken should not get any invalid |
| * nmToken error. (This can be best tested if we roll over NMToken |
| * master key twice). |
| */ |
| YarnRPC rpc = YarnRPC.create(testConf); |
| String user = "test"; |
| Resource r = Resource.newInstance(1024, 1); |
| |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| MockRMApp m = new MockRMApp(appId.getId(), appId.getClusterTimestamp(), |
| RMAppState.NEW); |
| yarnCluster.getResourceManager().getRMContext().getRMApps().put(appId, m); |
| ApplicationAttemptId validAppAttemptId = |
| ApplicationAttemptId.newInstance(appId, 1); |
| |
| ContainerId validContainerId = |
| ContainerId.newContainerId(validAppAttemptId, 0); |
| |
| NodeId validNode = yarnCluster.getNodeManager(0).getNMContext().getNodeId(); |
| NodeId invalidNode = NodeId.newInstance("InvalidHost", 1234); |
| |
| |
| org.apache.hadoop.yarn.api.records.Token validNMToken = |
| nmTokenSecretManagerRM.createNMToken(validAppAttemptId, validNode, user); |
| |
| org.apache.hadoop.yarn.api.records.Token validContainerToken = |
| containerTokenSecretManager.createContainerToken(validContainerId, |
| 0, validNode, user, r, Priority.newInstance(10), 1234); |
| ContainerTokenIdentifier identifier = |
| BuilderUtils.newContainerTokenIdentifier(validContainerToken); |
| Assert.assertEquals(Priority.newInstance(10), identifier.getPriority()); |
| Assert.assertEquals(1234, identifier.getCreationTime()); |
| |
| StringBuilder sb; |
| // testInvalidNMToken ... creating NMToken using different secret manager. |
| |
| NMTokenSecretManagerInRM tempManager = new NMTokenSecretManagerInRM(testConf); |
| tempManager.rollMasterKey(); |
| do { |
| tempManager.rollMasterKey(); |
| tempManager.activateNextMasterKey(); |
| // Making sure key id is different. |
| } while (tempManager.getCurrentKey().getKeyId() == nmTokenSecretManagerRM |
| .getCurrentKey().getKeyId()); |
| |
| // Testing that NM rejects the requests when we don't send any token. |
| if (UserGroupInformation.isSecurityEnabled()) { |
| sb = new StringBuilder("Client cannot authenticate via:[TOKEN]"); |
| } else { |
| sb = |
| new StringBuilder( |
| "SIMPLE authentication is not enabled. Available:[TOKEN]"); |
| } |
| String errorMsg = testStartContainer(rpc, validAppAttemptId, validNode, |
| validContainerToken, null, true); |
| Assert.assertTrue("In calling " + validNode + " exception was '" |
| + errorMsg + "' but doesn't contain '" |
| + sb.toString() + "'", errorMsg.contains(sb.toString())); |
| |
| org.apache.hadoop.yarn.api.records.Token invalidNMToken = |
| tempManager.createNMToken(validAppAttemptId, validNode, user); |
| sb = new StringBuilder("Given NMToken for application : "); |
| sb.append(validAppAttemptId.toString()) |
| .append(" seems to have been generated illegally."); |
| Assert.assertTrue(sb.toString().contains( |
| testStartContainer(rpc, validAppAttemptId, validNode, |
| validContainerToken, invalidNMToken, true))); |
| |
| // valid NMToken but belonging to other node |
| invalidNMToken = |
| nmTokenSecretManagerRM.createNMToken(validAppAttemptId, invalidNode, |
| user); |
| sb = new StringBuilder("Given NMToken for application : "); |
| sb.append(validAppAttemptId) |
| .append(" is not valid for current node manager.expected : ") |
| .append(validNode.toString()) |
| .append(" found : ").append(invalidNode.toString()); |
| Assert.assertTrue(sb.toString().contains( |
| testStartContainer(rpc, validAppAttemptId, validNode, |
| validContainerToken, invalidNMToken, true))); |
| |
| // using correct tokens. nmtoken for app attempt should get saved. |
| testConf.setInt(YarnConfiguration.RM_CONTAINER_ALLOC_EXPIRY_INTERVAL_MS, |
| 4 * 60 * 1000); |
| validContainerToken = |
| containerTokenSecretManager.createContainerToken(validContainerId, |
| 0, validNode, user, r, Priority.newInstance(0), 0); |
| Assert.assertTrue(testStartContainer(rpc, validAppAttemptId, validNode, |
| validContainerToken, validNMToken, false).isEmpty()); |
| Assert.assertTrue(nmTokenSecretManagerNM |
| .isAppAttemptNMTokenKeyPresent(validAppAttemptId)); |
| |
| // using a new compatible version nmtoken, expect container can be started |
| // successfully. |
| ApplicationAttemptId validAppAttemptId2 = |
| ApplicationAttemptId.newInstance(appId, 2); |
| |
| ContainerId validContainerId2 = |
| ContainerId.newContainerId(validAppAttemptId2, 0); |
| |
| org.apache.hadoop.yarn.api.records.Token validContainerToken2 = |
| containerTokenSecretManager.createContainerToken(validContainerId2, |
| 0, validNode, user, r, Priority.newInstance(0), 0); |
| |
| org.apache.hadoop.yarn.api.records.Token validNMToken2 = |
| nmTokenSecretManagerRM.createNMToken(validAppAttemptId2, validNode, user); |
| // First, get a new NMTokenIdentifier. |
| NMTokenIdentifier newIdentifier = new NMTokenIdentifier(); |
| byte[] tokenIdentifierContent = validNMToken2.getIdentifier().array(); |
| DataInputBuffer dib = new DataInputBuffer(); |
| dib.reset(tokenIdentifierContent, tokenIdentifierContent.length); |
| newIdentifier.readFields(dib); |
| |
| // Then, generate a new version NMTokenIdentifier (NMTokenIdentifierNewForTest) |
| // with additional field of message. |
| NMTokenIdentifierNewForTest newVersionIdentifier = |
| new NMTokenIdentifierNewForTest(newIdentifier, "message"); |
| |
| // check new version NMTokenIdentifier has correct info. |
| Assert.assertEquals("The ApplicationAttemptId is changed after set to " + |
| "newVersionIdentifier", validAppAttemptId2.getAttemptId(), |
| newVersionIdentifier.getApplicationAttemptId().getAttemptId() |
| ); |
| |
| Assert.assertEquals("The message is changed after set to newVersionIdentifier", |
| "message", newVersionIdentifier.getMessage()); |
| |
| Assert.assertEquals("The NodeId is changed after set to newVersionIdentifier", |
| validNode, newVersionIdentifier.getNodeId()); |
| |
| // create new Token based on new version NMTokenIdentifier. |
| org.apache.hadoop.yarn.api.records.Token newVersionedNMToken = |
| BaseNMTokenSecretManager.newInstance( |
| nmTokenSecretManagerRM.retrievePassword(newVersionIdentifier), |
| newVersionIdentifier); |
| |
| // Verify startContainer is successful and no exception is thrown. |
| Assert.assertTrue(testStartContainer(rpc, validAppAttemptId2, validNode, |
| validContainerToken2, newVersionedNMToken, false).isEmpty()); |
| Assert.assertTrue(nmTokenSecretManagerNM |
| .isAppAttemptNMTokenKeyPresent(validAppAttemptId2)); |
| |
| //Now lets wait till container finishes and is removed from node manager. |
| waitForContainerToFinishOnNM(validContainerId); |
| sb = new StringBuilder("Attempt to relaunch the same container with id "); |
| sb.append(validContainerId); |
| Assert.assertTrue(testStartContainer(rpc, validAppAttemptId, validNode, |
| validContainerToken, validNMToken, true).contains(sb.toString())); |
| |
| // Container is removed from node manager's memory by this time. |
| // trying to stop the container. It should not throw any exception. |
| testStopContainer(rpc, validAppAttemptId, validNode, validContainerId, |
| validNMToken, false); |
| |
| // Rolling over master key twice so that we can check whether older keys |
| // are used for authentication. |
| rollNMTokenMasterKey(nmTokenSecretManagerRM, nmTokenSecretManagerNM); |
| // Key rolled over once.. rolling over again |
| rollNMTokenMasterKey(nmTokenSecretManagerRM, nmTokenSecretManagerNM); |
| |
| // trying get container status. Now saved nmToken should be used for |
| // authentication... It should complain saying container was recently |
| // stopped. |
| sb = new StringBuilder("Container "); |
| sb.append(validContainerId); |
| sb.append(" was recently stopped on node manager"); |
| Assert.assertTrue(testGetContainer(rpc, validAppAttemptId, validNode, |
| validContainerId, validNMToken, true).contains(sb.toString())); |
| |
| // Now lets remove the container from nm-memory |
| nm.getNodeStatusUpdater().clearFinishedContainersFromCache(); |
| |
| // This should fail as container is removed from recently tracked finished |
| // containers. |
| sb = new StringBuilder("Container "); |
| sb.append(validContainerId.toString()); |
| sb.append(" is not handled by this NodeManager"); |
| Assert.assertTrue(testGetContainer(rpc, validAppAttemptId, validNode, |
| validContainerId, validNMToken, false).contains(sb.toString())); |
| |
| // using appAttempt-1 NMtoken for launching container for appAttempt-2 |
| // should succeed. |
| ApplicationAttemptId attempt2 = ApplicationAttemptId.newInstance(appId, 2); |
| Token attempt1NMToken = |
| nmTokenSecretManagerRM |
| .createNMToken(validAppAttemptId, validNode, user); |
| org.apache.hadoop.yarn.api.records.Token newContainerToken = |
| containerTokenSecretManager.createContainerToken( |
| ContainerId.newContainerId(attempt2, 1), 0, validNode, user, r, |
| Priority.newInstance(0), 0); |
| Assert.assertTrue(testStartContainer(rpc, attempt2, validNode, |
| newContainerToken, attempt1NMToken, false).isEmpty()); |
| } |
| |
| private void waitForContainerToFinishOnNM(ContainerId containerId) { |
| Context nmContext = yarnCluster.getNodeManager(0).getNMContext(); |
| int interval = 4 * 60; // Max time for container token to expire. |
| |
| Assert.assertNotNull(nmContext.getContainers().containsKey(containerId)); |
| |
| // Get the container first, as it may be removed from the Context |
| // by asynchronous calls. |
| // This was leading to a flakey test as otherwise the container could |
| // be removed and end up null. |
| Container waitContainer = nmContext.getContainers().get(containerId); |
| |
| while ((interval-- > 0) |
| && !waitContainer.cloneAndGetContainerStatus() |
| .getState().equals(ContainerState.COMPLETE)) { |
| try { |
| LOG.info("Waiting for " + containerId + " to complete."); |
| Thread.sleep(1000); |
| } catch (InterruptedException e) { |
| } |
| } |
| // Normally, Containers will be removed from NM context after they are |
| // explicitly acked by RM. Now, manually remove it for testing. |
| yarnCluster.getNodeManager(0).getNodeStatusUpdater() |
| .addCompletedContainer(containerId); |
| LOG.info("Removing container from NMContext, containerID = " + containerId); |
| nmContext.getContainers().remove(containerId); |
| } |
| |
| protected void waitForNMToReceiveNMTokenKey( |
| NMTokenSecretManagerInNM nmTokenSecretManagerNM) |
| throws InterruptedException { |
| int attempt = 60; |
| while (nmTokenSecretManagerNM.getNodeId() == null && attempt-- > 0) { |
| Thread.sleep(2000); |
| } |
| } |
| |
| protected void rollNMTokenMasterKey( |
| NMTokenSecretManagerInRM nmTokenSecretManagerRM, |
| NMTokenSecretManagerInNM nmTokenSecretManagerNM) throws Exception { |
| int oldKeyId = nmTokenSecretManagerRM.getCurrentKey().getKeyId(); |
| nmTokenSecretManagerRM.rollMasterKey(); |
| int interval = 40; |
| while (nmTokenSecretManagerNM.getCurrentKey().getKeyId() == oldKeyId |
| && interval-- > 0) { |
| Thread.sleep(1000); |
| } |
| nmTokenSecretManagerRM.activateNextMasterKey(); |
| Assert.assertTrue((nmTokenSecretManagerNM.getCurrentKey().getKeyId() |
| == nmTokenSecretManagerRM.getCurrentKey().getKeyId())); |
| } |
| |
| private String testStopContainer(YarnRPC rpc, |
| ApplicationAttemptId appAttemptId, NodeId nodeId, |
| ContainerId containerId, Token nmToken, boolean isExceptionExpected) { |
| try { |
| stopContainer(rpc, nmToken, |
| Arrays.asList(new ContainerId[] {containerId}), appAttemptId, |
| nodeId); |
| if (isExceptionExpected) { |
| fail("Exception was expected!!"); |
| } |
| return ""; |
| } catch (Exception e) { |
| e.printStackTrace(); |
| return e.getMessage(); |
| } |
| } |
| |
| private String testGetContainer(YarnRPC rpc, |
| ApplicationAttemptId appAttemptId, NodeId nodeId, |
| ContainerId containerId, |
| org.apache.hadoop.yarn.api.records.Token nmToken, |
| boolean isExceptionExpected) { |
| try { |
| getContainerStatus(rpc, nmToken, containerId, appAttemptId, nodeId, |
| isExceptionExpected); |
| if (isExceptionExpected) { |
| fail("Exception was expected!!"); |
| } |
| return ""; |
| } catch (Exception e) { |
| e.printStackTrace(); |
| return e.getMessage(); |
| } |
| } |
| |
| private String testStartContainer(YarnRPC rpc, |
| ApplicationAttemptId appAttemptId, NodeId nodeId, |
| org.apache.hadoop.yarn.api.records.Token containerToken, |
| org.apache.hadoop.yarn.api.records.Token nmToken, |
| boolean isExceptionExpected) { |
| try { |
| startContainer(rpc, nmToken, containerToken, nodeId, |
| appAttemptId.toString()); |
| if (isExceptionExpected){ |
| fail("Exception was expected!!"); |
| } |
| return ""; |
| } catch (Exception e) { |
| e.printStackTrace(); |
| return e.getMessage(); |
| } |
| } |
| |
| private void stopContainer(YarnRPC rpc, Token nmToken, |
| List<ContainerId> containerId, ApplicationAttemptId appAttemptId, |
| NodeId nodeId) throws Exception { |
| StopContainersRequest request = |
| StopContainersRequest.newInstance(containerId); |
| ContainerManagementProtocol proxy = null; |
| try { |
| proxy = |
| getContainerManagementProtocolProxy(rpc, nmToken, nodeId, |
| appAttemptId.toString()); |
| StopContainersResponse response = proxy.stopContainers(request); |
| if (response.getFailedRequests() != null && |
| response.getFailedRequests().containsKey(containerId)) { |
| parseAndThrowException(response.getFailedRequests().get(containerId) |
| .deSerialize()); |
| } |
| } catch (Exception e) { |
| if (proxy != null) { |
| rpc.stopProxy(proxy, conf); |
| } |
| } |
| } |
| |
| private void |
| getContainerStatus(YarnRPC rpc, |
| org.apache.hadoop.yarn.api.records.Token nmToken, |
| ContainerId containerId, |
| ApplicationAttemptId appAttemptId, NodeId nodeId, |
| boolean isExceptionExpected) throws Exception { |
| List<ContainerId> containerIds = new ArrayList<ContainerId>(); |
| containerIds.add(containerId); |
| GetContainerStatusesRequest request = |
| GetContainerStatusesRequest.newInstance(containerIds); |
| ContainerManagementProtocol proxy = null; |
| try { |
| proxy = |
| getContainerManagementProtocolProxy(rpc, nmToken, nodeId, |
| appAttemptId.toString()); |
| GetContainerStatusesResponse statuses |
| = proxy.getContainerStatuses(request); |
| if (statuses.getFailedRequests() != null |
| && statuses.getFailedRequests().containsKey(containerId)) { |
| parseAndThrowException(statuses.getFailedRequests().get(containerId) |
| .deSerialize()); |
| } |
| } finally { |
| if (proxy != null) { |
| rpc.stopProxy(proxy, conf); |
| } |
| } |
| } |
| |
| private void startContainer(final YarnRPC rpc, |
| org.apache.hadoop.yarn.api.records.Token nmToken, |
| org.apache.hadoop.yarn.api.records.Token containerToken, |
| NodeId nodeId, String user) throws Exception { |
| |
| ContainerLaunchContext context = |
| Records.newRecord(ContainerLaunchContext.class); |
| StartContainerRequest scRequest = |
| StartContainerRequest.newInstance(context, containerToken); |
| List<StartContainerRequest> list = new ArrayList<StartContainerRequest>(); |
| list.add(scRequest); |
| StartContainersRequest allRequests = |
| StartContainersRequest.newInstance(list); |
| ContainerManagementProtocol proxy = null; |
| try { |
| proxy = getContainerManagementProtocolProxy(rpc, nmToken, nodeId, user); |
| StartContainersResponse response = proxy.startContainers(allRequests); |
| for(SerializedException ex : response.getFailedRequests().values()){ |
| parseAndThrowException(ex.deSerialize()); |
| } |
| } finally { |
| if (proxy != null) { |
| rpc.stopProxy(proxy, conf); |
| } |
| } |
| } |
| |
| private void parseAndThrowException(Throwable t) throws YarnException, |
| IOException { |
| if (t instanceof YarnException) { |
| throw (YarnException) t; |
| } else if (t instanceof InvalidToken) { |
| throw (InvalidToken) t; |
| } else { |
| throw (IOException) t; |
| } |
| } |
| |
| protected ContainerManagementProtocol getContainerManagementProtocolProxy( |
| final YarnRPC rpc, org.apache.hadoop.yarn.api.records.Token nmToken, |
| NodeId nodeId, String user) { |
| ContainerManagementProtocol proxy; |
| UserGroupInformation ugi = UserGroupInformation.createRemoteUser(user); |
| final InetSocketAddress addr = |
| new InetSocketAddress(nodeId.getHost(), nodeId.getPort()); |
| if (nmToken != null) { |
| ugi.addToken(ConverterUtils.convertFromYarn(nmToken, addr)); |
| } |
| proxy = |
| NMProxy.createNMProxy(conf, ContainerManagementProtocol.class, ugi, |
| rpc, addr); |
| return proxy; |
| } |
| |
| /** |
| * This tests a malice user getting a proper token but then messing with it by |
| * tampering with containerID/Resource etc.. His/her containers should be |
| * rejected. |
| * |
| * @throws IOException |
| * @throws InterruptedException |
| * @throws YarnException |
| */ |
| private void testContainerToken(Configuration conf) throws IOException, |
| InterruptedException, YarnException { |
| |
| LOG.info("Running test for malice user"); |
| /* |
| * We need to check for containerToken (authorization). |
| * Here we will be assuming that we have valid NMToken |
| * 1) ContainerToken used is expired. |
| * 2) ContainerToken is tampered (resource is modified). |
| */ |
| NMTokenSecretManagerInRM nmTokenSecretManagerInRM = |
| yarnCluster.getResourceManager().getRMContext() |
| .getNMTokenSecretManager(); |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(appId, 0); |
| ContainerId cId = ContainerId.newContainerId(appAttemptId, 0); |
| NodeManager nm = yarnCluster.getNodeManager(0); |
| NMTokenSecretManagerInNM nmTokenSecretManagerInNM = |
| nm.getNMContext().getNMTokenSecretManager(); |
| String user = "test"; |
| |
| waitForNMToReceiveNMTokenKey(nmTokenSecretManagerInNM); |
| |
| NodeId nodeId = nm.getNMContext().getNodeId(); |
| |
| // Both id should be equal. |
| Assert.assertEquals(nmTokenSecretManagerInNM.getCurrentKey().getKeyId(), |
| nmTokenSecretManagerInRM.getCurrentKey().getKeyId()); |
| |
| |
| RMContainerTokenSecretManager containerTokenSecretManager = |
| yarnCluster.getResourceManager().getRMContext(). |
| getContainerTokenSecretManager(); |
| |
| Resource r = Resource.newInstance(1230, 2); |
| |
| Token containerToken = |
| containerTokenSecretManager.createContainerToken( |
| cId, 0, nodeId, user, r, Priority.newInstance(0), 0); |
| |
| ContainerTokenIdentifier containerTokenIdentifier = |
| getContainerTokenIdentifierFromToken(containerToken); |
| |
| // Verify new compatible version ContainerTokenIdentifier |
| // can work successfully. |
| ContainerTokenIdentifierForTest newVersionTokenIdentifier = |
| new ContainerTokenIdentifierForTest(containerTokenIdentifier, |
| "message"); |
| byte[] password = |
| containerTokenSecretManager.createPassword(newVersionTokenIdentifier); |
| |
| Token newContainerToken = BuilderUtils.newContainerToken( |
| nodeId, password, newVersionTokenIdentifier); |
| |
| Token nmToken = |
| nmTokenSecretManagerInRM.createNMToken(appAttemptId, nodeId, user); |
| YarnRPC rpc = YarnRPC.create(conf); |
| Assert.assertTrue(testStartContainer(rpc, appAttemptId, nodeId, |
| newContainerToken, nmToken, false).isEmpty()); |
| |
| // Creating a tampered Container Token |
| RMContainerTokenSecretManager tamperedContainerTokenSecretManager = |
| new RMContainerTokenSecretManager(conf); |
| tamperedContainerTokenSecretManager.rollMasterKey(); |
| do { |
| tamperedContainerTokenSecretManager.rollMasterKey(); |
| tamperedContainerTokenSecretManager.activateNextMasterKey(); |
| } while (containerTokenSecretManager.getCurrentKey().getKeyId() |
| == tamperedContainerTokenSecretManager.getCurrentKey().getKeyId()); |
| |
| ContainerId cId2 = ContainerId.newContainerId(appAttemptId, 1); |
| // Creating modified containerToken |
| Token containerToken2 = |
| tamperedContainerTokenSecretManager.createContainerToken(cId2, 0, |
| nodeId, user, r, Priority.newInstance(0), 0); |
| |
| StringBuilder sb = new StringBuilder("Given Container "); |
| sb.append(cId2); |
| sb.append(" seems to have an illegally generated token."); |
| Assert.assertTrue(testStartContainer(rpc, appAttemptId, nodeId, |
| containerToken2, nmToken, true).contains(sb.toString())); |
| } |
| |
| private ContainerTokenIdentifier getContainerTokenIdentifierFromToken( |
| Token containerToken) throws IOException { |
| ContainerTokenIdentifier containerTokenIdentifier; |
| containerTokenIdentifier = new ContainerTokenIdentifier(); |
| byte[] tokenIdentifierContent = containerToken.getIdentifier().array(); |
| DataInputBuffer dib = new DataInputBuffer(); |
| dib.reset(tokenIdentifierContent, tokenIdentifierContent.length); |
| containerTokenIdentifier.readFields(dib); |
| return containerTokenIdentifier; |
| } |
| |
| /** |
| * This tests whether a containerId is serialized/deserialized with epoch. |
| * |
| * @throws IOException |
| * @throws InterruptedException |
| * @throws YarnException |
| */ |
| private void testContainerTokenWithEpoch(Configuration conf) |
| throws IOException, InterruptedException, YarnException { |
| |
| LOG.info("Running test for serializing/deserializing containerIds"); |
| |
| NMTokenSecretManagerInRM nmTokenSecretManagerInRM = |
| yarnCluster.getResourceManager().getRMContext() |
| .getNMTokenSecretManager(); |
| ApplicationId appId = ApplicationId.newInstance(1, 1); |
| ApplicationAttemptId appAttemptId = |
| ApplicationAttemptId.newInstance(appId, 0); |
| ContainerId cId = ContainerId.newContainerId(appAttemptId, (5L << 40) | 3L); |
| NodeManager nm = yarnCluster.getNodeManager(0); |
| NMTokenSecretManagerInNM nmTokenSecretManagerInNM = |
| nm.getNMContext().getNMTokenSecretManager(); |
| String user = "test"; |
| |
| waitForNMToReceiveNMTokenKey(nmTokenSecretManagerInNM); |
| |
| NodeId nodeId = nm.getNMContext().getNodeId(); |
| |
| // Both id should be equal. |
| Assert.assertEquals(nmTokenSecretManagerInNM.getCurrentKey().getKeyId(), |
| nmTokenSecretManagerInRM.getCurrentKey().getKeyId()); |
| |
| // Creating a normal Container Token |
| RMContainerTokenSecretManager containerTokenSecretManager = |
| yarnCluster.getResourceManager().getRMContext(). |
| getContainerTokenSecretManager(); |
| Resource r = Resource.newInstance(1230, 2); |
| Token containerToken = |
| containerTokenSecretManager.createContainerToken(cId, 0, nodeId, user, |
| r, Priority.newInstance(0), 0); |
| |
| ContainerTokenIdentifier containerTokenIdentifier = |
| new ContainerTokenIdentifier(); |
| byte[] tokenIdentifierContent = containerToken.getIdentifier().array(); |
| DataInputBuffer dib = new DataInputBuffer(); |
| dib.reset(tokenIdentifierContent, tokenIdentifierContent.length); |
| containerTokenIdentifier.readFields(dib); |
| |
| |
| Assert.assertEquals(cId, containerTokenIdentifier.getContainerID()); |
| Assert.assertEquals( |
| cId.toString(), containerTokenIdentifier.getContainerID().toString()); |
| |
| Token nmToken = |
| nmTokenSecretManagerInRM.createNMToken(appAttemptId, nodeId, user); |
| |
| YarnRPC rpc = YarnRPC.create(conf); |
| testStartContainer(rpc, appAttemptId, nodeId, containerToken, nmToken, |
| false); |
| |
| List<ContainerId> containerIds = new LinkedList<ContainerId>(); |
| containerIds.add(cId); |
| ContainerManagementProtocol proxy |
| = getContainerManagementProtocolProxy(rpc, nmToken, nodeId, user); |
| GetContainerStatusesResponse res = proxy.getContainerStatuses( |
| GetContainerStatusesRequest.newInstance(containerIds)); |
| Assert.assertNotNull(res.getContainerStatuses().get(0)); |
| Assert.assertEquals( |
| cId, res.getContainerStatuses().get(0).getContainerId()); |
| Assert.assertEquals(cId.toString(), |
| res.getContainerStatuses().get(0).getContainerId().toString()); |
| } |
| } |
