MAPREDUCE-5475. MRClientService does not verify ACLs properly. Contributed by Jason Lowe

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/branch-0.23@1520161 13f79535-47bb-0310-9956-ffa450edef68

hadoop-mapreduce-project/CHANGES.txt

@@ -38,6 +38,8 @@ Release 0.23.10 - UNRELEASED
     MAPREDUCE-5468. Fix MR AM recovery for map-only jobs. (vinodkv via
     acmurthy)
 
+    MAPREDUCE-5475. MRClientService does not verify ACLs properly (jlowe)
+
 Release 0.23.9 - 2013-07-08
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/main/java/org/apache/hadoop/mapreduce/v2/app/client/MRClientService.java

@@ -18,6 +18,7 @@
 
 package org.apache.hadoop.mapreduce.v2.app.client;
 
+import java.io.IOException;
 import java.net.InetSocketAddress;
 import java.util.Arrays;
 import java.util.Collection;
@@ -28,6 +29,7 @@ import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
 import org.apache.hadoop.ipc.Server;
+import org.apache.hadoop.mapreduce.JobACL;
 import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.TypeConverter;
 import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
@@ -188,15 +190,26 @@ public class MRClientService extends AbstractService
     }
     
     private Job verifyAndGetJob(JobId jobID, 
-        boolean modifyAccess) throws YarnRemoteException {
+        JobACL accessType) throws YarnRemoteException {
       Job job = appContext.getJob(jobID);
+      UserGroupInformation ugi;
+      try {
+        ugi = UserGroupInformation.getCurrentUser();
+      } catch (IOException e) {
+        throw RPCUtil.getRemoteException(e);
+      }
+      if (!job.checkAccess(ugi, accessType)) {
+        throw RPCUtil.getRemoteException("User " + ugi.getShortUserName()
+            + " cannot perform operation " + accessType.name() + " on "
+            + jobID);
+      }
       return job;
     }
  
     private Task verifyAndGetTask(TaskId taskID, 
-        boolean modifyAccess) throws YarnRemoteException {
+        JobACL accessType) throws YarnRemoteException {
       Task task = verifyAndGetJob(taskID.getJobId(), 
-          modifyAccess).getTask(taskID);
+          accessType).getTask(taskID);
       if (task == null) {
         throw RPCUtil.getRemoteException("Unknown Task " + taskID);
       }
@@ -204,9 +217,9 @@ public class MRClientService extends AbstractService
     }
 
     private TaskAttempt verifyAndGetAttempt(TaskAttemptId attemptID, 
-        boolean modifyAccess) throws YarnRemoteException {
+        JobACL accessType) throws YarnRemoteException {
       TaskAttempt attempt = verifyAndGetTask(attemptID.getTaskId(), 
-          modifyAccess).getAttempt(attemptID);
+          accessType).getAttempt(attemptID);
       if (attempt == null) {
         throw RPCUtil.getRemoteException("Unknown TaskAttempt " + attemptID);
       }
@@ -217,7 +230,7 @@ public class MRClientService extends AbstractService
     public GetCountersResponse getCounters(GetCountersRequest request) 
       throws YarnRemoteException {
       JobId jobId = request.getJobId();
-      Job job = verifyAndGetJob(jobId, false);
+      Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
       GetCountersResponse response =
         recordFactory.newRecordInstance(GetCountersResponse.class);
       response.setCounters(TypeConverter.toYarn(job.getAllCounters()));
@@ -228,7 +241,7 @@ public class MRClientService extends AbstractService
     public GetJobReportResponse getJobReport(GetJobReportRequest request) 
       throws YarnRemoteException {
       JobId jobId = request.getJobId();
-      Job job = verifyAndGetJob(jobId, false);
+      Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
       GetJobReportResponse response = 
         recordFactory.newRecordInstance(GetJobReportResponse.class);
       if (job != null) {
@@ -247,7 +260,7 @@ public class MRClientService extends AbstractService
       GetTaskAttemptReportResponse response =
         recordFactory.newRecordInstance(GetTaskAttemptReportResponse.class);
       response.setTaskAttemptReport(
-          verifyAndGetAttempt(taskAttemptId, false).getReport());
+          verifyAndGetAttempt(taskAttemptId, JobACL.VIEW_JOB).getReport());
       return response;
     }
 
@@ -257,7 +270,8 @@ public class MRClientService extends AbstractService
       TaskId taskId = request.getTaskId();
       GetTaskReportResponse response = 
         recordFactory.newRecordInstance(GetTaskReportResponse.class);
-      response.setTaskReport(verifyAndGetTask(taskId, false).getReport());
+      response.setTaskReport(
+          verifyAndGetTask(taskId, JobACL.VIEW_JOB).getReport());
       return response;
     }
 
@@ -268,7 +282,7 @@ public class MRClientService extends AbstractService
       JobId jobId = request.getJobId();
       int fromEventId = request.getFromEventId();
       int maxEvents = request.getMaxEvents();
-      Job job = verifyAndGetJob(jobId, false);
+      Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
       
       GetTaskAttemptCompletionEventsResponse response = 
         recordFactory.newRecordInstance(GetTaskAttemptCompletionEventsResponse.class);
@@ -282,9 +296,16 @@ public class MRClientService extends AbstractService
     public KillJobResponse killJob(KillJobRequest request) 
       throws YarnRemoteException {
       JobId jobId = request.getJobId();
-      String message = "Kill Job received from client " + jobId;
+      UserGroupInformation callerUGI;
+      try {
+        callerUGI = UserGroupInformation.getCurrentUser();
+      } catch (IOException e) {
+        throw RPCUtil.getRemoteException(e);
+      }
+      String message = "Kill job " + jobId + " received from " + callerUGI
+          + " at " + Server.getRemoteAddress();
       LOG.info(message);
-  	  verifyAndGetJob(jobId, true);
+      verifyAndGetJob(jobId, JobACL.MODIFY_JOB);
       appContext.getEventHandler().handle(
           new JobDiagnosticsUpdateEvent(jobId, message));
       appContext.getEventHandler().handle(
@@ -299,9 +320,16 @@ public class MRClientService extends AbstractService
     public KillTaskResponse killTask(KillTaskRequest request) 
       throws YarnRemoteException {
       TaskId taskId = request.getTaskId();
-      String message = "Kill task received from client " + taskId;
+      UserGroupInformation callerUGI;
+      try {
+        callerUGI = UserGroupInformation.getCurrentUser();
+      } catch (IOException e) {
+        throw RPCUtil.getRemoteException(e);
+      }
+      String message = "Kill task " + taskId + " received from " + callerUGI
+          + " at " + Server.getRemoteAddress();
       LOG.info(message);
-      verifyAndGetTask(taskId, true);
+      verifyAndGetTask(taskId, JobACL.MODIFY_JOB);
       appContext.getEventHandler().handle(
           new TaskEvent(taskId, TaskEventType.T_KILL));
       KillTaskResponse response = 
@@ -314,9 +342,17 @@ public class MRClientService extends AbstractService
     public KillTaskAttemptResponse killTaskAttempt(
         KillTaskAttemptRequest request) throws YarnRemoteException {
       TaskAttemptId taskAttemptId = request.getTaskAttemptId();
-      String message = "Kill task attempt received from client " + taskAttemptId;
+      UserGroupInformation callerUGI;
+      try {
+        callerUGI = UserGroupInformation.getCurrentUser();
+      } catch (IOException e) {
+        throw RPCUtil.getRemoteException(e);
+      }
+      String message = "Kill task attempt " + taskAttemptId
+          + " received from " + callerUGI + " at "
+          + Server.getRemoteAddress();
       LOG.info(message);
-      verifyAndGetAttempt(taskAttemptId, true);
+      verifyAndGetAttempt(taskAttemptId, JobACL.MODIFY_JOB);
       appContext.getEventHandler().handle(
           new TaskAttemptDiagnosticsUpdateEvent(taskAttemptId, message));
       appContext.getEventHandler().handle(
@@ -334,8 +370,8 @@ public class MRClientService extends AbstractService
       
       GetDiagnosticsResponse response = 
         recordFactory.newRecordInstance(GetDiagnosticsResponse.class);
-      response.addAllDiagnostics(
-          verifyAndGetAttempt(taskAttemptId, false).getDiagnostics());
+      response.addAllDiagnostics(verifyAndGetAttempt(taskAttemptId,
+          JobACL.VIEW_JOB).getDiagnostics());
       return response;
     }
 
@@ -344,9 +380,17 @@ public class MRClientService extends AbstractService
     public FailTaskAttemptResponse failTaskAttempt(
         FailTaskAttemptRequest request) throws YarnRemoteException {
       TaskAttemptId taskAttemptId = request.getTaskAttemptId();
-      String message = "Fail task attempt received from client " + taskAttemptId;
+      UserGroupInformation callerUGI;
+      try {
+        callerUGI = UserGroupInformation.getCurrentUser();
+      } catch (IOException e) {
+        throw RPCUtil.getRemoteException(e);
+      }
+      String message = "Fail task attempt " + taskAttemptId
+          + " received from " + callerUGI + " at "
+          + Server.getRemoteAddress();
       LOG.info(message);
-      verifyAndGetAttempt(taskAttemptId, true);
+      verifyAndGetAttempt(taskAttemptId, JobACL.MODIFY_JOB);
       appContext.getEventHandler().handle(
           new TaskAttemptDiagnosticsUpdateEvent(taskAttemptId, message));
       appContext.getEventHandler().handle(
@@ -368,7 +412,7 @@ public class MRClientService extends AbstractService
       GetTaskReportsResponse response = 
         recordFactory.newRecordInstance(GetTaskReportsResponse.class);
       
-      Job job = verifyAndGetJob(jobId, false);
+      Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
       Collection<Task> tasks = job.getTasks(taskType).values();
       LOG.info("Getting task report for " + taskType + "   " + jobId
           + ". Report-size will be " + tasks.size());

hadoop-mapreduce-project/hadoop-mapreduce-client/hadoop-mapreduce-client-app/src/test/java/org/apache/hadoop/mapreduce/v2/app/TestMRClientService.java

@@ -18,13 +18,20 @@
 
 package org.apache.hadoop.mapreduce.v2.app;
 
+import static org.junit.Assert.fail;
+
+import java.security.PrivilegedExceptionAction;
 import java.util.Iterator;
 import java.util.List;
 
 import junit.framework.Assert;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.mapreduce.JobACL;
+import org.apache.hadoop.mapreduce.MRConfig;
+import org.apache.hadoop.mapreduce.MRJobConfig;
 import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
+import org.apache.hadoop.mapreduce.v2.api.protocolrecords.FailTaskAttemptRequest;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetCountersRequest;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetDiagnosticsRequest;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetJobReportRequest;
@@ -32,6 +39,9 @@ import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskAttemptCompleti
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskAttemptReportRequest;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskReportRequest;
 import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskReportsRequest;
+import org.apache.hadoop.mapreduce.v2.api.protocolrecords.KillJobRequest;
+import org.apache.hadoop.mapreduce.v2.api.protocolrecords.KillTaskAttemptRequest;
+import org.apache.hadoop.mapreduce.v2.api.protocolrecords.KillTaskRequest;
 import org.apache.hadoop.mapreduce.v2.api.records.AMInfo;
 import org.apache.hadoop.mapreduce.v2.api.records.JobReport;
 import org.apache.hadoop.mapreduce.v2.api.records.JobState;
@@ -51,6 +61,9 @@ import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent;
 import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent.TaskAttemptStatus;
+import org.apache.hadoop.security.AccessControlException;
+import org.apache.hadoop.security.UserGroupInformation;
+import org.apache.hadoop.yarn.exceptions.YarnRemoteException;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
 import org.apache.hadoop.yarn.ipc.YarnRPC;
@@ -170,6 +183,79 @@ public class TestMRClientService {
     app.waitForState(job, JobState.SUCCEEDED);
   }
 
+  @Test
+  public void testViewAclOnlyCannotModify() throws Exception {
+    final MRAppWithClientService app = new MRAppWithClientService(1, 0, false);
+    final Configuration conf = new Configuration();
+    conf.setBoolean(MRConfig.MR_ACLS_ENABLED, true);
+    conf.set(MRJobConfig.JOB_ACL_VIEW_JOB, "viewonlyuser");
+    Job job = app.submit(conf);
+    app.waitForState(job, JobState.RUNNING);
+    Assert.assertEquals("Num tasks not correct", 1, job.getTasks().size());
+    Iterator<Task> it = job.getTasks().values().iterator();
+    Task task = it.next();
+    app.waitForState(task, TaskState.RUNNING);
+    TaskAttempt attempt = task.getAttempts().values().iterator().next();
+    app.waitForState(attempt, TaskAttemptState.RUNNING);
+
+    UserGroupInformation viewOnlyUser =
+        UserGroupInformation.createUserForTesting(
+            "viewonlyuser", new String[] {});
+    Assert.assertTrue("viewonlyuser cannot view job",
+        job.checkAccess(viewOnlyUser, JobACL.VIEW_JOB));
+    Assert.assertFalse("viewonlyuser can modify job",
+        job.checkAccess(viewOnlyUser, JobACL.MODIFY_JOB));
+    MRClientProtocol client = viewOnlyUser.doAs(
+        new PrivilegedExceptionAction<MRClientProtocol>() {
+          @Override
+          public MRClientProtocol run() throws Exception {
+            YarnRPC rpc = YarnRPC.create(conf);
+            return (MRClientProtocol) rpc.getProxy(MRClientProtocol.class,
+                app.clientService.getBindAddress(), conf);
+          }
+        });
+
+    KillJobRequest killJobRequest = recordFactory.newRecordInstance(
+        KillJobRequest.class);
+    killJobRequest.setJobId(app.getJobId());
+    try {
+      client.killJob(killJobRequest);
+      fail("viewonlyuser killed job");
+    } catch (YarnRemoteException e) {
+      Assert.assertTrue(e.getMessage().contains("User viewonlyuser cannot"));
+    }
+
+    KillTaskRequest killTaskRequest = recordFactory.newRecordInstance(
+        KillTaskRequest.class);
+    killTaskRequest.setTaskId(task.getID());
+    try {
+      client.killTask(killTaskRequest);
+      fail("viewonlyuser killed task");
+    } catch (YarnRemoteException e) {
+      Assert.assertTrue(e.getMessage().contains("User viewonlyuser cannot"));
+    }
+
+    KillTaskAttemptRequest killTaskAttemptRequest =
+        recordFactory.newRecordInstance(KillTaskAttemptRequest.class);
+    killTaskAttemptRequest.setTaskAttemptId(attempt.getID());
+    try {
+      client.killTaskAttempt(killTaskAttemptRequest);
+      fail("viewonlyuser killed task attempt");
+    } catch (YarnRemoteException e) {
+      Assert.assertTrue(e.getMessage().contains("User viewonlyuser cannot"));
+    }
+
+    FailTaskAttemptRequest failTaskAttemptRequest =
+        recordFactory.newRecordInstance(FailTaskAttemptRequest.class);
+    failTaskAttemptRequest.setTaskAttemptId(attempt.getID());
+    try {
+      client.failTaskAttempt(failTaskAttemptRequest);
+      fail("viewonlyuser killed task attempt");
+    } catch (YarnRemoteException e) {
+      Assert.assertTrue(e.getMessage().contains("User viewonlyuser cannot"));
+    }
+  }
+
   private void verifyJobReport(JobReport jr) {
     Assert.assertNotNull("JobReport is null", jr);
     List<AMInfo> amInfos = jr.getAMInfos();