HADOOP-15723. ABFS: Ranger Support.

Contributed by Yuan Gao.

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

@@ -20,6 +20,7 @@ package org.apache.hadoop.fs.azurebfs;
 
 import java.io.IOException;
 import java.lang.reflect.Field;
+import java.lang.reflect.InvocationTargetException;
 import java.util.Map;
 
 import com.google.common.annotations.VisibleForTesting;
@@ -42,6 +43,8 @@ import org.apache.hadoop.fs.azurebfs.diagnostics.BooleanConfigurationBasicValida
 import org.apache.hadoop.fs.azurebfs.diagnostics.IntegerConfigurationBasicValidator;
 import org.apache.hadoop.fs.azurebfs.diagnostics.LongConfigurationBasicValidator;
 import org.apache.hadoop.fs.azurebfs.diagnostics.StringConfigurationBasicValidator;
+import org.apache.hadoop.fs.azurebfs.extensions.AbfsAuthorizationException;
+import org.apache.hadoop.fs.azurebfs.extensions.AbfsAuthorizer;
 import org.apache.hadoop.fs.azurebfs.extensions.CustomTokenProviderAdaptee;
 import org.apache.hadoop.fs.azurebfs.oauth2.AccessTokenProvider;
 import org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider;
@@ -155,6 +158,10 @@ public class AbfsConfiguration{
       DefaultValue = DEFAULT_ENABLE_DELEGATION_TOKEN)
   private boolean enableDelegationToken;
 
+  @StringConfigurationValidatorAnnotation(ConfigurationKey = ABFS_EXTERNAL_AUTHORIZATION_CLASS,
+      DefaultValue = "")
+  private String abfsExternalAuthorizationClass;
+
   private Map<String, String> storageAccountKeys;
 
   public AbfsConfiguration(final Configuration rawConfig, String accountName)
@@ -490,6 +497,35 @@ public class AbfsConfiguration{
     }
   }
 
+  public String getAbfsExternalAuthorizationClass() {
+    return this.abfsExternalAuthorizationClass;
+  }
+
+  public AbfsAuthorizer getAbfsAuthorizer() throws IOException {
+    String authClassName = getAbfsExternalAuthorizationClass();
+    AbfsAuthorizer authorizer = null;
+
+    try {
+      if (authClassName != null && !authClassName.isEmpty()) {
+        @SuppressWarnings("unchecked")
+        Class<AbfsAuthorizer> authClass = (Class<AbfsAuthorizer>) rawConfig.getClassByName(authClassName);
+        authorizer = authClass.getConstructor(new Class[] {Configuration.class}).newInstance(rawConfig);
+        authorizer.init();
+      }
+    } catch (
+        IllegalAccessException
+        | InstantiationException
+        | ClassNotFoundException
+        | IllegalArgumentException
+        | InvocationTargetException
+        | NoSuchMethodException
+        | SecurityException
+        | AbfsAuthorizationException e) {
+      throw new IOException(e);
+    }
+    return authorizer;
+  }
+
   void validateStorageAccountKeys() throws InvalidConfigurationValueException {
     Base64StringConfigurationBasicValidator validator = new Base64StringConfigurationBasicValidator(
         FS_AZURE_ACCOUNT_KEY_PROPERTY_NAME, "", true);

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

@@ -28,6 +28,7 @@ import java.net.URI;
 import java.net.URISyntaxException;
 import java.util.List;
 import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.EnumSet;
 import java.util.concurrent.Callable;
 import java.util.concurrent.ExecutionException;
@@ -62,9 +63,12 @@ import org.apache.hadoop.fs.azurebfs.contracts.exceptions.FileSystemOperationUnh
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidUriAuthorityException;
 import org.apache.hadoop.fs.azurebfs.contracts.exceptions.InvalidUriException;
 import org.apache.hadoop.fs.azurebfs.contracts.services.AzureServiceErrorCode;
+import org.apache.hadoop.fs.azurebfs.extensions.AbfsAuthorizationException;
+import org.apache.hadoop.fs.azurebfs.extensions.AbfsAuthorizer;
 import org.apache.hadoop.fs.azurebfs.security.AbfsDelegationTokenManager;
 import org.apache.hadoop.fs.permission.AclEntry;
 import org.apache.hadoop.fs.permission.AclStatus;
+import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.fs.permission.FsPermission;
 import org.apache.hadoop.security.token.Token;
 import org.apache.hadoop.security.UserGroupInformation;
@@ -87,6 +91,7 @@ public class AzureBlobFileSystem extends FileSystem {
 
   private boolean delegationTokenEnabled = false;
   private AbfsDelegationTokenManager delegationTokenManager;
+  private AbfsAuthorizer authorizer;
 
   @Override
   public void initialize(URI uri, Configuration configuration)
@@ -132,6 +137,10 @@ public class AzureBlobFileSystem extends FileSystem {
     }
 
     AbfsClientThrottlingIntercept.initializeSingleton(abfsConfiguration.isAutoThrottlingEnabled());
+
+    // Initialize ABFS authorizer
+    //
+    this.authorizer = abfsConfiguration.getAbfsAuthorizer();
   }
 
   @Override
@@ -158,8 +167,11 @@ public class AzureBlobFileSystem extends FileSystem {
   public FSDataInputStream open(final Path path, final int bufferSize) throws IOException {
     LOG.debug("AzureBlobFileSystem.open path: {} bufferSize: {}", path, bufferSize);
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.READ, qualifiedPath);
+
     try {
-      InputStream inputStream = abfsStore.openFileForRead(makeQualified(path), statistics);
+      InputStream inputStream = abfsStore.openFileForRead(qualifiedPath, statistics);
       return new FSDataInputStream(inputStream);
     } catch(AzureBlobFileSystemException ex) {
       checkException(path, ex);
@@ -176,8 +188,11 @@ public class AzureBlobFileSystem extends FileSystem {
         overwrite,
         blockSize);
 
+    Path qualifiedPath = makeQualified(f);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      OutputStream outputStream = abfsStore.createFile(makeQualified(f), overwrite,
+      OutputStream outputStream = abfsStore.createFile(qualifiedPath, overwrite,
           permission == null ? FsPermission.getFileDefault() : permission, FsPermission.getUMask(getConf()));
       return new FSDataOutputStream(outputStream, statistics);
     } catch(AzureBlobFileSystemException ex) {
@@ -236,8 +251,11 @@ public class AzureBlobFileSystem extends FileSystem {
         f.toString(),
         bufferSize);
 
+    Path qualifiedPath = makeQualified(f);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      OutputStream outputStream = abfsStore.openFileForWrite(makeQualified(f), false);
+      OutputStream outputStream = abfsStore.openFileForWrite(qualifiedPath, false);
       return new FSDataOutputStream(outputStream, statistics);
     } catch(AzureBlobFileSystemException ex) {
       checkException(f, ex);
@@ -267,7 +285,11 @@ public class AzureBlobFileSystem extends FileSystem {
         adjustedDst = new Path(dst, sourceFileName);
       }
 
-      abfsStore.rename(makeQualified(src), makeQualified(adjustedDst));
+      Path qualifiedSrcPath = makeQualified(src);
+      Path qualifiedDstPath = makeQualified(adjustedDst);
+      performAbfsAuthCheck(FsAction.READ_WRITE, qualifiedSrcPath, qualifiedDstPath);
+
+      abfsStore.rename(qualifiedSrcPath, qualifiedDstPath);
       return true;
     } catch(AzureBlobFileSystemException ex) {
       checkException(
@@ -289,6 +311,9 @@ public class AzureBlobFileSystem extends FileSystem {
     LOG.debug(
         "AzureBlobFileSystem.delete path: {} recursive: {}", f.toString(), recursive);
 
+    Path qualifiedPath = makeQualified(f);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     if (f.isRoot()) {
       if (!recursive) {
         return false;
@@ -298,7 +323,7 @@ public class AzureBlobFileSystem extends FileSystem {
     }
 
     try {
-      abfsStore.delete(makeQualified(f), recursive);
+      abfsStore.delete(qualifiedPath, recursive);
       return true;
     } catch (AzureBlobFileSystemException ex) {
       checkException(f, ex, AzureServiceErrorCode.PATH_NOT_FOUND);
@@ -312,8 +337,11 @@ public class AzureBlobFileSystem extends FileSystem {
     LOG.debug(
         "AzureBlobFileSystem.listStatus path: {}", f.toString());
 
+    Path qualifiedPath = makeQualified(f);
+    performAbfsAuthCheck(FsAction.READ, qualifiedPath);
+
     try {
-      FileStatus[] result = abfsStore.listStatus(makeQualified(f));
+      FileStatus[] result = abfsStore.listStatus(qualifiedPath);
       return result;
     } catch (AzureBlobFileSystemException ex) {
       checkException(f, ex);
@@ -332,8 +360,11 @@ public class AzureBlobFileSystem extends FileSystem {
       return true;
     }
 
+    Path qualifiedPath = makeQualified(f);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.createDirectory(makeQualified(f), permission == null ? FsPermission.getDirDefault() : permission,
+      abfsStore.createDirectory(qualifiedPath, permission == null ? FsPermission.getDirDefault() : permission,
           FsPermission.getUMask(getConf()));
       return true;
     } catch (AzureBlobFileSystemException ex) {
@@ -357,8 +388,11 @@ public class AzureBlobFileSystem extends FileSystem {
   public FileStatus getFileStatus(final Path f) throws IOException {
     LOG.debug("AzureBlobFileSystem.getFileStatus path: {}", f);
 
+    Path qualifiedPath = makeQualified(f);
+    performAbfsAuthCheck(FsAction.READ, qualifiedPath);
+
     try {
-      return abfsStore.getFileStatus(makeQualified(f));
+      return abfsStore.getFileStatus(qualifiedPath);
     } catch(AzureBlobFileSystemException ex) {
       checkException(f, ex);
       return null;
@@ -528,8 +562,11 @@ public class AzureBlobFileSystem extends FileSystem {
       throw new IllegalArgumentException("A valid owner or group must be specified.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.setOwner(makeQualified(path),
+      abfsStore.setOwner(qualifiedPath,
               owner,
               group);
     } catch (AzureBlobFileSystemException ex) {
@@ -556,8 +593,11 @@ public class AzureBlobFileSystem extends FileSystem {
       throw new IllegalArgumentException("The permission can't be null");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.setPermission(makeQualified(path),
+      abfsStore.setPermission(qualifiedPath,
               permission);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
@@ -589,8 +629,11 @@ public class AzureBlobFileSystem extends FileSystem {
       throw new IllegalArgumentException("The value of the aclSpec parameter is invalid.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.modifyAclEntries(makeQualified(path),
+      abfsStore.modifyAclEntries(qualifiedPath,
               aclSpec);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
@@ -620,8 +663,11 @@ public class AzureBlobFileSystem extends FileSystem {
       throw new IllegalArgumentException("The aclSpec argument is invalid.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.removeAclEntries(makeQualified(path), aclSpec);
+      abfsStore.removeAclEntries(qualifiedPath, aclSpec);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -643,8 +689,11 @@ public class AzureBlobFileSystem extends FileSystem {
           + "hierarchical namespace enabled.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.removeDefaultAcl(makeQualified(path));
+      abfsStore.removeDefaultAcl(qualifiedPath);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -668,8 +717,11 @@ public class AzureBlobFileSystem extends FileSystem {
           + "hierarchical namespace enabled.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.removeAcl(makeQualified(path));
+      abfsStore.removeAcl(qualifiedPath);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -700,8 +752,11 @@ public class AzureBlobFileSystem extends FileSystem {
       throw new IllegalArgumentException("The aclSpec argument is invalid.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.WRITE, qualifiedPath);
+
     try {
-      abfsStore.setAcl(makeQualified(path), aclSpec);
+      abfsStore.setAcl(qualifiedPath, aclSpec);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
     }
@@ -724,8 +779,11 @@ public class AzureBlobFileSystem extends FileSystem {
           + "hierarchical namespace enabled.");
     }
 
+    Path qualifiedPath = makeQualified(path);
+    performAbfsAuthCheck(FsAction.READ, qualifiedPath);
+
     try {
-      return abfsStore.getAclStatus(makeQualified(path));
+      return abfsStore.getAclStatus(qualifiedPath);
     } catch (AzureBlobFileSystemException ex) {
       checkException(path, ex);
       return null;
@@ -950,4 +1008,30 @@ public class AzureBlobFileSystem extends FileSystem {
   boolean getIsNamespaceEnabeld() throws AzureBlobFileSystemException {
     return abfsStore.getIsNamespaceEnabled();
   }
+
+  /**
+   * Use ABFS authorizer to check if user is authorized to perform specific
+   * {@link FsAction} on specified {@link Path}s.
+   *
+   * @param action The {@link FsAction} being requested on the provided {@link Path}s.
+   * @param paths The absolute paths of the storage being accessed.
+   * @throws AbfsAuthorizationException on authorization failure.
+   * @throws IOException network problems or similar.
+   * @throws IllegalArgumentException if the required parameters are not provided.
+   */
+  private void performAbfsAuthCheck(FsAction action, Path... paths)
+      throws AbfsAuthorizationException, IOException {
+    if (authorizer == null) {
+      LOG.debug("ABFS authorizer is not initialized. No authorization check will be performed.");
+    } else {
+      Preconditions.checkArgument(paths.length > 0, "no paths supplied for authorization check");
+
+      LOG.debug("Auth check for action: {} on paths: {}", action.toString(), Arrays.toString(paths));
+      if (!authorizer.isAuthorized(action, paths)) {
+        throw new AbfsAuthorizationException(
+            "User is not authorized for action " + action.toString()
+            + " on paths: " + Arrays.toString(paths));
+      }
+    }
+  }
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/constants/ConfigurationKeys.java

@@ -85,5 +85,7 @@ public final class ConfigurationKeys {
   public static final String FS_AZURE_ENABLE_DELEGATION_TOKEN = "fs.azure.enable.delegation.token";
   public static final String FS_AZURE_DELEGATION_TOKEN_PROVIDER_TYPE = "fs.azure.delegation.token.provider.type";
 
+  public static final String ABFS_EXTERNAL_AUTHORIZATION_CLASS = "abfs.external.authorization.class";
+
   private ConfigurationKeys() {}
 }

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/extensions/AbfsAuthorizationException.java

@@ -0,0 +1,41 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.extensions;
+
+import java.io.IOException;
+
+/**
+ * Exception raised on ABFS Authorization failures.
+ */
+public class AbfsAuthorizationException extends IOException {
+
+  private static final long serialVersionUID = 1L;
+
+  public AbfsAuthorizationException(String message, Exception e) {
+    super(message, e);
+  }
+
+  public AbfsAuthorizationException(String message) {
+    super(message);
+  }
+
+  public AbfsAuthorizationException(Throwable e) {
+    super(e);
+  }
+}

hadoop-tools/hadoop-azure/src/main/java/org/apache/hadoop/fs/azurebfs/extensions/AbfsAuthorizer.java

@@ -0,0 +1,57 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.extensions;
+
+import java.io.IOException;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsAction;
+
+/**
+ * Interface to support authorization in Azure Blob File System.
+ */
+@InterfaceAudience.LimitedPrivate("authorization-subsystems")
+@InterfaceStability.Unstable
+public interface AbfsAuthorizer {
+
+  /**
+   * Initialize authorizer for Azure Blob File System.
+   *
+   * @throws AbfsAuthorizationException if unable to initialize the authorizer.
+   * @throws IOException network problems or similar.
+   * @throws IllegalArgumentException if the required parameters are not provided.
+   */
+  void init() throws AbfsAuthorizationException, IOException;
+
+  /**
+   * Checks if the provided {@link FsAction} is allowed on the provided {@link Path}s.
+   *
+   * @param action the {@link FsAction} being requested on the provided {@link Path}s.
+   * @param absolutePaths The absolute paths of the storage being accessed.
+   * @return true if authorized, otherwise false.
+   * @throws AbfsAuthorizationException on authorization failure.
+   * @throws IOException network problems or similar.
+   * @throws IllegalArgumentException if the required parameters are not provided.
+   */
+  boolean isAuthorized(FsAction action, Path... absolutePaths)
+      throws AbfsAuthorizationException, IOException;
+
+}

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

@@ -257,12 +257,12 @@ public abstract class AbstractAbfsIntegrationTest extends
     return abfsConfig.getRawConfiguration();
   }
 
-  protected boolean isIPAddress() {
+  public AuthType getAuthType() {
-    return isIPAddress;
+    return this.authType;
   }
 
-  protected AuthType getAuthType() {
+  protected boolean isIPAddress() {
-    return this.authType;
+    return isIPAddress;
   }
 
   /**

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

@@ -0,0 +1,346 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs;
+
+import java.util.Arrays;
+import java.util.List;
+
+import org.junit.Test;
+
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys;
+import org.apache.hadoop.fs.azurebfs.extensions.AbfsAuthorizationException;
+import org.apache.hadoop.fs.permission.AclEntry;
+import org.apache.hadoop.fs.permission.FsAction;
+import org.apache.hadoop.fs.permission.FsPermission;
+
+import static org.apache.hadoop.fs.azurebfs.extensions.MockAbfsAuthorizer.*;
+import static org.apache.hadoop.fs.azurebfs.utils.AclTestHelpers.aclEntry;
+import static org.apache.hadoop.fs.permission.AclEntryScope.ACCESS;
+import static org.apache.hadoop.fs.permission.AclEntryType.GROUP;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+
+/**
+ * Test Perform Authorization Check operation
+ */
+public class ITestAzureBlobFileSystemAuthorization extends AbstractAbfsIntegrationTest {
+
+  private static final Path TEST_READ_ONLY_FILE_PATH_0 = new Path(TEST_READ_ONLY_FILE_0);
+  private static final Path TEST_READ_ONLY_FOLDER_PATH = new Path(TEST_READ_ONLY_FOLDER);
+  private static final Path TEST_WRITE_ONLY_FILE_PATH_0 = new Path(TEST_WRITE_ONLY_FILE_0);
+  private static final Path TEST_WRITE_ONLY_FILE_PATH_1 = new Path(TEST_WRITE_ONLY_FILE_1);
+  private static final Path TEST_READ_WRITE_FILE_PATH_0 = new Path(TEST_READ_WRITE_FILE_0);
+  private static final Path TEST_READ_WRITE_FILE_PATH_1 = new Path(TEST_READ_WRITE_FILE_1);
+  private static final Path TEST_WRITE_ONLY_FOLDER_PATH = new Path(TEST_WRITE_ONLY_FOLDER);
+  private static final Path TEST_WRITE_THEN_READ_ONLY_PATH = new Path(TEST_WRITE_THEN_READ_ONLY);
+  private static final String TEST_AUTHZ_CLASS = "org.apache.hadoop.fs.azurebfs.extensions.MockAbfsAuthorizer";
+
+  public ITestAzureBlobFileSystemAuthorization() throws Exception {
+  }
+
+  @Override
+  public void setup() throws Exception {
+    this.getConfiguration().set(ConfigurationKeys.ABFS_EXTERNAL_AUTHORIZATION_CLASS, TEST_AUTHZ_CLASS);
+    super.setup();
+  }
+
+  @Test
+  public void testOpenFileWithInvalidPath() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    intercept(IllegalArgumentException.class,
+        ()-> {
+          fs.open(new Path("")).close();
+    });
+  }
+
+  @Test
+  public void testOpenFileAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    fs.open(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+  }
+
+  @Test
+  public void testOpenFileUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.open(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    });
+  }
+
+  @Test
+  public void testCreateFileAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+  }
+
+  @Test
+  public void testCreateFileUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.create(TEST_READ_ONLY_FILE_PATH_0).close();
+    });
+  }
+
+  @Test
+  public void testAppendFileAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    fs.append(TEST_WRITE_ONLY_FILE_PATH_0).close();
+  }
+
+  @Test
+  public void testAppendFileUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.append(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    });
+  }
+
+  @Test
+  public void testRenameAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.rename(TEST_READ_WRITE_FILE_PATH_0, TEST_READ_WRITE_FILE_PATH_1);
+  }
+
+  @Test
+  public void testRenameUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.rename(TEST_WRITE_ONLY_FILE_PATH_0, TEST_WRITE_ONLY_FILE_PATH_1);
+    });
+  }
+
+  @Test
+  public void testDeleteFileAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    fs.delete(TEST_WRITE_ONLY_FILE_PATH_0, false);
+  }
+
+  @Test
+  public void testDeleteFileUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = this.getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.delete(TEST_WRITE_THEN_READ_ONLY_PATH, false);
+    });
+  }
+
+  @Test
+  public void testListStatusAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    fs.listStatus(TEST_WRITE_THEN_READ_ONLY_PATH);
+  }
+
+  @Test
+  public void testListStatusUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.listStatus(TEST_WRITE_ONLY_FILE_PATH_0);
+    });
+  }
+
+  @Test
+  public void testMkDirsAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.mkdirs(TEST_WRITE_ONLY_FOLDER_PATH, new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE));
+  }
+
+  @Test
+  public void testMkDirsUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.mkdirs(TEST_READ_ONLY_FOLDER_PATH, new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE));
+    });
+  }
+
+  @Test
+  public void testGetFileStatusAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    fs.getFileStatus(TEST_WRITE_THEN_READ_ONLY_PATH);
+  }
+
+  @Test
+  public void testGetFileStatusUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.getFileStatus(TEST_WRITE_ONLY_FILE_PATH_0);
+    });
+  }
+
+  @Test
+  public void testSetOwnerAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    fs.setOwner(TEST_WRITE_ONLY_FILE_PATH_0, "testUser", "testGroup");
+  }
+
+  @Test
+  public void testSetOwnerUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.setOwner(TEST_WRITE_THEN_READ_ONLY_PATH, "testUser", "testGroup");
+    });
+  }
+
+  @Test
+  public void testSetPermissionAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    fs.setPermission(TEST_WRITE_ONLY_FILE_PATH_0, new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE));
+  }
+
+  @Test
+  public void testSetPermissionUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.setPermission(TEST_WRITE_THEN_READ_ONLY_PATH, new FsPermission(FsAction.ALL, FsAction.NONE, FsAction.NONE));
+    });
+  }
+
+  @Test
+  public void testModifyAclEntriesAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    fs.modifyAclEntries(TEST_WRITE_ONLY_FILE_PATH_0, aclSpec);
+  }
+
+  @Test
+  public void testModifyAclEntriesUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.modifyAclEntries(TEST_WRITE_THEN_READ_ONLY_PATH, aclSpec);
+    });
+  }
+
+  @Test
+  public void testRemoveAclEntriesAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    //fs.modifyAclEntries(TEST_WRITE_ONLY_FILE_PATH_0, aclSpec);
+    fs.removeAclEntries(TEST_WRITE_ONLY_FILE_PATH_0, aclSpec);
+  }
+
+  @Test
+  public void testRemoveAclEntriesUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.removeAclEntries(TEST_WRITE_THEN_READ_ONLY_PATH, aclSpec);
+    });
+  }
+
+  @Test
+  public void testRemoveDefaultAclAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    fs.removeDefaultAcl(TEST_WRITE_ONLY_FILE_PATH_0);
+  }
+
+  @Test
+  public void testRemoveDefaultAclUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.removeDefaultAcl(TEST_WRITE_THEN_READ_ONLY_PATH);
+    });
+  }
+
+  @Test
+  public void testRemoveAclAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    fs.removeAcl(TEST_WRITE_ONLY_FILE_PATH_0);
+  }
+
+  @Test
+  public void testRemoveAclUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.removeAcl(TEST_WRITE_THEN_READ_ONLY_PATH);
+    });
+  }
+
+  @Test
+  public void testSetAclAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    fs.setAcl(TEST_WRITE_ONLY_FILE_PATH_0, aclSpec);
+  }
+
+  @Test
+  public void testSetAclUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.setAcl(TEST_WRITE_THEN_READ_ONLY_PATH, aclSpec);
+    });
+  }
+
+  @Test
+  public void testGetAclStatusAuthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_THEN_READ_ONLY_PATH).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    fs.getAclStatus(TEST_WRITE_THEN_READ_ONLY_PATH);
+  }
+
+  @Test
+  public void testGetAclStatusUnauthorized() throws Exception {
+    final AzureBlobFileSystem fs = getFileSystem();
+    fs.create(TEST_WRITE_ONLY_FILE_PATH_0).close();
+    List<AclEntry> aclSpec = Arrays.asList(aclEntry(ACCESS, GROUP, "bar", FsAction.ALL));
+    intercept(AbfsAuthorizationException.class,
+        ()-> {
+          fs.getAclStatus(TEST_WRITE_ONLY_FILE_PATH_0);
+    });
+  }
+}

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/contract/ITestAbfsFileSystemContractDistCp.java

@@ -19,7 +19,9 @@
 package org.apache.hadoop.fs.azurebfs.contract;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.azurebfs.services.AuthType;
 import org.apache.hadoop.tools.contract.AbstractContractDistCpTest;
+import org.junit.Assume;
 
 /**
  * Contract test for distCp operation.
@@ -29,6 +31,7 @@ public class ITestAbfsFileSystemContractDistCp extends AbstractContractDistCpTes
 
   public ITestAbfsFileSystemContractDistCp() throws Exception {
     binding = new ABFSContractTestBinding();
+    Assume.assumeTrue(binding.getAuthType() != AuthType.OAuth);
   }
 
   @Override

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/extensions/MockAbfsAuthorizer.java

@@ -0,0 +1,87 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.extensions;
+
+import java.io.IOException;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.fs.permission.FsAction;
+
+/**
+ * A mock Azure Blob File System Authorization Implementation
+ */
+public class MockAbfsAuthorizer implements AbfsAuthorizer {
+
+  public static final String TEST_READ_ONLY_FILE_0 = "readOnlyFile0";
+  public static final String TEST_READ_ONLY_FILE_1 = "readOnlyFile1";
+  public static final String TEST_READ_ONLY_FOLDER = "readOnlyFolder";
+  public static final String TEST_WRITE_ONLY_FILE_0 = "writeOnlyFile0";
+  public static final String TEST_WRITE_ONLY_FILE_1 = "writeOnlyFile1";
+  public static final String TEST_WRITE_ONLY_FOLDER = "writeOnlyFolder";
+  public static final String TEST_READ_WRITE_FILE_0 = "readWriteFile0";
+  public static final String TEST_READ_WRITE_FILE_1 = "readWriteFile1";
+  public static final String TEST_WRITE_THEN_READ_ONLY = "writeThenReadOnlyFile";
+  private Configuration conf;
+  private Set<Path> readOnlyPaths = new HashSet<Path>();
+  private Set<Path> writeOnlyPaths = new HashSet<Path>();
+  private Set<Path> readWritePaths = new HashSet<Path>();
+  private int writeThenReadOnly = 0;
+  public MockAbfsAuthorizer(Configuration conf) {
+    this.conf = conf;
+  }
+
+  @Override
+  public void init() throws AbfsAuthorizationException, IOException {
+    readOnlyPaths.add(new Path(TEST_READ_ONLY_FILE_0));
+    readOnlyPaths.add(new Path(TEST_READ_ONLY_FILE_1));
+    readOnlyPaths.add(new Path(TEST_READ_ONLY_FOLDER));
+    writeOnlyPaths.add(new Path(TEST_WRITE_ONLY_FILE_0));
+    writeOnlyPaths.add(new Path(TEST_WRITE_ONLY_FILE_1));
+    writeOnlyPaths.add(new Path(TEST_WRITE_ONLY_FOLDER));
+    readWritePaths.add(new Path(TEST_READ_WRITE_FILE_0));
+    readWritePaths.add(new Path(TEST_READ_WRITE_FILE_1));
+  }
+
+  @Override
+  public boolean isAuthorized(FsAction action, Path... absolutePaths) throws AbfsAuthorizationException, IOException {
+    Set<Path> paths = new HashSet<Path>();
+    for (Path path : absolutePaths) {
+      paths.add(new Path(path.getName()));
+    }
+
+    if (action.equals(FsAction.READ) && Stream.concat(readOnlyPaths.stream(), readWritePaths.stream()).collect(Collectors.toSet()).containsAll(paths)) {
+      return true;
+    } else if (action.equals(FsAction.READ) && paths.contains(new Path(TEST_WRITE_THEN_READ_ONLY)) && writeThenReadOnly == 1) {
+      return true;
+    } else if (action.equals(FsAction.WRITE)
+        && Stream.concat(writeOnlyPaths.stream(), readWritePaths.stream()).collect(Collectors.toSet()).containsAll(paths)) {
+      return true;
+    } else if (action.equals(FsAction.WRITE) && paths.contains(new Path(TEST_WRITE_THEN_READ_ONLY)) && writeThenReadOnly == 0) {
+      writeThenReadOnly = 1;
+      return true;
+    } else {
+      return action.equals(FsAction.READ_WRITE) && readWritePaths.containsAll(paths);
+    }
+  }
+}

hadoop-tools/hadoop-azure/src/test/java/org/apache/hadoop/fs/azurebfs/extensions/package-info.java

@@ -0,0 +1,22 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+@InterfaceAudience.Private
+@InterfaceStability.Evolving
+package org.apache.hadoop.fs.azurebfs.extensions;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;