Skip to content

Commit efaa2c4

Browse files
acmurthyacmurthy
committed
MAPREDUCE-5475. Ensure MRClientService verifies ACLs for users. Contributed by Jason Lowe.
git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/trunk@1516361 13f79535-47bb-0310-9956-ffa450edef68
1 parent 045707c commit efaa2c4

File tree

3 files changed

+130
-22
lines changed

3 files changed

+130
-22
lines changed

‎hadoop-mapreduce-project/CHANGES.txt‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -234,6 +234,9 @@ Release 2.1.1-beta - UNRELEASED
234234
MAPREDUCE-5468. Fix MR AM recovery for map-only jobs. (vinodkv via
235235
acmurthy)
236236

237+
MAPREDUCE-5475. Ensure MRClientService verifies ACLs for users. (jlowe via
238+
acmurthy)
239+
237240
Release 2.1.0-beta - 2013-08-22
238241

239242
INCOMPATIBLE CHANGES

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

Lines changed: 42 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import org.apache.hadoop.conf.Configuration;
2929
import org.apache.hadoop.fs.CommonConfigurationKeysPublic;
3030
import org.apache.hadoop.ipc.Server;
31+
import org.apache.hadoop.mapreduce.JobACL;
3132
import org.apache.hadoop.mapreduce.MRJobConfig;
3233
import org.apache.hadoop.mapreduce.TypeConverter;
3334
import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
@@ -78,6 +79,8 @@
7879
import org.apache.hadoop.mapreduce.v2.app.security.authorize.MRAMPolicyProvider;
7980
import org.apache.hadoop.mapreduce.v2.app.webapp.AMWebApp;
8081
import org.apache.hadoop.net.NetUtils;
82+
import org.apache.hadoop.security.AccessControlException;
83+
import org.apache.hadoop.security.UserGroupInformation;
8184
import org.apache.hadoop.security.authorize.PolicyProvider;
8285
import org.apache.hadoop.service.AbstractService;
8386
import org.apache.hadoop.yarn.factories.RecordFactory;
@@ -175,26 +178,32 @@ public InetSocketAddress getConnectAddress() {
175178
return getBindAddress();
176179
}
177180

178-
private Job verifyAndGetJob(JobId jobID,
179-
boolean modifyAccess) throws IOException {
181+
private Job verifyAndGetJob(JobId jobID,
182+
JobACL accessType) throws IOException {
180183
Job job = appContext.getJob(jobID);
184+
UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
185+
if (!job.checkAccess(ugi, accessType)) {
186+
throw new AccessControlException("User " + ugi.getShortUserName()
187+
+ " cannot perform operation " + accessType.name() + " on "
188+
+ jobID);
189+
}
181190
return job;
182191
}
183192

184193
private Task verifyAndGetTask(TaskId taskID,
185-
boolean modifyAccess) throws IOException {
194+
JobACL accessType) throws IOException {
186195
Task task = verifyAndGetJob(taskID.getJobId(),
187-
modifyAccess).getTask(taskID);
196+
accessType).getTask(taskID);
188197
if (task == null) {
189198
throw new IOException("Unknown Task " + taskID);
190199
}
191200
return task;
192201
}
193202

194203
private TaskAttempt verifyAndGetAttempt(TaskAttemptId attemptID,
195-
boolean modifyAccess) throws IOException {
204+
JobACL accessType) throws IOException {
196205
TaskAttempt attempt = verifyAndGetTask(attemptID.getTaskId(),
197-
modifyAccess).getAttempt(attemptID);
206+
accessType).getAttempt(attemptID);
198207
if (attempt == null) {
199208
throw new IOException("Unknown TaskAttempt " + attemptID);
200209
}
@@ -205,7 +214,7 @@ private TaskAttempt verifyAndGetAttempt(TaskAttemptId attemptID,
205214
public GetCountersResponse getCounters(GetCountersRequest request)
206215
throws IOException {
207216
JobId jobId = request.getJobId();
208-
Job job = verifyAndGetJob(jobId, false);
217+
Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
209218
GetCountersResponse response =
210219
recordFactory.newRecordInstance(GetCountersResponse.class);
211220
response.setCounters(TypeConverter.toYarn(job.getAllCounters()));
@@ -216,7 +225,7 @@ public GetCountersResponse getCounters(GetCountersRequest request)
216225
public GetJobReportResponse getJobReport(GetJobReportRequest request)
217226
throws IOException {
218227
JobId jobId = request.getJobId();
219-
Job job = verifyAndGetJob(jobId, false);
228+
Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
220229
GetJobReportResponse response =
221230
recordFactory.newRecordInstance(GetJobReportResponse.class);
222231
if (job != null) {
@@ -235,7 +244,7 @@ public GetTaskAttemptReportResponse getTaskAttemptReport(
235244
GetTaskAttemptReportResponse response =
236245
recordFactory.newRecordInstance(GetTaskAttemptReportResponse.class);
237246
response.setTaskAttemptReport(
238-
verifyAndGetAttempt(taskAttemptId, false).getReport());
247+
verifyAndGetAttempt(taskAttemptId, JobACL.VIEW_JOB).getReport());
239248
return response;
240249
}
241250

@@ -245,7 +254,8 @@ public GetTaskReportResponse getTaskReport(GetTaskReportRequest request)
245254
TaskId taskId = request.getTaskId();
246255
GetTaskReportResponse response =
247256
recordFactory.newRecordInstance(GetTaskReportResponse.class);
248-
response.setTaskReport(verifyAndGetTask(taskId, false).getReport());
257+
response.setTaskReport(
258+
verifyAndGetTask(taskId, JobACL.VIEW_JOB).getReport());
249259
return response;
250260
}
251261

@@ -256,7 +266,7 @@ public GetTaskAttemptCompletionEventsResponse getTaskAttemptCompletionEvents(
256266
JobId jobId = request.getJobId();
257267
int fromEventId = request.getFromEventId();
258268
int maxEvents = request.getMaxEvents();
259-
Job job = verifyAndGetJob(jobId, false);
269+
Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
260270

261271
GetTaskAttemptCompletionEventsResponse response =
262272
recordFactory.newRecordInstance(GetTaskAttemptCompletionEventsResponse.class);
@@ -270,9 +280,11 @@ public GetTaskAttemptCompletionEventsResponse getTaskAttemptCompletionEvents(
270280
public KillJobResponse killJob(KillJobRequest request)
271281
throws IOException {
272282
JobId jobId = request.getJobId();
273-
String message = "Kill Job received from client " + jobId;
283+
UserGroupInformation callerUGI = UserGroupInformation.getCurrentUser();
284+
String message = "Kill job " + jobId + " received from " + callerUGI
285+
+ " at " + Server.getRemoteAddress();
274286
LOG.info(message);
275-
verifyAndGetJob(jobId, true);
287+
verifyAndGetJob(jobId, JobACL.MODIFY_JOB);
276288
appContext.getEventHandler().handle(
277289
new JobDiagnosticsUpdateEvent(jobId, message));
278290
appContext.getEventHandler().handle(
@@ -287,9 +299,11 @@ public KillJobResponse killJob(KillJobRequest request)
287299
public KillTaskResponse killTask(KillTaskRequest request)
288300
throws IOException {
289301
TaskId taskId = request.getTaskId();
290-
String message = "Kill task received from client " + taskId;
302+
UserGroupInformation callerUGI = UserGroupInformation.getCurrentUser();
303+
String message = "Kill task " + taskId + " received from " + callerUGI
304+
+ " at " + Server.getRemoteAddress();
291305
LOG.info(message);
292-
verifyAndGetTask(taskId, true);
306+
verifyAndGetTask(taskId, JobACL.MODIFY_JOB);
293307
appContext.getEventHandler().handle(
294308
new TaskEvent(taskId, TaskEventType.T_KILL));
295309
KillTaskResponse response =
@@ -302,9 +316,12 @@ public KillTaskResponse killTask(KillTaskRequest request)
302316
public KillTaskAttemptResponse killTaskAttempt(
303317
KillTaskAttemptRequest request) throws IOException {
304318
TaskAttemptId taskAttemptId = request.getTaskAttemptId();
305-
String message = "Kill task attempt received from client " + taskAttemptId;
319+
UserGroupInformation callerUGI = UserGroupInformation.getCurrentUser();
320+
String message = "Kill task attempt " + taskAttemptId
321+
+ " received from " + callerUGI + " at "
322+
+ Server.getRemoteAddress();
306323
LOG.info(message);
307-
verifyAndGetAttempt(taskAttemptId, true);
324+
verifyAndGetAttempt(taskAttemptId, JobACL.MODIFY_JOB);
308325
appContext.getEventHandler().handle(
309326
new TaskAttemptDiagnosticsUpdateEvent(taskAttemptId, message));
310327
appContext.getEventHandler().handle(
@@ -322,8 +339,8 @@ public GetDiagnosticsResponse getDiagnostics(
322339

323340
GetDiagnosticsResponse response =
324341
recordFactory.newRecordInstance(GetDiagnosticsResponse.class);
325-
response.addAllDiagnostics(
326-
verifyAndGetAttempt(taskAttemptId, false).getDiagnostics());
342+
response.addAllDiagnostics(verifyAndGetAttempt(taskAttemptId,
343+
JobACL.VIEW_JOB).getDiagnostics());
327344
return response;
328345
}
329346

@@ -332,9 +349,12 @@ public GetDiagnosticsResponse getDiagnostics(
332349
public FailTaskAttemptResponse failTaskAttempt(
333350
FailTaskAttemptRequest request) throws IOException {
334351
TaskAttemptId taskAttemptId = request.getTaskAttemptId();
335-
String message = "Fail task attempt received from client " + taskAttemptId;
352+
UserGroupInformation callerUGI = UserGroupInformation.getCurrentUser();
353+
String message = "Fail task attempt " + taskAttemptId
354+
+ " received from " + callerUGI + " at "
355+
+ Server.getRemoteAddress();
336356
LOG.info(message);
337-
verifyAndGetAttempt(taskAttemptId, true);
357+
verifyAndGetAttempt(taskAttemptId, JobACL.MODIFY_JOB);
338358
appContext.getEventHandler().handle(
339359
new TaskAttemptDiagnosticsUpdateEvent(taskAttemptId, message));
340360
appContext.getEventHandler().handle(
@@ -356,7 +376,7 @@ public GetTaskReportsResponse getTaskReports(
356376
GetTaskReportsResponse response =
357377
recordFactory.newRecordInstance(GetTaskReportsResponse.class);
358378

359-
Job job = verifyAndGetJob(jobId, false);
379+
Job job = verifyAndGetJob(jobId, JobACL.VIEW_JOB);
360380
Collection<Task> tasks = job.getTasks(taskType).values();
361381
LOG.info("Getting task report for " + taskType + " " + jobId
362382
+ ". 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‎

Lines changed: 85 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -18,20 +18,30 @@
1818

1919
package org.apache.hadoop.mapreduce.v2.app;
2020

21+
import static org.junit.Assert.fail;
22+
23+
import java.security.PrivilegedExceptionAction;
2124
import java.util.Iterator;
2225
import java.util.List;
2326

2427
import junit.framework.Assert;
2528

2629
import org.apache.hadoop.conf.Configuration;
30+
import org.apache.hadoop.mapreduce.JobACL;
31+
import org.apache.hadoop.mapreduce.MRConfig;
32+
import org.apache.hadoop.mapreduce.MRJobConfig;
2733
import org.apache.hadoop.mapreduce.v2.api.MRClientProtocol;
34+
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.FailTaskAttemptRequest;
2835
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetCountersRequest;
2936
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetDiagnosticsRequest;
3037
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetJobReportRequest;
3138
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskAttemptCompletionEventsRequest;
3239
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskAttemptReportRequest;
3340
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskReportRequest;
3441
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.GetTaskReportsRequest;
42+
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.KillJobRequest;
43+
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.KillTaskAttemptRequest;
44+
import org.apache.hadoop.mapreduce.v2.api.protocolrecords.KillTaskRequest;
3545
import org.apache.hadoop.mapreduce.v2.api.records.AMInfo;
3646
import org.apache.hadoop.mapreduce.v2.api.records.JobReport;
3747
import org.apache.hadoop.mapreduce.v2.api.records.JobState;
@@ -51,6 +61,8 @@
5161
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptEventType;
5262
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent;
5363
import org.apache.hadoop.mapreduce.v2.app.job.event.TaskAttemptStatusUpdateEvent.TaskAttemptStatus;
64+
import org.apache.hadoop.security.AccessControlException;
65+
import org.apache.hadoop.security.UserGroupInformation;
5466
import org.apache.hadoop.yarn.factories.RecordFactory;
5567
import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
5668
import org.apache.hadoop.yarn.ipc.YarnRPC;
@@ -169,6 +181,79 @@ public void test() throws Exception {
169181
app.waitForState(job, JobState.SUCCEEDED);
170182
}
171183

184+
@Test
185+
public void testViewAclOnlyCannotModify() throws Exception {
186+
final MRAppWithClientService app = new MRAppWithClientService(1, 0, false);
187+
final Configuration conf = new Configuration();
188+
conf.setBoolean(MRConfig.MR_ACLS_ENABLED, true);
189+
conf.set(MRJobConfig.JOB_ACL_VIEW_JOB, "viewonlyuser");
190+
Job job = app.submit(conf);
191+
app.waitForState(job, JobState.RUNNING);
192+
Assert.assertEquals("Num tasks not correct", 1, job.getTasks().size());
193+
Iterator<Task> it = job.getTasks().values().iterator();
194+
Task task = it.next();
195+
app.waitForState(task, TaskState.RUNNING);
196+
TaskAttempt attempt = task.getAttempts().values().iterator().next();
197+
app.waitForState(attempt, TaskAttemptState.RUNNING);
198+
199+
UserGroupInformation viewOnlyUser =
200+
UserGroupInformation.createUserForTesting(
201+
"viewonlyuser", new String[] {});
202+
Assert.assertTrue("viewonlyuser cannot view job",
203+
job.checkAccess(viewOnlyUser, JobACL.VIEW_JOB));
204+
Assert.assertFalse("viewonlyuser can modify job",
205+
job.checkAccess(viewOnlyUser, JobACL.MODIFY_JOB));
206+
MRClientProtocol client = viewOnlyUser.doAs(
207+
new PrivilegedExceptionAction<MRClientProtocol>() {
208+
@Override
209+
public MRClientProtocol run() throws Exception {
210+
YarnRPC rpc = YarnRPC.create(conf);
211+
return (MRClientProtocol) rpc.getProxy(MRClientProtocol.class,
212+
app.clientService.getBindAddress(), conf);
213+
}
214+
});
215+
216+
KillJobRequest killJobRequest = recordFactory.newRecordInstance(
217+
KillJobRequest.class);
218+
killJobRequest.setJobId(app.getJobId());
219+
try {
220+
client.killJob(killJobRequest);
221+
fail("viewonlyuser killed job");
222+
} catch (AccessControlException e) {
223+
// pass
224+
}
225+
226+
KillTaskRequest killTaskRequest = recordFactory.newRecordInstance(
227+
KillTaskRequest.class);
228+
killTaskRequest.setTaskId(task.getID());
229+
try {
230+
client.killTask(killTaskRequest);
231+
fail("viewonlyuser killed task");
232+
} catch (AccessControlException e) {
233+
// pass
234+
}
235+
236+
KillTaskAttemptRequest killTaskAttemptRequest =
237+
recordFactory.newRecordInstance(KillTaskAttemptRequest.class);
238+
killTaskAttemptRequest.setTaskAttemptId(attempt.getID());
239+
try {
240+
client.killTaskAttempt(killTaskAttemptRequest);
241+
fail("viewonlyuser killed task attempt");
242+
} catch (AccessControlException e) {
243+
// pass
244+
}
245+
246+
FailTaskAttemptRequest failTaskAttemptRequest =
247+
recordFactory.newRecordInstance(FailTaskAttemptRequest.class);
248+
failTaskAttemptRequest.setTaskAttemptId(attempt.getID());
249+
try {
250+
client.failTaskAttempt(failTaskAttemptRequest);
251+
fail("viewonlyuser killed task attempt");
252+
} catch (AccessControlException e) {
253+
// pass
254+
}
255+
}
256+
172257
private void verifyJobReport(JobReport jr) {
173258
Assert.assertNotNull("JobReport is null", jr);
174259
List<AMInfo> amInfos = jr.getAMInfos();

0 commit comments

Comments
 (0)