HDFS-14743. Enhance INodeAttributeProvider/ AccessControlEnforcer Interface in HDFS to support Authorization of mkdir, rm, rmdir, copy, move etc... (apache#1829)

Reviewed-by: Xiaoyu Yao <xyao@apache.org>
(cherry picked from commit 4b95c24)

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

Change-Id: I549fbdbc7a0a45af058e8b298334b6adf554f055

Author: jojochuang

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

@@ -24,6 +24,7 @@
 import java.util.Stack;
 
 import com.google.common.base.Preconditions;
+import org.apache.hadoop.ipc.CallerContext;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.apache.hadoop.fs.FSExceptionMessages;
@@ -84,18 +85,57 @@ private String toAccessControlString(INodeAttributes inodeAttrib,
   private final Collection<String> groups;
   private final boolean isSuper;
   private final INodeAttributeProvider attributeProvider;
+  private final boolean authorizeWithContext;
+
+  private static ThreadLocal<String> operationType = new ThreadLocal<>();
 
 
   protected FSPermissionChecker(String fsOwner, String supergroup,
       UserGroupInformation callerUgi,
       INodeAttributeProvider attributeProvider) {
+    boolean useNewAuthorizationWithContextAPI;
     this.fsOwner = fsOwner;
     this.supergroup = supergroup;
     this.callerUgi = callerUgi;
     this.groups = callerUgi.getGroups();
     user = callerUgi.getShortUserName();
     isSuper = user.equals(fsOwner) || groups.contains(supergroup);
     this.attributeProvider = attributeProvider;
+
+    // If the AccessControlEnforcer supports context enrichment, call
+    // the new API. Otherwise choose the old API.
+    Class[] cArg = new Class[1];
+    cArg[0] = INodeAttributeProvider.AuthorizationContext.class;
+
+    AccessControlEnforcer ace;
+    if (attributeProvider == null) {
+      // If attribute provider is null, use FSPermissionChecker default
+      // implementation to authorize, which supports authorization with context.
+      useNewAuthorizationWithContextAPI = true;
+      LOG.info("Default authorization provider supports the new authorization" +
+          " provider API");
+    } else {
+      ace = attributeProvider.getExternalAccessControlEnforcer(this);
+      // if the runtime external authorization provider doesn't support
+      // checkPermissionWithContext(), fall back to the old API
+      // checkPermission().
+      try {
+        Class<?> clazz = ace.getClass();
+        clazz.getDeclaredMethod("checkPermissionWithContext", cArg);
+        useNewAuthorizationWithContextAPI = true;
+        LOG.info("Use the new authorization provider API");
+      } catch (NoSuchMethodException e) {
+        useNewAuthorizationWithContextAPI = false;
+        LOG.info("Fallback to the old authorization provider API because " +
+            "the expected method is not found.");
+      }
+    }
+
+    authorizeWithContext = useNewAuthorizationWithContextAPI;
+  }
+
+  public static void setOperationType(String opType) {
+    operationType.set(opType);
   }
 
   public boolean isMemberOfGroup(String group) {
@@ -190,9 +230,35 @@ void checkPermission(INodesInPath inodesInPath, boolean doCheckOwner,
     int ancestorIndex = inodes.length - 2;
 
     AccessControlEnforcer enforcer = getAccessControlEnforcer();
-    enforcer.checkPermission(fsOwner, supergroup, callerUgi, inodeAttrs, inodes,
-        components, snapshotId, path, ancestorIndex, doCheckOwner,
-        ancestorAccess, parentAccess, access, subAccess, ignoreEmptyDir);
+
+    String opType = operationType.get();
+    if (this.authorizeWithContext && opType != null) {
+      INodeAttributeProvider.AuthorizationContext.Builder builder =
+          new INodeAttributeProvider.AuthorizationContext.Builder();
+      builder.fsOwner(fsOwner).
+          supergroup(supergroup).
+          callerUgi(callerUgi).
+          inodeAttrs(inodeAttrs).
+          inodes(inodes).
+          pathByNameArr(components).
+          snapshotId(snapshotId).
+          path(path).
+          ancestorIndex(ancestorIndex).
+          doCheckOwner(doCheckOwner).
+          ancestorAccess(ancestorAccess).
+          parentAccess(parentAccess).
+          access(access).
+          subAccess(subAccess).
+          ignoreEmptyDir(ignoreEmptyDir).
+          operationName(opType).
+          callerContext(CallerContext.getCurrent());
+      enforcer.checkPermissionWithContext(builder.build());
+    } else {
+      enforcer.checkPermission(fsOwner, supergroup, callerUgi, inodeAttrs,
+          inodes, components, snapshotId, path, ancestorIndex, doCheckOwner,
+          ancestorAccess, parentAccess, access, subAccess, ignoreEmptyDir);
+    }
+
   }
 
   /**
@@ -212,17 +278,45 @@ void checkPermission(INode inode, int snapshotId, FsAction access)
     try {
       INodeAttributes[] iNodeAttr = {nodeAttributes};
       AccessControlEnforcer enforcer = getAccessControlEnforcer();
-      enforcer.checkPermission(
-          fsOwner, supergroup, callerUgi,
-          iNodeAttr, // single inode attr in the array
-          new INode[]{inode}, // single inode in the array
-          pathComponents, snapshotId,
-          null, -1, // this will skip checkTraverse() because
-          // not checking ancestor here
-          false, null, null,
-          access, // the target access to be checked against the inode
-          null, // passing null sub access avoids checking children
-          false);
+      String opType = operationType.get();
+      if (this.authorizeWithContext && opType != null) {
+        INodeAttributeProvider.AuthorizationContext.Builder builder =
+            new INodeAttributeProvider.AuthorizationContext.Builder();
+        builder.fsOwner(fsOwner)
+            .supergroup(supergroup)
+            .callerUgi(callerUgi)
+            .inodeAttrs(iNodeAttr) // single inode attr in the array
+            .inodes(new INode[] { inode }) // single inode attr in the array
+            .pathByNameArr(pathComponents)
+            .snapshotId(snapshotId)
+            .path(null)
+            .ancestorIndex(-1)     // this will skip checkTraverse()
+                                   // because not checking ancestor here
+            .doCheckOwner(false)
+            .ancestorAccess(null)
+            .parentAccess(null)
+            .access(access)        // the target access to be checked against
+                                   // the inode
+            .subAccess(null)       // passing null sub access avoids checking
+                                   // children
+            .ignoreEmptyDir(false)
+            .operationName(opType)
+            .callerContext(CallerContext.getCurrent());
+
+        enforcer.checkPermissionWithContext(builder.build());
+      } else {
+        enforcer.checkPermission(
+            fsOwner, supergroup, callerUgi,
+            iNodeAttr, // single inode attr in the array
+            new INode[]{inode}, // single inode in the array
+            pathComponents, snapshotId,
+            null, -1, // this will skip checkTraverse() because
+            // not checking ancestor here
+            false, null, null,
+            access, // the target access to be checked against the inode
+            null, // passing null sub access avoids checking children
+            false);
+      }
     } catch (AccessControlException ace) {
       throw new AccessControlException(
           toAccessControlString(nodeAttributes, inode.getFullPathName(),
@@ -273,6 +367,22 @@ public void checkPermission(String fsOwner, String supergroup,
     }
   }
 
+  @Override
+  public void checkPermissionWithContext(
+      INodeAttributeProvider.AuthorizationContext authzContext)
+      throws AccessControlException {
+    // The default authorization provider does not use the additional context
+    // parameters including operationName and callerContext.
+    this.checkPermission(authzContext.getFsOwner(),
+        authzContext.getSupergroup(), authzContext.getCallerUgi(),
+        authzContext.getInodeAttrs(), authzContext.getInodes(),
+        authzContext.getPathByNameArr(), authzContext.getSnapshotId(),
+        authzContext.getPath(), authzContext.getAncestorIndex(),
+        authzContext.isDoCheckOwner(), authzContext.getAncestorAccess(),
+        authzContext.getParentAccess(), authzContext.getAccess(),
+        authzContext.getSubAccess(), authzContext.isIgnoreEmptyDir());
+  }
+
   private INodeAttributes getINodeAttrs(byte[][] pathByNameArr, int pathIdx,
       INode inode, int snapshotId) {
     INodeAttributes inodeAttrs = inode.getSnapshotINode(snapshotId);