From b406d8e3755d24ce72c443fd893a5672fd56babc Mon Sep 17 00:00:00 2001
From: Manoj Govindassamy <manojpec@apache.org>
Date: Mon, 16 Oct 2017 17:42:41 -0700
Subject: [PATCH] HDFS-12614. FSPermissionChecker#getINodeAttrs() throws NPE
 when INodeAttributesProvider configured.

---
 .../server/namenode/FSPermissionChecker.java  | 12 +++-
 .../namenode/TestINodeAttributeProvider.java  | 60 ++++++++++++++-----
 2 files changed, 54 insertions(+), 18 deletions(-)

diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java
index f745a6ceb0..c854b49fb4 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/main/java/org/apache/hadoop/hdfs/server/namenode/FSPermissionChecker.java
@@ -275,8 +275,16 @@ private INodeAttributes getINodeAttrs(byte[][] pathByNameArr, int pathIdx,
     INodeAttributes inodeAttrs = inode.getSnapshotINode(snapshotId);
     if (getAttributesProvider() != null) {
       String[] elements = new String[pathIdx + 1];
-      for (int i = 0; i < elements.length; i++) {
-        elements[i] = DFSUtil.bytes2String(pathByNameArr[i]);
+      /**
+       * {@link INode#getPathComponents(String)} returns a null component
+       * for the root only path "/". Assign an empty string if so.
+       */
+      if (pathByNameArr.length == 1 && pathByNameArr[0] == null) {
+        elements[0] = "";
+      } else {
+        for (int i = 0; i < elements.length; i++) {
+          elements[i] = DFSUtil.bytes2String(pathByNameArr[i]);
+        }
       }
       inodeAttrs = getAttributesProvider().getAttributes(elements, inodeAttrs);
     }
diff --git a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestINodeAttributeProvider.java b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestINodeAttributeProvider.java
index bbc5fa0f9d..9c7dcd3352 100644
--- a/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestINodeAttributeProvider.java
+++ b/hadoop-hdfs-project/hadoop-hdfs/src/test/java/org/apache/hadoop/hdfs/server/namenode/TestINodeAttributeProvider.java
@@ -313,31 +313,59 @@ public void testAuthzBypassingProvider() throws Exception {
     testBypassProviderHelper(users, HDFS_PERMISSION, true);
   }
 
-  @Test
-  public void testCustomProvider() throws Exception {
+  private void verifyFileStatus(UserGroupInformation ugi) throws IOException {
     FileSystem fs = FileSystem.get(miniDFS.getConfiguration(0));
-    fs.mkdirs(new Path("/user/xxx"));
-    FileStatus status = fs.getFileStatus(new Path("/user/xxx"));
-    Assert.assertEquals(System.getProperty("user.name"), status.getOwner());
+
+    FileStatus status = fs.getFileStatus(new Path("/"));
+    LOG.info("Path '/' is owned by: "
+        + status.getOwner() + ":" + status.getGroup());
+
+    Path userDir = new Path("/user/" + ugi.getShortUserName());
+    fs.mkdirs(userDir);
+    status = fs.getFileStatus(userDir);
+    Assert.assertEquals(ugi.getShortUserName(), status.getOwner());
     Assert.assertEquals("supergroup", status.getGroup());
     Assert.assertEquals(new FsPermission((short) 0755), status.getPermission());
-    fs.mkdirs(new Path("/user/authz"));
-    Path p = new Path("/user/authz");
-    status = fs.getFileStatus(p);
+
+    Path authzDir = new Path("/user/authz");
+    fs.mkdirs(authzDir);
+    status = fs.getFileStatus(authzDir);
     Assert.assertEquals("foo", status.getOwner());
     Assert.assertEquals("bar", status.getGroup());
     Assert.assertEquals(new FsPermission((short) 0770), status.getPermission());
-    AclStatus aclStatus = fs.getAclStatus(p);
+
+    AclStatus aclStatus = fs.getAclStatus(authzDir);
     Assert.assertEquals(1, aclStatus.getEntries().size());
-    Assert.assertEquals(AclEntryType.GROUP, aclStatus.getEntries().get(0)
-            .getType());
-    Assert.assertEquals("xxx", aclStatus.getEntries().get(0)
-            .getName());
-    Assert.assertEquals(FsAction.ALL, aclStatus.getEntries().get(0)
-            .getPermission());
-    Map<String, byte[]> xAttrs = fs.getXAttrs(p);
+    Assert.assertEquals(AclEntryType.GROUP,
+        aclStatus.getEntries().get(0).getType());
+    Assert.assertEquals("xxx",
+        aclStatus.getEntries().get(0).getName());
+    Assert.assertEquals(FsAction.ALL,
+        aclStatus.getEntries().get(0).getPermission());
+    Map<String, byte[]> xAttrs = fs.getXAttrs(authzDir);
     Assert.assertTrue(xAttrs.containsKey("user.test"));
     Assert.assertEquals(2, xAttrs.get("user.test").length);
   }
 
+  /**
+   * With the custom provider configured, verify file status attributes.
+   * A superuser can bypass permission check while resolving paths. So,
+   * verify file status for both superuser and non-superuser.
+   */
+  @Test
+  public void testCustomProvider() throws Exception {
+    final UserGroupInformation[] users = new UserGroupInformation[]{
+        UserGroupInformation.createUserForTesting(
+            System.getProperty("user.name"), new String[]{"supergroup"}),
+        UserGroupInformation.createUserForTesting(
+            "normaluser", new String[]{"normalusergroup"}),
+    };
+
+    for (final UserGroupInformation user : users) {
+      user.doAs((PrivilegedExceptionAction<Object>) () -> {
+        verifyFileStatus(user);
+        return null;
+      });
+    }
+  }
 }