HADOOP-3186. Fix incorrect permission checkding for mv and renameTo in HDFS. (Tsz Wo (Nicholas), SZE via mukund)

git-svn-id: https://svn.apache.org/repos/asf/hadoop/core/branches/branch-0.16@652209 13f79535-47bb-0310-9956-ffa450edef68

CHANGES.txt

@@ -13,6 +13,9 @@ Release 0.16.4 - 2008-05-05
     HADOOP-3304. [HOD] Fixes the way the logcondense.py utility searches for log
     files that need to be deleted. (yhemanth via mukund)
 
+    HADOOP-3186. Fix incorrect permission checkding for mv and renameTo
+    in HDFS. (Tsz Wo (Nicholas), SZE via mukund)
+
 Release 0.16.3 - 2008-04-16
 
   BUG FIXES

src/java/org/apache/hadoop/dfs/FSDirectory.java

@@ -273,8 +273,10 @@ class FSDirectory implements FSConstants {
    * Change the filename
    */
   public boolean renameTo(String src, String dst) {
-    NameNode.stateChangeLog.debug("DIR* FSDirectory.renameTo: "
+    if (NameNode.stateChangeLog.isDebugEnabled()) {
+      NameNode.stateChangeLog.debug("DIR* FSDirectory.renameTo: "
                                   +src+" to "+dst);
+    }
     waitForReady();
     long now = FSNamesystem.now();
     if (!unprotectedRenameTo(src, dst, now))
@@ -294,7 +296,7 @@ class FSDirectory implements FSConstants {
         return false;
       }
       if (isDir(dst)) {
-        dst += "/" + new File(src).getName();
+        dst += Path.SEPARATOR + new Path(src).getName();
       }
       if (rootDir.getNode(dst) != null) {
         NameNode.stateChangeLog.warn("DIR* FSDirectory.unprotectedRenameTo: "

src/java/org/apache/hadoop/dfs/FSNamesystem.java

@@ -1438,8 +1438,12 @@ class FSNamesystem implements FSConstants, FSNamesystemMBean {
     }
 
     if (isPermissionEnabled) {
+      //We should not be doing this.  This is move() not renameTo().
+      //but for now,
+      String actualdst = dir.isDir(dst)?
+          dst + Path.SEPARATOR + new Path(src).getName(): dst;
       checkParentAccess(src, FsAction.WRITE);
-      checkAncestorAccess(dst, FsAction.WRITE);
+      checkAncestorAccess(actualdst, FsAction.WRITE);
     }
 
     return dir.renameTo(src, dst);

src/java/org/apache/hadoop/dfs/PermissionChecker.java

@@ -84,12 +84,14 @@ class PermissionChecker {
   void checkPermission(String path, INodeDirectory root, boolean doCheckOwner,
       FsAction ancestorAccess, FsAction parentAccess, FsAction access,
       FsAction subAccess) throws AccessControlException {
-    LOG.debug("ACCESS CHECK: " + this
-        + ", doCheckOwner=" + doCheckOwner
-        + ", ancestorAccess=" + ancestorAccess
-        + ", parentAccess=" + parentAccess
-        + ", access=" + access
-        + ", subAccess=" + subAccess);
+    if (LOG.isDebugEnabled()) {
+      LOG.debug("ACCESS CHECK: " + this
+          + ", doCheckOwner=" + doCheckOwner
+          + ", ancestorAccess=" + ancestorAccess
+          + ", parentAccess=" + parentAccess
+          + ", access=" + access
+          + ", subAccess=" + subAccess);
+    }
 
     synchronized(root) {
       INode[] inodes = root.getExistingPathINodes(path);

src/test/org/apache/hadoop/security/TestPermission.java

@@ -41,8 +41,8 @@ public class TestPermission extends TestCase {
   }
 
   final private static Path ROOT_PATH = new Path("/data");
-  final private static Path CHILD_DIR1 = new Path(ROOT_PATH, "web1");
-  final private static Path CHILD_DIR2 = new Path(ROOT_PATH, "web2");
+  final private static Path CHILD_DIR1 = new Path(ROOT_PATH, "child1");
+  final private static Path CHILD_DIR2 = new Path(ROOT_PATH, "child2");
   final private static Path CHILD_FILE1 = new Path(ROOT_PATH, "file1");
   final private static Path CHILD_FILE2 = new Path(ROOT_PATH, "file2");
 
@@ -50,10 +50,9 @@ public class TestPermission extends TestCase {
   final private static String PERMISSION_EXCEPTION_NAME =
     AccessControlException.class.getName();
 
-  final private static String USER_NAME = "Who";
-  final private static String GROUP1_NAME = "group1";
-  final private static String GROUP2_NAME = "group2";
-  final private static String[] GROUP_NAMES = {GROUP1_NAME, GROUP2_NAME};
+  final private static Random RAN = new Random();
+  final private static String USER_NAME = "user" + RAN.nextInt();
+  final private static String[] GROUP_NAMES = {"group1", "group2"};
 
   static FsPermission checkPermission(FileSystem fs,
       String path, FsPermission expected) throws IOException {
@@ -116,67 +115,72 @@ public class TestPermission extends TestCase {
     conf.setBoolean("dfs.permissions", true);
     MiniDFSCluster cluster = new MiniDFSCluster(conf, 3, true, null);
     cluster.waitActive();
-    FileSystem fs = FileSystem.get(conf);
 
     try {
+      FileSystem nnfs = FileSystem.get(conf);
       // test permissions on files that do not exist
-      assertFalse(fs.exists(CHILD_FILE1));
+      assertFalse(nnfs.exists(CHILD_FILE1));
       try {
-        fs.setOwner(CHILD_FILE1, "foo", "bar");
+        nnfs.setOwner(CHILD_FILE1, "foo", "bar");
         assertTrue(false);
       }
       catch(java.io.FileNotFoundException e) {
         LOG.info("GOOD: got " + e);
       }
       try {
-        fs.setPermission(CHILD_FILE1, new FsPermission((short)0777));
+        nnfs.setPermission(CHILD_FILE1, new FsPermission((short)0777));
         assertTrue(false);
       }
       catch(java.io.FileNotFoundException e) {
         LOG.info("GOOD: got " + e);
       }
       // following dir/file creations are legal
-      fs.mkdirs(CHILD_DIR1);
-      FSDataOutputStream out = fs.create(CHILD_FILE1);
+      nnfs.mkdirs(CHILD_DIR1);
+      FSDataOutputStream out = nnfs.create(CHILD_FILE1);
       byte data[] = new byte[FILE_LEN];
-      Random r = new Random();
-      r.nextBytes(data);
+      RAN.nextBytes(data);
       out.write(data);
       out.close();
-      fs.setPermission(CHILD_FILE1, new FsPermission((short)0700));
+      nnfs.setPermission(CHILD_FILE1, new FsPermission((short)0700));
 
       // following read is legal
       byte dataIn[] = new byte[FILE_LEN];
-      FSDataInputStream fin = fs.open(CHILD_FILE1);
-      fin.read(dataIn);
+      FSDataInputStream fin = nnfs.open(CHILD_FILE1);
+      int bytesRead = fin.read(dataIn);
+      assertTrue(bytesRead == FILE_LEN);
       for(int i=0; i<FILE_LEN; i++) {
         assertEquals(data[i], dataIn[i]);
       }
-      fs.close();
 
+      ////////////////////////////////////////////////////////////////
       // test illegal file/dir creation
       UnixUserGroupInformation userGroupInfo = new UnixUserGroupInformation(
           USER_NAME, GROUP_NAMES );
-      conf.set(UnixUserGroupInformation.UGI_PROPERTY_NAME,
-          userGroupInfo.toString());
-      fs = FileSystem.get(conf);
+      UnixUserGroupInformation.saveToConf(conf,
+          UnixUserGroupInformation.UGI_PROPERTY_NAME, userGroupInfo);
+      FileSystem userfs = FileSystem.get(conf);
 
       // make sure mkdir of a existing directory that is not owned by 
       // this user does not throw an exception.
-      fs.mkdirs(CHILD_DIR1);
+      userfs.mkdirs(CHILD_DIR1);
       
       // illegal mkdir
-      assertTrue(!canMkdirs(fs, CHILD_DIR2));
+      assertTrue(!canMkdirs(userfs, CHILD_DIR2));
 
       // illegal file creation
-      assertTrue(!canCreate(fs, CHILD_FILE2));
+      assertTrue(!canCreate(userfs, CHILD_FILE2));
 
       // illegal file open
-      assertTrue(!canOpen(fs, CHILD_FILE1));
-    }
-    finally {
-      try{fs.close();} catch(Exception e) {}
-      try{cluster.shutdown();} catch(Exception e) {}
+      assertTrue(!canOpen(userfs, CHILD_FILE1));
+
+      nnfs.setPermission(ROOT_PATH, new FsPermission((short)0755));
+      nnfs.setPermission(CHILD_DIR1, new FsPermission((short)0777));
+      nnfs.setPermission(new Path("/"), new FsPermission((short)0777));
+      final Path RENAME_PATH = new Path("/foo/bar");
+      userfs.mkdirs(RENAME_PATH);
+      assertTrue(canRename(userfs, RENAME_PATH, CHILD_DIR1));
+    } finally {
+      if(cluster != null) cluster.shutdown();
     }
   }
 
@@ -209,4 +213,15 @@ public class TestPermission extends TestCase {
       return false;
     }
   }
+
+  static boolean canRename(FileSystem fs, Path src, Path dst
+      ) throws IOException {
+    try {
+      fs.rename(src, dst);
+      return true;
+    } catch(RemoteException e) {
+      assertEquals(PERMISSION_EXCEPTION_NAME, e.getClassName());
+      return false;
+    }
+  }
 }