HDFS-6374. setXAttr should require the user to be the owner of the file or directory. Contributed by Charles Lamb.

git-svn-id: https://svn.apache.org/repos/asf/hadoop/common/branches/HDFS-2006@1595387 13f79535-47bb-0310-9956-ffa450edef68

hadoop-hdfs-project/hadoop-hdfs/CHANGES-HDFS-2006.txt

@@ -67,3 +67,6 @@ HDFS-2006 (Unreleased)
 
     HDFS-6414. xattr modification operations are based on state of latest
     snapshot instead of current version of inode. (Andrew Wang via cnauroth)
+
+    HDFS-6374. setXAttr should require the user to be the owner of the file
+    or directory (Charles Lamb via wang)

hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSNamesystem.java

@@ -7853,6 +7853,7 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       checkNameNodeSafeMode("Cannot set XAttr on " + src);
       src = FSDirectory.resolvePath(src, pathComponents, dir);
       if (isPermissionEnabled) {
+        checkOwner(pc, src);
         checkPathAccess(pc, src, FsAction.WRITE);
       }
       dir.setXAttr(src, xAttr, flag, logRetryCache);
@@ -7949,6 +7950,7 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       checkNameNodeSafeMode("Cannot remove XAttr entry on " + src);
       src = FSDirectory.resolvePath(src, pathComponents, dir);
       if (isPermissionEnabled) {
+        checkOwner(pc, src);
         checkPathAccess(pc, src, FsAction.WRITE);
       }
       
@@ -8048,6 +8050,5 @@ public class FSNamesystem implements Namesystem, FSClusterStats,
       logger.addAppender(asyncAppender);        
     }
   }
-
 }
 

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/TestDFSShell.java

@@ -155,7 +155,7 @@ public class TestDFSShell {
   }
   
   @Test (timeout = 30000)
-  public void testRecrusiveRm() throws IOException {
+  public void testRecursiveRm() throws IOException {
 	  Configuration conf = new HdfsConfiguration();
 	  MiniDFSCluster cluster = new MiniDFSCluster.Builder(conf).numDataNodes(2).build();
 	  FileSystem fs = cluster.getFileSystem();
@@ -1583,6 +1583,7 @@ public class TestDFSShell {
       cluster.shutdown();
     }
   }
+
   private static String runLsr(final FsShell shell, String root, int returnvalue
       ) throws Exception {
     System.out.println("root=" + root + ", returnvalue=" + returnvalue);
@@ -2052,6 +2053,156 @@ public class TestDFSShell {
     out.reset();
   }
 
+  /**
+   * HDFS-6374 setXAttr should require the user to be the owner of the file
+   * or directory.
+   *
+   * Test to make sure that only the owner of a file or directory can set
+   * or remove the xattrs.
+   *
+   * As user1:
+   * Create a directory (/foo) as user1, chown it to user1 (and user1's group),
+   * grant rwx to "other".
+   *
+   * As user2:
+   * Set an xattr (should fail).
+   *
+   * As user1:
+   * Set an xattr (should pass).
+   *
+   * As user2:
+   * Read the xattr (should pass).
+   * Remove the xattr (should fail).
+   *
+   * As user1:
+   * Read the xattr (should pass).
+   * Remove the xattr (should pass).
+   */
+  @Test (timeout = 30000)
+  public void testSetXAttrPermissionAsDifferentOwner() throws Exception {
+    final String USER1 = "user1";
+    final String GROUP1 = "mygroup1";
+    final UserGroupInformation user1 = UserGroupInformation.
+        createUserForTesting(USER1, new String[] {GROUP1});
+    final UserGroupInformation user2 = UserGroupInformation.
+        createUserForTesting("user2", new String[] {"mygroup2"});
+    MiniDFSCluster cluster = null;
+    PrintStream bak = null;
+    try {
+      final Configuration conf = new HdfsConfiguration();
+      cluster = new MiniDFSCluster.Builder(conf).numDataNodes(1).build();
+      cluster.waitActive();
+
+      final FileSystem fs = cluster.getFileSystem();
+      fs.setOwner(new Path("/"), USER1, GROUP1);
+      bak = System.err;
+
+      final FsShell fshell = new FsShell(conf);
+      final ByteArrayOutputStream out = new ByteArrayOutputStream();
+      System.setErr(new PrintStream(out));
+
+      // mkdir foo as user1
+      user1.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          final int ret = ToolRunner.run(fshell, new String[]{
+              "-mkdir", "/foo"});
+          assertEquals("Return should be 0", 0, ret);
+          out.reset();
+          return null;
+        }
+      });
+
+      user1.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          // Give access to "other"
+          final int ret = ToolRunner.run(fshell, new String[]{
+              "-chmod", "707", "/foo"});
+          assertEquals("Return should be 0", 0, ret);
+          out.reset();
+          return null;
+        }
+      });
+
+      // No permission to write xattr for non-owning user (user2).
+      user2.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          final int ret = ToolRunner.run(fshell, new String[]{
+              "-setfattr", "-n", "user.a1", "-v", "1234", "/foo"});
+          assertEquals("Returned should be 1", 1, ret);
+          final String str = out.toString();
+          assertTrue("Permission denied printed",
+              str.indexOf("Permission denied") != -1);
+          out.reset();
+          return null;
+        }
+      });
+
+      // But there should be permission to write xattr for
+      // the owning user.
+      user1.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          final int ret = ToolRunner.run(fshell, new String[]{
+              "-setfattr", "-n", "user.a1", "-v", "1234", "/foo"});
+          assertEquals("Returned should be 0", 0, ret);
+          out.reset();
+          return null;
+        }
+      });
+
+      // There should be permission to read,but not to remove for
+      // non-owning user (user2).
+      user2.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          // Read
+          int ret = ToolRunner.run(fshell, new String[]{
+              "-getfattr", "-n", "user.a1", "/foo"});
+          assertEquals("Returned should be 0", 0, ret);
+          out.reset();
+          // Remove
+          ret = ToolRunner.run(fshell, new String[]{
+              "-setfattr", "-x", "user.a1", "/foo"});
+          assertEquals("Returned should be 1", 1, ret);
+          final String str = out.toString();
+          assertTrue("Permission denied printed",
+              str.indexOf("Permission denied") != -1);
+          out.reset();
+          return null;
+        }
+      });
+
+      // But there should be permission to read/remove for
+      // the owning user.
+      user1.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          // Read
+          int ret = ToolRunner.run(fshell, new String[]{
+              "-getfattr", "-n", "user.a1", "/foo"});
+          assertEquals("Returned should be 0", 0, ret);
+          out.reset();
+          // Remove
+          ret = ToolRunner.run(fshell, new String[]{
+              "-setfattr", "-x", "user.a1", "/foo"});
+          assertEquals("Returned should be 0", 0, ret);
+          out.reset();
+          return null;
+        }
+      });
+    } finally {
+      if (bak != null) {
+        System.setErr(bak);
+      }
+      if (cluster != null) {
+        cluster.shutdown();
+      }
+    }
+  }
+
   /**
    * Test that the server trash configuration is respected when
    * the client configuration is not set.