MAPREDUCE-2766. Fixed NM to set secure permissions for files and directories in distributed-cache. Contributed by Hitesh Shah.

svn merge -c r1195340 --ignore-ancestry ../../trunk/


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

hadoop-mapreduce-project/CHANGES.txt

@@ -1826,6 +1826,9 @@ Release 0.23.0 - Unreleased
     MAPREDUCE-3313. Fixed initialization of ClusterMetrics which was failing
     TestResourceTrackerService sometimes. (Hitesh Shah via vinodkv)
 
+    MAPREDUCE-2766. Fixed NM to set secure permissions for files and directories
+    in distributed-cache. (Hitesh Shah via vinodkv)
+
 Release 0.22.0 - Unreleased
 
   INCOMPATIBLE CHANGES

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/FSDownload.java

@@ -40,6 +40,7 @@ import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.util.RunJar;
 import org.apache.hadoop.yarn.api.records.LocalResource;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 import org.apache.hadoop.yarn.util.ConverterUtils;
 
 /**
@@ -56,7 +57,13 @@ public class FSDownload implements Callable<Path> {
   private Configuration conf;
   private LocalResource resource;
   private LocalDirAllocator dirs;
-  private FsPermission cachePerms = new FsPermission((short) 0755);
+  private static final FsPermission cachePerms = new FsPermission(
+      (short) 0755);
+  static final FsPermission PUBLIC_FILE_PERMS = new FsPermission((short) 0555);
+  static final FsPermission PRIVATE_FILE_PERMS = new FsPermission(
+      (short) 0500);
+  static final FsPermission PUBLIC_DIR_PERMS = new FsPermission((short) 0755);
+  static final FsPermission PRIVATE_DIR_PERMS = new FsPermission((short) 0700);
 
   FSDownload(FileContext files, UserGroupInformation ugi, Configuration conf,
       LocalDirAllocator dirs, LocalResource resource, Random rand) {
@@ -150,6 +157,7 @@ public class FSDownload implements Callable<Path> {
             };
       });
       unpack(new File(dTmp.toUri()), new File(dFinal.toUri()));
+      changePermissions(dFinal.getFileSystem(conf), dFinal);
       files.rename(dst_work, dst, Rename.OVERWRITE);
     } catch (Exception e) {
       try { files.delete(dst, true); } catch (IOException ignore) { }
@@ -163,11 +171,56 @@ public class FSDownload implements Callable<Path> {
       conf = null;
       resource = null;
       dirs = null;
-      cachePerms = null;
     }
     return files.makeQualified(new Path(dst, sCopy.getName()));
   }
 
+  /**
+   * Recursively change permissions of all files/dirs on path based 
+   * on resource visibility.
+   * Change to 755 or 700 for dirs, 555 or 500 for files.
+   * @param fs FileSystem
+   * @param path Path to modify perms for
+   * @throws IOException
+   * @throws InterruptedException 
+   */
+  private void changePermissions(FileSystem fs, final Path path)
+      throws IOException, InterruptedException {
+    FileStatus fStatus = fs.getFileStatus(path);
+    FsPermission perm = cachePerms;
+    // set public perms as 755 or 555 based on dir or file
+    if (resource.getVisibility() == LocalResourceVisibility.PUBLIC) {
+      perm = fStatus.isDirectory() ? PUBLIC_DIR_PERMS : PUBLIC_FILE_PERMS;
+    }
+    // set private perms as 700 or 500
+    else {
+      // PRIVATE:
+      // APPLICATION:
+      perm = fStatus.isDirectory() ? PRIVATE_DIR_PERMS : PRIVATE_FILE_PERMS;
+    }
+    LOG.debug("Changing permissions for path " + path
+        + " to perm " + perm);
+    final FsPermission fPerm = perm;
+    if (null == userUgi) {
+      files.setPermission(path, perm);
+    }
+    else {
+      userUgi.doAs(new PrivilegedExceptionAction<Void>() {
+        public Void run() throws Exception {
+          files.setPermission(path, fPerm);
+          return null;
+        }
+      });
+    }
+    if (fStatus.isDirectory()
+        && !fStatus.isSymlink()) {
+      FileStatus[] statuses = fs.listStatus(path);
+      for (FileStatus status : statuses) {
+        changePermissions(fs, status.getPath());
+      }
+    }
+  }
+
   private static long getEstimatedSize(LocalResource rsrc) {
     if (rsrc.getSize() < 0) {
       return -1;

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/main/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/LocalResourceRequest.java

@@ -33,6 +33,7 @@ public class LocalResourceRequest
   private final Path loc;
   private final long timestamp;
   private final LocalResourceType type;
+  private final LocalResourceVisibility visibility;
 
   /**
    * Wrap API resource to match against cache of localized resources.
@@ -43,13 +44,16 @@ public class LocalResourceRequest
       throws URISyntaxException {
     this(ConverterUtils.getPathFromYarnURL(resource.getResource()),
         resource.getTimestamp(),
-        resource.getType());
+        resource.getType(),
+        resource.getVisibility());
   }
 
-  LocalResourceRequest(Path loc, long timestamp, LocalResourceType type) {
+  LocalResourceRequest(Path loc, long timestamp, LocalResourceType type,
+      LocalResourceVisibility visibility) {
     this.loc = loc;
     this.timestamp = timestamp;
     this.type = type;
+    this.visibility = visibility;
   }
 
   @Override
@@ -114,7 +118,7 @@ public class LocalResourceRequest
 
   @Override
   public LocalResourceVisibility getVisibility() {
-    throw new UnsupportedOperationException();
+    return visibility;
   }
 
   @Override

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestFSDownload.java

@@ -18,6 +18,12 @@
 
 package org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer;
 
+import static org.apache.hadoop.fs.CreateFlag.CREATE;
+import static org.apache.hadoop.fs.CreateFlag.OVERWRITE;
+import static org.junit.Assert.assertEquals;
+
+import java.io.File;
+import java.io.FileOutputStream;
 import java.io.IOException;
 import java.net.URISyntaxException;
 import java.util.EnumSet;
@@ -28,29 +34,35 @@ import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
+import java.util.jar.JarOutputStream;
+import java.util.jar.Manifest;
+
+import junit.framework.Assert;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.FSDataOutputStream;
 import org.apache.hadoop.fs.FileContext;
-import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.FileStatus;
+import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.LocalDirAllocator;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.yarn.api.records.LocalResource;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 import org.apache.hadoop.yarn.factories.RecordFactory;
 import org.apache.hadoop.yarn.factory.providers.RecordFactoryProvider;
-import org.apache.hadoop.yarn.server.nodemanager.containermanager.localizer.FSDownload;
 import org.apache.hadoop.yarn.util.ConverterUtils;
-
-import static org.apache.hadoop.fs.CreateFlag.*;
-
-
 import org.junit.AfterClass;
 import org.junit.Test;
-import static org.junit.Assert.*;
 
 public class TestFSDownload {
 
+  private static final Log LOG = LogFactory.getLog(TestFSDownload.class);
+  
   @AfterClass
   public static void deleteTestDir() throws IOException {
     FileContext fs = FileContext.getLocalFSFileContext();
@@ -61,7 +73,7 @@ public class TestFSDownload {
     RecordFactoryProvider.getRecordFactory(null);
 
   static LocalResource createFile(FileContext files, Path p, int len,
-      Random r) throws IOException, URISyntaxException {
+      Random r, LocalResourceVisibility vis) throws IOException {
     FSDataOutputStream out = null;
     try {
       byte[] bytes = new byte[len];
@@ -75,10 +87,30 @@ public class TestFSDownload {
     ret.setResource(ConverterUtils.getYarnUrlFromPath(p));
     ret.setSize(len);
     ret.setType(LocalResourceType.FILE);
+    ret.setVisibility(vis);
     ret.setTimestamp(files.getFileStatus(p).getModificationTime());
     return ret;
   }
 
+  static LocalResource createJar(FileContext files, Path p,
+      LocalResourceVisibility vis) throws IOException {
+    LOG.info("Create jar file " + p);
+    File jarFile = new File((files.makeQualified(p)).toUri());
+    FileOutputStream stream = new FileOutputStream(jarFile);
+    LOG.info("Create jar out stream ");
+    JarOutputStream out = new JarOutputStream(stream, new Manifest());
+    LOG.info("Done writing jar stream ");
+    out.close();
+    LocalResource ret = recordFactory.newRecordInstance(LocalResource.class);
+    ret.setResource(ConverterUtils.getYarnUrlFromPath(p));
+    FileStatus status = files.getFileStatus(p);
+    ret.setSize(status.getLen());
+    ret.setTimestamp(status.getModificationTime());
+    ret.setType(LocalResourceType.ARCHIVE);
+    ret.setVisibility(vis);
+    return ret;
+  }
+  
   @Test
   public void testDownload() throws IOException, URISyntaxException,
       InterruptedException {
@@ -88,6 +120,9 @@ public class TestFSDownload {
       TestFSDownload.class.getSimpleName()));
     files.mkdir(basedir, null, true);
     conf.setStrings(TestFSDownload.class.getName(), basedir.toString());
+    
+    Map<LocalResource, LocalResourceVisibility> rsrcVis =
+        new HashMap<LocalResource, LocalResourceVisibility>();
 
     Random rand = new Random();
     long sharedSeed = rand.nextLong();
@@ -102,8 +137,19 @@ public class TestFSDownload {
     int[] sizes = new int[10];
     for (int i = 0; i < 10; ++i) {
       sizes[i] = rand.nextInt(512) + 512;
+      LocalResourceVisibility vis = LocalResourceVisibility.PUBLIC;
+      switch (i%3) {
+      case 1:
+        vis = LocalResourceVisibility.PRIVATE;
+        break;
+      case 2:
+        vis = LocalResourceVisibility.APPLICATION;
+        break;       
+      }
+      
       LocalResource rsrc = createFile(files, new Path(basedir, "" + i),
-          sizes[i], rand);
+          sizes[i], rand, vis);
+      rsrcVis.put(rsrc, vis);
       FSDownload fsd =
           new FSDownload(files, UserGroupInformation.getCurrentUser(), conf,
               dirs, rsrc, new Random(sharedSeed));
@@ -115,6 +161,22 @@ public class TestFSDownload {
         Path localized = p.getValue().get();
         assertEquals(sizes[Integer.valueOf(localized.getName())], p.getKey()
             .getSize());
+        FileStatus status = files.getFileStatus(localized);
+        FsPermission perm = status.getPermission();
+        System.out.println("File permission " + perm + 
+            " for rsrc vis " + p.getKey().getVisibility().name());
+        assert(rsrcVis.containsKey(p.getKey()));
+        switch (rsrcVis.get(p.getKey())) {
+        case PUBLIC:
+          Assert.assertTrue("Public file should be 555",
+              perm.toShort() == FSDownload.PUBLIC_FILE_PERMS.toShort());
+          break;
+        case PRIVATE:
+        case APPLICATION:
+          Assert.assertTrue("Private file should be 500",
+              perm.toShort() == FSDownload.PRIVATE_FILE_PERMS.toShort());          
+          break;
+        }
       }
     } catch (ExecutionException e) {
       throw new IOException("Failed exec", e);
@@ -122,5 +184,101 @@ public class TestFSDownload {
       exec.shutdown();
     }
   }
+  
+  private void verifyPermsRecursively(FileSystem fs,
+      FileContext files, Path p,
+      LocalResourceVisibility vis) throws IOException {
+    FileStatus status = files.getFileStatus(p);
+    if (status.isDirectory()) {
+      if (vis == LocalResourceVisibility.PUBLIC) {
+        Assert.assertTrue(status.getPermission().toShort() ==
+          FSDownload.PUBLIC_DIR_PERMS.toShort());
+      }
+      else {
+        Assert.assertTrue(status.getPermission().toShort() ==
+          FSDownload.PRIVATE_DIR_PERMS.toShort());
+      }
+      if (!status.isSymlink()) {
+        FileStatus[] statuses = fs.listStatus(p);
+        for (FileStatus stat : statuses) {
+          verifyPermsRecursively(fs, files, stat.getPath(), vis);
+        }
+      }
+    }
+    else {
+      if (vis == LocalResourceVisibility.PUBLIC) {
+        Assert.assertTrue(status.getPermission().toShort() ==
+          FSDownload.PUBLIC_FILE_PERMS.toShort());
+      }
+      else {
+        Assert.assertTrue(status.getPermission().toShort() ==
+          FSDownload.PRIVATE_FILE_PERMS.toShort());
+      }
+    }      
+  }
+  
+  @Test
+  public void testDirDownload() throws IOException, InterruptedException {
+    Configuration conf = new Configuration();
+    FileContext files = FileContext.getLocalFSFileContext(conf);
+    final Path basedir = files.makeQualified(new Path("target",
+      TestFSDownload.class.getSimpleName()));
+    files.mkdir(basedir, null, true);
+    conf.setStrings(TestFSDownload.class.getName(), basedir.toString());
+    
+    Map<LocalResource, LocalResourceVisibility> rsrcVis =
+        new HashMap<LocalResource, LocalResourceVisibility>();
+  
+    Random rand = new Random();
+    long sharedSeed = rand.nextLong();
+    rand.setSeed(sharedSeed);
+    System.out.println("SEED: " + sharedSeed);
+
+    Map<LocalResource,Future<Path>> pending =
+      new HashMap<LocalResource,Future<Path>>();
+    ExecutorService exec = Executors.newSingleThreadExecutor();
+    LocalDirAllocator dirs =
+      new LocalDirAllocator(TestFSDownload.class.getName());
+    for (int i = 0; i < 5; ++i) {
+      LocalResourceVisibility vis = LocalResourceVisibility.PUBLIC;
+      switch (rand.nextInt()%3) {
+      case 1:
+        vis = LocalResourceVisibility.PRIVATE;
+        break;
+      case 2:
+        vis = LocalResourceVisibility.APPLICATION;
+        break;       
+      }
+      
+      LocalResource rsrc = createJar(files, new Path(basedir, "dir" + i
+          + ".jar"), vis);
+      rsrcVis.put(rsrc, vis);
+      FSDownload fsd =
+          new FSDownload(files, UserGroupInformation.getCurrentUser(), conf,
+              dirs, rsrc, new Random(sharedSeed));
+      pending.put(rsrc, exec.submit(fsd));
+    }
+    
+    try {
+      
+      for (Map.Entry<LocalResource,Future<Path>> p : pending.entrySet()) {
+        Path localized = p.getValue().get();
+        FileStatus status = files.getFileStatus(localized);
 
+        System.out.println("Testing path " + localized);
+        assert(status.isDirectory());
+        assert(rsrcVis.containsKey(p.getKey()));
+        
+        verifyPermsRecursively(localized.getFileSystem(conf),
+            files, localized, rsrcVis.get(p.getKey()));
+      }
+    } catch (ExecutionException e) {
+      throw new IOException("Failed exec", e);
+    } finally {
+      exec.shutdown();
+    }
+    
+    
+
+  }
 }

hadoop-mapreduce-project/hadoop-yarn/hadoop-yarn-server/hadoop-yarn-server-nodemanager/src/test/java/org/apache/hadoop/yarn/server/nodemanager/containermanager/localizer/TestResourceRetention.java

@@ -23,6 +23,7 @@ import java.util.concurrent.ConcurrentMap;
 
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.yarn.api.records.LocalResourceType;
+import org.apache.hadoop.yarn.api.records.LocalResourceVisibility;
 import org.apache.hadoop.yarn.server.nodemanager.DeletionService;
 
 import org.junit.Test;
@@ -82,7 +83,7 @@ public class TestResourceRetention {
     for (int i = 0; i < nRsrcs; ++i) {
       final LocalResourceRequest req = new LocalResourceRequest(
           new Path("file:///" + user + "/rsrc" + i), timestamp + i * tsstep,
-          LocalResourceType.FILE);
+          LocalResourceType.FILE, LocalResourceVisibility.PUBLIC);
       final long ts = timestamp + i * tsstep;
       final Path p = new Path("file:///local/" + user + "/rsrc" + i);
       LocalizedResource rsrc = new LocalizedResource(req, null) {