xattr api cleanup

(cherry picked from commit da59acd8ca)

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

@@ -136,8 +136,7 @@ class FSDirXAttrOp {
     final boolean isRawPath = FSDirectory.isReservedRawName(src);
     final INodesInPath iip = fsd.resolvePath(pc, src, DirOp.READ);
     if (fsd.isPermissionEnabled()) {
-      /* To access xattr names, you need EXECUTE in the owning directory. */
+      fsd.checkPathAccess(pc, iip, FsAction.READ);
-      fsd.checkParentAccess(pc, iip, FsAction.EXECUTE);
     }
     final List<XAttr> all = FSDirXAttrOp.getXAttrs(fsd, iip);
     return XAttrPermissionFilter.

hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/FSXAttrBaseTest.java

@@ -842,15 +842,20 @@ public class FSXAttrBaseTest {
     }
 
     /*
-     * Check that execute/scan access to the parent dir is sufficient to get
+     * Check that execute/scan access to the parent dir is not
-     * xattr names.
+     * sufficient to get xattr names.
      */
     fs.setPermission(path, new FsPermission((short) 0701));
     user.doAs(new PrivilegedExceptionAction<Object>() {
         @Override
         public Object run() throws Exception {
+        try {
           final FileSystem userFs = dfsCluster.getFileSystem();
           userFs.listXAttrs(childDir);
+          fail("expected AccessControlException");
+        } catch (AccessControlException ace) {
+          GenericTestUtils.assertExceptionContains("Permission denied", ace);
+        }
         return null;
       }
       });
@@ -858,12 +863,16 @@ public class FSXAttrBaseTest {
     /*
      * Test that xattrs in the "trusted" namespace are filtered correctly.
      */
+    // Allow the user to read child path.
+    fs.setPermission(childDir, new FsPermission((short) 0704));
     fs.setXAttr(childDir, "trusted.myxattr", "1234".getBytes());
     user.doAs(new PrivilegedExceptionAction<Object>() {
         @Override
         public Object run() throws Exception {
           final FileSystem userFs = dfsCluster.getFileSystem();
-          assertTrue(userFs.listXAttrs(childDir).size() == 1);
+          List<String> xattrs = userFs.listXAttrs(childDir);
+          assertTrue(xattrs.size() == 1);
+          assertEquals(name1, xattrs.get(0));
           return null;
         }
       });
@@ -1108,17 +1117,45 @@ public class FSXAttrBaseTest {
             }
 
             /*
-            * Test that only user who have parent directory execute access
+            * Test that user who have parent directory execute access
-            *  can see raw.* xattrs returned from listXAttr
+            *  can also not see raw.* xattrs returned from listXAttr
             */
+            try {
               // non-raw path
-            final List<String> xattrNames = userFs.listXAttrs(path);
+              userFs.listXAttrs(path);
-            assertTrue(xattrNames.size() == 0);
+              fail("listXAttr should have thrown AccessControlException");
+            } catch (AccessControlException ace) {
+              // expected
+            }
 
+            try {
               // raw path
-            List<String> rawXattrs = userFs.listXAttrs(rawPath);
+              userFs.listXAttrs(rawPath);
-            assertTrue(rawXattrs.size() == 1);
+              fail("listXAttr should have thrown AccessControlException");
-            assertTrue(rawXattrs.get(0).equals(raw1));
+            } catch (AccessControlException ace) {
+              // expected
+            }
+            return null;
+          }
+        });
+      /*
+        Test user who have read access can list xattrs in "raw.*" namespace
+       */
+      fs.setPermission(path, new FsPermission((short) 0751));
+      final Path childDir = new Path(path, "child" + pathCount);
+      FileSystem.mkdirs(fs, childDir, FsPermission.createImmutable((short)
+          0704));
+      final Path rawChildDir =
+          new Path("/.reserved/raw" + childDir.toString());
+      fs.setXAttr(rawChildDir, raw1, value1);
+      user.doAs(new PrivilegedExceptionAction<Object>() {
+        @Override
+        public Object run() throws Exception {
+          final FileSystem userFs = dfsCluster.getFileSystem();
+          // raw path
+          List<String> xattrs = userFs.listXAttrs(rawChildDir);
+          assertEquals(1, xattrs.size());
+          assertEquals(raw1, xattrs.get(0));
           return null;
         }
       });