commit 06585af93b7081098a0ae5e83c1957c7a20eb8f0

Author: Arun C Murthy <acmurthy@apache.org>
Date:   Tue Jul 20 10:05:33 2010 -0700

    MAPREDUCE-1207. Sanitize user environment of map/reduce tasks and allow admins to set environment and java options. Contributed by Krishna Ramachandran.
    
    +++ b/YAHOO-CHANGES.txt
    +    MAPREDUCE-1207. Sanitize user environment of map/reduce tasks and allow
    +    admins to set environment and java options. (Krishna Ramachandran via
    +    acmurthy)
    +


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

src/mapred/org/apache/hadoop/mapred/MapTaskRunner.java

@@ -46,11 +46,16 @@ class MapTaskRunner extends TaskRunner {
 
   @Override
   public String getChildJavaOpts(JobConf jobConf, String defaultValue) {
-    return jobConf.get(JobConf.MAPRED_MAP_TASK_JAVA_OPTS, 
-                       super.getChildJavaOpts(jobConf, 
-                           JobConf.DEFAULT_MAPRED_TASK_JAVA_OPTS));
+    String user = 
+      jobConf.get(JobConf.MAPRED_MAP_TASK_JAVA_OPTS, 
+        super.getChildJavaOpts(jobConf, 
+            JobConf.DEFAULT_MAPRED_TASK_JAVA_OPTS));
+    String admin = 
+      jobConf.get(TaskRunner.MAPRED_MAP_ADMIN_JAVA_OPTS,
+        TaskRunner.DEFAULT_MAPRED_ADMIN_JAVA_OPTS);
+    return user + " " + admin;
   }
-  
+
   @Override
   public int getChildUlimit(JobConf jobConf) {
     return jobConf.getInt(JobConf.MAPRED_MAP_TASK_ULIMIT, 

src/mapred/org/apache/hadoop/mapred/ReduceTaskRunner.java

@@ -48,14 +48,18 @@ class ReduceTaskRunner extends TaskRunner {
     getTask().getProgress().setStatus("closed");
     mapOutputFile.removeAll();
   }
-
+  
   @Override
   public String getChildJavaOpts(JobConf jobConf, String defaultValue) {
-    return jobConf.get(JobConf.MAPRED_REDUCE_TASK_JAVA_OPTS, 
-                       super.getChildJavaOpts(jobConf, 
-                           JobConf.DEFAULT_MAPRED_TASK_JAVA_OPTS));
+    String user = 
+      jobConf.get(JobConf.MAPRED_REDUCE_TASK_JAVA_OPTS, 
+        super.getChildJavaOpts(jobConf, 
+            JobConf.DEFAULT_MAPRED_TASK_JAVA_OPTS));
+    String admin = jobConf.get(TaskRunner.MAPRED_REDUCE_ADMIN_JAVA_OPTS,
+        TaskRunner.DEFAULT_MAPRED_ADMIN_JAVA_OPTS);
+    return user + " " + admin;
   }
- 
+
   @Override
   public int getChildUlimit(JobConf jobConf) {
     return jobConf.getInt(JobConf.MAPRED_REDUCE_TASK_ULIMIT, 

src/mapred/org/apache/hadoop/mapred/TaskRunner.java

@@ -53,6 +53,29 @@ import org.apache.hadoop.mapreduce.JobContext;
  * user supplied map and reduce functions.
  */
 abstract class TaskRunner extends Thread {
+
+  static final String MAPRED_MAP_ADMIN_JAVA_OPTS =
+    "mapreduce.admin.map.child.java.opts";
+
+  static final String MAPRED_REDUCE_ADMIN_JAVA_OPTS =
+    "mapreduce.admin.reduce.child.java.opts";
+
+  static final String DEFAULT_MAPRED_ADMIN_JAVA_OPTS =
+    "-Djava.net.preferIPv4Stack=true";
+
+  static final String MAPRED_ADMIN_USER_SHELL =
+    "mapreduce.admin.user.shell";
+  
+  static final String DEFAULT_SHELL = "/bin/bash";
+  
+  static final String MAPRED_ADMIN_USER_HOME_DIR =
+    "mapreduce.admin.user.home.dir";
+
+  static final String DEFAULT_HOME_DIR= "/homes/";
+  
+  static final String MAPRED_ADMIN_USER_ENV =
+    "mapreduce.admin.user.env";
+
   public static final Log LOG =
     LogFactory.getLog(TaskRunner.class);
 
@@ -162,6 +185,7 @@ abstract class TaskRunner extends Thread {
       
       // We don't create any symlinks yet, so presence/absence of workDir
       // actually on the file system doesn't matter.
+      String user = tip.getUGI().getUserName();
       tip.getUGI().doAs(new PrivilegedExceptionAction<Void>() {
         public Void run() throws IOException {
           taskDistributedCacheManager =
@@ -204,7 +228,7 @@ abstract class TaskRunner extends Thread {
                  stderr);
       
       Map<String, String> env = new HashMap<String, String>();
-      errorInfo = getVMEnvironment(errorInfo, workDir, conf, env, taskid,
+      errorInfo = getVMEnvironment(errorInfo, user, workDir, conf, env, taskid,
                                    logSize);
 
       launchJvmAndWait(setup, vargs, stdout, stderr, logSize, workDir, env);
@@ -510,14 +534,7 @@ abstract class TaskRunner extends Thread {
     return classPaths;
   }
 
-  /**
-   * @param errorInfo
-   * @param workDir
-   * @param env
-   * @return
-   * @throws Throwable
-   */
-  private String getVMEnvironment(String errorInfo, File workDir, JobConf conf,
+  private String getVMEnvironment(String errorInfo, String user, File workDir, JobConf conf,
       Map<String, String> env, TaskAttemptID taskid, long logSize)
       throws Throwable {
     StringBuffer ldLibraryPath = new StringBuffer();
@@ -529,11 +546,11 @@ abstract class TaskRunner extends Thread {
       ldLibraryPath.append(oldLdLibraryPath);
     }
     env.put("LD_LIBRARY_PATH", ldLibraryPath.toString());
-
+    //update user configured login-shell properties
+    updateUserLoginEnv(errorInfo, user, conf, env);
     String jobTokenFile = conf.get(TokenCache.JOB_TOKENS_FILENAME);
     LOG.debug("putting jobToken file name into environment fn=" + jobTokenFile);
-    env.put(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION, jobTokenFile);
-    
+    env.put(UserGroupInformation.HADOOP_TOKEN_FILE_LOCATION, jobTokenFile);    
     // for the child of task jvm, set hadoop.root.logger
     env.put("HADOOP_ROOT_LOGGER","INFO,TLA");
     String hadoopClientOpts = System.getenv("HADOOP_CLIENT_OPTS");
@@ -549,6 +566,30 @@ abstract class TaskRunner extends Thread {
 
     // add the env variables passed by the user
     String mapredChildEnv = getChildEnv(conf);
+    return setEnvFromInputString(errorInfo, env, mapredChildEnv);
+  }
+
+  void updateUserLoginEnv(String errorInfo, String user, JobConf config, 
+      Map<String, String> env) 
+      throws Throwable {
+    env.put("USER",user);
+    env.put("SHELL", config.get(MAPRED_ADMIN_USER_SHELL, DEFAULT_SHELL));
+    env.put("LOGNAME", user);
+    env.put("HOME", config.get(MAPRED_ADMIN_USER_HOME_DIR, DEFAULT_HOME_DIR));
+    // additional user configured login properties
+    String customEnv = config.get(MAPRED_ADMIN_USER_ENV);
+    setEnvFromInputString(errorInfo, env, customEnv);
+  }
+
+  /**
+   * @param errorInfo
+   * @param env
+   * @param mapredChildEnv
+   * @return
+   * @throws Throwable
+   */
+  private String setEnvFromInputString(String errorInfo, Map<String, String> env,
+      String mapredChildEnv) throws Throwable {
     if (mapredChildEnv != null && mapredChildEnv.length() > 0) {
       String childEnvs[] = mapredChildEnv.split(",");
       for (String cEnv : childEnvs) {
@@ -560,7 +601,7 @@ abstract class TaskRunner extends Thread {
             // example LD_LIBRARY_PATH=$LD_LIBRARY_PATH:/tmp
             value = parts[1].replace("$" + parts[0], value);
           } else {
-            // this key is not configured by the tt for the child .. get it 
+            // this key is not configured by the tt for the child .. get it
             // from the tt's env
             // example PATH=$PATH:/tmp
             value = System.getenv(parts[0]);
@@ -576,9 +617,9 @@ abstract class TaskRunner extends Thread {
           env.put(parts[0], value);
         } catch (Throwable t) {
           // set the error msg
-          errorInfo = "Invalid User environment settings : " + mapredChildEnv 
-                      + ". Failed to parse user-passed environment param."
-                      + " Expecting : env1=value1,env2=value2...";
+          errorInfo = "Invalid User environment settings : " + mapredChildEnv
+              + ". Failed to parse user-passed environment param."
+              + " Expecting : env1=value1,env2=value2...";
           LOG.warn(errorInfo);
           throw t;
         }

src/test/org/apache/hadoop/mapred/TestTaskEnvironment.java

@@ -0,0 +1,124 @@
+/**
+ * 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.mapred;
+
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.FileReader;
+import java.io.IOException;
+import java.util.Vector;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileUtil;
+import org.apache.hadoop.mapred.JvmManager.JvmManagerForType;
+import org.apache.hadoop.mapred.JvmManager.JvmManagerForType.JvmRunner;
+import org.apache.hadoop.mapred.JvmManager.JvmEnv;
+import org.apache.hadoop.mapred.TaskTracker.TaskInProgress;
+import org.apache.hadoop.mapreduce.TaskType;
+import org.apache.hadoop.mapreduce.server.tasktracker.userlogs.UserLogManager;
+import org.junit.After;
+import static org.junit.Assert.*;
+import org.junit.Before;
+import org.junit.Test;
+
+public class TestTaskEnvironment {
+  private static File TEST_DIR = new File(System.getProperty("test.build.data",
+      "/tmp"), TestJvmManager.class.getSimpleName());
+  private static int MAP_SLOTS = 1;
+  private static int REDUCE_SLOTS = 1;
+  private TaskTracker tt;
+  private JvmManager jvmManager;
+  private JobConf ttConf;
+
+  @Before
+  public void setUp() {
+    TEST_DIR.mkdirs();
+  }
+
+  @After
+  public void tearDown() throws IOException {
+    FileUtil.fullyDelete(TEST_DIR);
+  }
+
+  public TestTaskEnvironment() throws Exception {
+    tt = new TaskTracker();
+    ttConf = new JobConf();
+    ttConf.setLong("mapred.tasktracker.tasks.sleeptime-before-sigkill", 2000);
+    tt.setConf(ttConf);
+    tt.setMaxMapSlots(MAP_SLOTS);
+    tt.setMaxReduceSlots(REDUCE_SLOTS);
+    tt.setTaskController(new DefaultTaskController());
+    jvmManager = new JvmManager(tt);
+    tt.setJvmManagerInstance(jvmManager);
+    tt.setUserLogManager(new UserLogManager(ttConf));
+  }
+
+  // write a shell script to execute the command.
+  private File writeScript(String fileName, String cmd, File pidFile) throws IOException {
+    File script = new File(TEST_DIR, fileName);
+    FileOutputStream out = new FileOutputStream(script);
+    // write pid into a file
+    out.write(("echo $$ >" + pidFile.toString() + ";").getBytes());
+    // ignore SIGTERM
+    out.write(("trap '' 15\n").getBytes());
+    // write the actual command it self.
+    out.write(cmd.getBytes());
+    out.close();
+    script.setExecutable(true);
+    return script;
+  }
+
+  @Test
+  public void testTaskEnv() throws Throwable {
+    ttConf.set("mapreduce.admin.user.shell", "/bin/testshell");
+    ttConf.set("mapreduce.admin.user.env", "key1=value1,key2=value2");
+    final Map<String, String> env = new HashMap<String, String>();
+    String user = "test";
+    JobConf taskConf = new JobConf(ttConf);
+    TaskAttemptID attemptID = new TaskAttemptID("test", 0, true, 0, 0);
+    Task task = new MapTask(null, attemptID, 0, null, MAP_SLOTS);
+    task.setConf(taskConf);
+    TaskInProgress tip = tt.new TaskInProgress(task, taskConf);
+    final TaskRunner taskRunner = task.createRunner(tt, tip);
+    String errorInfo = "Child error";
+    taskRunner.updateUserLoginEnv(errorInfo, user, taskConf, env);
+
+    final Vector<String> vargs = new Vector<String>(1);
+    File pidFile = new File(TEST_DIR, "pid");
+    vargs.add(writeScript("ENV", "/bin/env ", pidFile).getAbsolutePath());
+    final File workDir = new File(TEST_DIR, "work");
+    workDir.mkdir();
+    final File stdout = new File(TEST_DIR, "stdout");
+    final File stderr = new File(TEST_DIR, "stderr");
+    JvmEnv jenv = jvmManager.constructJvmEnv(null, vargs,
+      stdout,stderr, 100, workDir, env, taskConf);
+    Map<String, String> jvmenvmap = jenv.env;
+    String javaOpts = taskRunner.getChildJavaOpts(ttConf, 
+      JobConf.MAPRED_MAP_TASK_JAVA_OPTS);
+
+    assertTrue(jvmenvmap.containsKey("SHELL"));
+    assertTrue(jvmenvmap.containsValue("/bin/testshell"));
+    assertTrue(jvmenvmap.containsKey("key2"));
+    assertTrue(jvmenvmap.containsValue("value2"));
+    assertTrue(javaOpts, javaOpts.contains("Xmx"));
+    assertTrue(javaOpts, javaOpts.contains("IPv4"));
+  }
+}