HADOOP-14442. Owner support for ranger-wasb integration. Contributed by Varada Hemeswari

Author: liuml07

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/NativeAzureFileSystem.java

@@ -200,7 +200,7 @@ public FolderRenamePending(Path redoFile, NativeAzureFileSystem fs)
       JsonNode oldFolderName = json.get("OldFolderName");
       JsonNode newFolderName = json.get("NewFolderName");
       if (oldFolderName == null || newFolderName == null) {
-    	  this.committed = false;
+        this.committed = false;
       } else {
         this.srcKey = oldFolderName.textValue();
         this.dstKey = newFolderName.textValue();
@@ -349,7 +349,7 @@ public String makeRenamePendingFileContents() {
 
       return contents;
     }
-    
+
     /**
      * This is an exact copy of org.codehaus.jettison.json.JSONObject.quote 
      * method.
@@ -639,7 +639,7 @@ public String getScheme() {
     return "wasb";
   }
 
-  
+
   /**
    * <p>
    * A {@link FileSystem} for reading and writing files stored on <a
@@ -1441,12 +1441,14 @@ private void performAuthCheck(Path requestingAccessForPath, WasbAuthorizationOpe
       requestingAccessForPath = requestingAccessForPath.makeQualified(getUri(), getWorkingDirectory());
       originalPath = originalPath.makeQualified(getUri(), getWorkingDirectory());
 
-      if (!this.authorizer.authorize(requestingAccessForPath.toString(), accessType.toString())) {
+      String owner = getOwnerForPath(requestingAccessForPath);
+
+      if (!this.authorizer.authorize(requestingAccessForPath.toString(), accessType.toString(), owner)) {
         throw new WasbAuthorizationException(operation
             + " operation for Path : " + originalPath.toString() + " not allowed");
       }
 
-    }
+   }
   }
 
   /**
@@ -3173,4 +3175,40 @@ private static String encodeKey(String aKey) {
     // Return to the caller with the randomized key.
     return randomizedKey;
   }
+
+  /*
+   * Helper method to retrieve owner information for a given path.
+   * The method returns empty string in case the file is not found or the metadata does not contain owner information
+  */
+  @VisibleForTesting
+  public String getOwnerForPath(Path absolutePath) throws IOException {
+    String owner = "";
+    FileMetadata meta = null;
+    String key = pathToKey(absolutePath);
+    try {
+
+      meta = store.retrieveMetadata(key);
+
+      if (meta != null) {
+        owner = meta.getPermissionStatus().getUserName();
+        LOG.debug("Retrieved '{}' as owner for path - {}", owner, absolutePath);
+      } else {
+        // meta will be null if file/folder doen not exist
+        LOG.debug("Cannot find file/folder - '{}'. Returning owner as empty string", absolutePath);
+      }
+    } catch(IOException ex) {
+
+          Throwable innerException = NativeAzureFileSystemHelper.checkForAzureStorageException(ex);
+          boolean isfileNotFoundException = innerException instanceof StorageException
+            && NativeAzureFileSystemHelper.isFileNotFoundException((StorageException) innerException);
+
+          // should not throw when the exception is related to blob/container/file/folder not found
+          if (!isfileNotFoundException) {
+            String errorMsg = "Could not retrieve owner information for path - " + absolutePath;
+            LOG.error(errorMsg);
+            throw new IOException(errorMsg, ex);
+          }
+      }
+    return owner;
+  }
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/RemoteWasbAuthorizerImpl.java

@@ -87,6 +87,12 @@ public class RemoteWasbAuthorizerImpl implements WasbAuthorizerInterface {
   private static final String DELEGATION_TOKEN_QUERY_PARAM_NAME =
       "delegation";
 
+  /**
+   *  Query parameter name for sending owner of the specific resource {@value}
+   */
+  private static final String WASB_RESOURCE_OWNER_QUERY_PARAM_NAME =
+      "wasb_resource_owner";
+
   private WasbRemoteCallHelper remoteCallHelper = null;
   private String delegationToken;
   private boolean isSecurityEnabled;
@@ -119,7 +125,7 @@ public void init(Configuration conf)
   }
 
   @Override
-  public boolean authorize(String wasbAbsolutePath, String accessType)
+  public boolean authorize(String wasbAbsolutePath, String accessType, String resourceOwner)
       throws WasbAuthorizationException, IOException {
 
       try {
@@ -140,6 +146,10 @@ public boolean authorize(String wasbAbsolutePath, String accessType)
           uriBuilder.addParameter(DELEGATION_TOKEN_QUERY_PARAM_NAME,
               delegationToken);
         }
+        if (resourceOwner != null && StringUtils.isNotEmpty(resourceOwner)) {
+          uriBuilder.addParameter(WASB_RESOURCE_OWNER_QUERY_PARAM_NAME,
+              resourceOwner);
+        }
 
         String responseBody = null;
         UserGroupInformation ugi = UserGroupInformation.getCurrentUser();

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azure/WasbAuthorizerInterface.java

@@ -43,10 +43,11 @@ public void init(Configuration conf)
 
    * @param wasbAbolutePath : Absolute WASB Path used for access.
    * @param accessType : Type of access
+   * @param owner : owner of the file/folder specified in the wasb path
    * @return : true - If access allowed false - If access is not allowed.
    * @throws WasbAuthorizationException - On authorization exceptions
    * @throws IOException - When not able to reach the authorizer
    */
-  boolean authorize(String wasbAbolutePath, String accessType)
+  boolean authorize(String wasbAbolutePath, String accessType, String owner)
       throws WasbAuthorizationException, IOException;
 }

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azure/MockWasbAuthorizerImpl.java

@@ -22,7 +22,9 @@
 import java.util.Map;
 import java.util.regex.Pattern;
 
+import org.apache.hadoop.security.UserGroupInformation;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.util.StringUtils;
 import org.apache.hadoop.fs.Path;
 
 /**
@@ -32,8 +34,9 @@
 public class MockWasbAuthorizerImpl implements WasbAuthorizerInterface {
 
   private Map<AuthorizationComponent, Boolean> authRules;
+  private boolean performOwnerMatch;
 
-  // The full qualified URL to the root directory
+ // The full qualified URL to the root directory
   private String qualifiedPrefixUrl;
 
   public MockWasbAuthorizerImpl(NativeAzureFileSystem fs) {
@@ -43,34 +46,63 @@ public MockWasbAuthorizerImpl(NativeAzureFileSystem fs) {
 
   @Override
   public void init(Configuration conf) {
+    init(conf, false);
+  }
+
+  /*
+  authorization matches owner with currentUserShortName while evaluating auth rules
+  if currentUserShortName is set to a string that is not empty
+  */
+  public void init(Configuration conf, boolean matchOwner) {
     authRules = new HashMap<AuthorizationComponent, Boolean>();
+    this.performOwnerMatch = matchOwner;
   }
 
   public void addAuthRule(String wasbAbsolutePath,
       String accessType, boolean access) {
-
-    wasbAbsolutePath = qualifiedPrefixUrl + wasbAbsolutePath;
-
-    AuthorizationComponent component = wasbAbsolutePath.endsWith("*")
+        wasbAbsolutePath = qualifiedPrefixUrl + wasbAbsolutePath;
+        AuthorizationComponent component = wasbAbsolutePath.endsWith("*")
         ? new AuthorizationComponent("^" + wasbAbsolutePath.replace("*", ".*"), accessType)
         : new AuthorizationComponent(wasbAbsolutePath, accessType);
 
     this.authRules.put(component, access);
   }
 
   @Override
-  public boolean authorize(String wasbAbsolutePath, String accessType)
+  public boolean authorize(String wasbAbsolutePath, String accessType, String owner)
       throws WasbAuthorizationException {
 
     if (wasbAbsolutePath.endsWith(NativeAzureFileSystem.FolderRenamePending.SUFFIX)) {
       return true;
     }
 
+    String currentUserShortName = "";
+    if (this.performOwnerMatch) {
+      try {
+        UserGroupInformation ugi = UserGroupInformation.getCurrentUser();
+        currentUserShortName = ugi.getShortUserName();
+      } catch (Exception e) {
+        //no op
+      }
+    }
+
+    // In case of root("/"), owner match does not happen because owner is returned as empty string.
+    // we try to force owner match just for purpose of tests to make sure all operations work seemlessly with owner.
+    if (this.performOwnerMatch
+      && StringUtils.equalsIgnoreCase(wasbAbsolutePath, qualifiedPrefixUrl + "/")) {
+      owner = currentUserShortName;
+    }
+
+    boolean shouldEvaluateOwnerAccess = owner != null && !owner.isEmpty()
+      && this.performOwnerMatch;
+
+    boolean isOwnerMatch =  StringUtils.equalsIgnoreCase(currentUserShortName, owner);
+
     AuthorizationComponent component =
         new AuthorizationComponent(wasbAbsolutePath, accessType);
 
     if (authRules.containsKey(component)) {
-      return authRules.get(component);
+      return shouldEvaluateOwnerAccess ? isOwnerMatch && authRules.get(component) : authRules.get(component);
     } else {
       // Regex-pattern match if we don't have a straight match
       for (Map.Entry<AuthorizationComponent, Boolean> entry : authRules.entrySet()) {
@@ -79,12 +111,16 @@ public boolean authorize(String wasbAbsolutePath, String accessType)
         String keyAccess = key.getAccessType();
 
         if (keyPath.endsWith("*") && Pattern.matches(keyPath, wasbAbsolutePath) && keyAccess.equals(accessType)) {
-          return entry.getValue();
+          return shouldEvaluateOwnerAccess ? isOwnerMatch && entry.getValue() : entry.getValue();
         }
       }
       return false;
     }
   }
+
+  public void deleteAllAuthRules() {
+    authRules.clear();
+  }
 }
 
 class AuthorizationComponent {