HADOOP-17183. ABFS: Enabling checkaccess on ABFS (#2331)

Contributed by Bilahari TH

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

@@ -85,7 +85,7 @@ public final class FileSystemConfigurations {
   public static final boolean DEFAULT_ENABLE_HTTPS = true;
 
   public static final boolean DEFAULT_USE_UPN = false;
-  public static final boolean DEFAULT_ENABLE_CHECK_ACCESS = false;
+  public static final boolean DEFAULT_ENABLE_CHECK_ACCESS = true;
   public static final boolean DEFAULT_ABFS_LATENCY_TRACK = false;
   public static final long DEFAULT_SAS_TOKEN_RENEW_PERIOD_FOR_STREAMS_IN_SECONDS = 120;
 

hadoop-tools/hadoop-azure/src/site/markdown/testing_azure.md

@@ -868,6 +868,63 @@ hierarchical namespace enabled and set the following configuration settings:
 
 ```
 
+To run CheckAccess test cases you must register an app with no RBAC and set
+the following configurations.
+```xml
+<!--===========================   FOR CheckAccess =========================-->
+<!-- To run ABFS CheckAccess SAS tests, you must register an app, with no role
+ assignments, and set the configuration discussed below:
+
+    1) Register a new app with no RBAC
+    2) As part of the test configs you need to provide the guid for the above
+created app. Please follow the below steps to fetch the guid.
+      a) Get an access token with the above created app. Please refer the
+ following documentation for the same. https://docs.microsoft
+.com/en-us/azure/active-directory/develop/v2-oauth2-client-creds-grant-flow#get-a-token
+      b) Decode the token fetched with the above step. You may use https
+://jwt.ms/ to decode the token
+      d) The oid field in the decoded string is the guid.
+    3) Set the following configurations:
+-->
+
+  <property>
+    <name>fs.azure.enable.check.access</name>
+    <value>true</value>
+    <description>By default the check access will be on. Checkaccess can
+    be turned off by changing this flag to false.</description>
+  </property>
+  <property>
+    <name>fs.azure.account.test.oauth2.client.id</name>
+    <value>{client id}</value>
+    <description>The client id(app id) for the app created on step 1
+    </description>
+  </property>
+  <property>
+    <name>fs.azure.account.test.oauth2.client.secret</name>
+    <value>{client secret}</value>
+    <description>
+The client secret(application's secret) for the app created on step 1
+    </description>
+  </property>
+  <property>
+    <name>fs.azure.check.access.testuser.guid</name>
+    <value>{guid}</value>
+    <description>The guid fetched on step 2</description>
+  </property>
+  <property>
+    <name>fs.azure.account.oauth2.client.endpoint.{account name}.dfs.core
+.windows.net</name>
+    <value>https://login.microsoftonline.com/{TENANTID}/oauth2/token</value>
+    <description>
+Token end point. This can be found through Azure portal. As part of CheckAccess
+test cases. The access will be tested for an FS instance created with the
+above mentioned client credentials. So this configuration is necessary to
+create the test FS instance.
+    </description>
+  </property>
+
+```
+
 If running tests against an endpoint that uses the URL format
 http[s]://[ip]:[port]/[account]/[filesystem] instead of
 http[s]://[account][domain-suffix]/[filesystem], please use the following:

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

@@ -17,16 +17,20 @@
  */
 package org.apache.hadoop.fs.azurebfs;
 
-import com.google.common.collect.Lists;
-
 import java.io.FileNotFoundException;
 import java.io.IOException;
+import java.lang.reflect.Field;
 import java.util.List;
 
+import com.google.common.base.Preconditions;
+import com.google.common.collect.Lists;
 import org.junit.Assume;
 import org.junit.Test;
+import org.mockito.Mockito;
 
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.azurebfs.oauth2.ClientCredsTokenProvider;
+import org.apache.hadoop.fs.azurebfs.services.AuthType;
 import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.azurebfs.utils.AclTestHelpers;
@@ -37,6 +41,9 @@ import org.apache.hadoop.fs.permission.FsAction;
 import org.apache.hadoop.security.AccessControlException;
 
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_OAUTH_CLIENT_ENDPOINT;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ACCOUNT_TOKEN_PROVIDER_TYPE_PROPERTY_NAME;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_ENABLE_CHECK_ACCESS;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_ID;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET;
@@ -44,9 +51,15 @@ import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_A
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_BLOB_FS_CLIENT_ID;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_BLOB_FS_CLIENT_SECRET;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
 
 /**
  * Test cases for AzureBlobFileSystem.access()
+ *
+ * Some of the tests in this class require additional configs set in the test
+ * config file.
+ * Refer testing_azure.md for how to set the configs.
+ *
  */
 public class ITestAzureBlobFileSystemCheckAccess
     extends AbstractAbfsIntegrationTest {
@@ -72,25 +85,27 @@ public class ITestAzureBlobFileSystemCheckAccess
     if (this.testUserFs != null) {
       return;
     }
-    String orgClientId = getConfiguration().get(FS_AZURE_BLOB_FS_CLIENT_ID);
+    checkIfConfigIsSet(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ENDPOINT
-    String orgClientSecret = getConfiguration()
+        + "." + getAccountName());
-        .get(FS_AZURE_BLOB_FS_CLIENT_SECRET);
+    Configuration conf = getRawConfiguration();
-    Boolean orgCreateFileSystemDurungInit = getConfiguration()
+    setTestFsConf(FS_AZURE_BLOB_FS_CLIENT_ID,
-        .getBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION, true);
+        FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_ID);
-    getRawConfiguration().set(FS_AZURE_BLOB_FS_CLIENT_ID,
+    setTestFsConf(FS_AZURE_BLOB_FS_CLIENT_SECRET,
-        getConfiguration().get(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_ID));
+        FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET);
-    getRawConfiguration().set(FS_AZURE_BLOB_FS_CLIENT_SECRET, getConfiguration()
+    conf.set(FS_AZURE_ACCOUNT_AUTH_TYPE_PROPERTY_NAME, AuthType.OAuth.name());
-        .get(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET));
+    conf.set(FS_AZURE_ACCOUNT_TOKEN_PROVIDER_TYPE_PROPERTY_NAME + "."
-    getRawConfiguration()
+        + getAccountName(), ClientCredsTokenProvider.class.getName());
-        .setBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION,
+    conf.setBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION,
-            false);
+        false);
-    FileSystem fs = FileSystem.newInstance(getRawConfiguration());
+    this.testUserFs = FileSystem.newInstance(getRawConfiguration());
-    getRawConfiguration().set(FS_AZURE_BLOB_FS_CLIENT_ID, orgClientId);
+  }
-    getRawConfiguration().set(FS_AZURE_BLOB_FS_CLIENT_SECRET, orgClientSecret);
+
-    getRawConfiguration()
+  private void setTestFsConf(final String fsConfKey,
-        .setBoolean(AZURE_CREATE_REMOTE_FILESYSTEM_DURING_INITIALIZATION,
+      final String testFsConfKey) {
-            orgCreateFileSystemDurungInit);
+    final String confKeyWithAccountName = fsConfKey + "." + getAccountName();
-    this.testUserFs = fs;
+    final String confValue = getConfiguration()
+        .getString(testFsConfKey, "");
+    getRawConfiguration().set(confKeyWithAccountName, confValue);
   }
 
   @Test(expected = IllegalArgumentException.class)
@@ -100,15 +115,17 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test(expected = NullPointerException.class)
   public void testCheckAccessForFileWithNullFsAction() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    Assume.assumeTrue(FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT + " is false",
+        isHNSEnabled);
+    Assume.assumeTrue(FS_AZURE_ENABLE_CHECK_ACCESS + " is false",
+        isCheckAccessEnabled);
     //  NPE when trying to convert null FsAction enum
     superUserFs.access(new Path("test.txt"), null);
   }
 
   @Test(expected = FileNotFoundException.class)
   public void testCheckAccessForNonExistentFile() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path nonExistentFile = setupTestDirectoryAndUserAccess(
         "/nonExistentFile1.txt", FsAction.ALL);
     superUserFs.delete(nonExistentFile, true);
@@ -153,15 +170,38 @@ public class ITestAzureBlobFileSystemCheckAccess
         getConfiguration()
             .getBoolean(FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT, true));
     Assume.assumeTrue(FS_AZURE_ENABLE_CHECK_ACCESS + " is false",
-            isCheckAccessEnabled);
+        isCheckAccessEnabled);
+    checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_ID);
+    checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET);
+    checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_USER_GUID);
+
     setTestUserFs();
+
+    //  When the driver does not know if the account is HNS enabled or not it
+    //  makes a server call and fails
+    intercept(AccessControlException.class,
+        "\"This request is not authorized to perform this operation using "
+            + "this permission.\", 403",
+        () -> testUserFs.access(new Path("/"), FsAction.READ));
+
+    //  When the driver has already determined if the account is HNS enabled
+    //  or not, and as the account is non HNS the AzureBlobFileSystem#access
+    //  acts as noop
+    AzureBlobFileSystemStore mockAbfsStore =
+        Mockito.mock(AzureBlobFileSystemStore.class);
+    Mockito.when(mockAbfsStore.getIsNamespaceEnabled()).thenReturn(true);
+    Field abfsStoreField = AzureBlobFileSystem.class.getDeclaredField(
+        "abfsStore");
+    abfsStoreField.setAccessible(true);
+    abfsStoreField.set(testUserFs, mockAbfsStore);
     testUserFs.access(new Path("/"), FsAction.READ);
+
+    superUserFs.access(new Path("/"), FsAction.READ);
   }
 
   @Test
   public void testFsActionNONE() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test2.txt",
         FsAction.NONE);
     assertInaccessible(testFilePath, FsAction.EXECUTE);
@@ -175,8 +215,7 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test
   public void testFsActionEXECUTE() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test3.txt",
         FsAction.EXECUTE);
     assertAccessible(testFilePath, FsAction.EXECUTE);
@@ -191,8 +230,7 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test
   public void testFsActionREAD() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test4.txt",
         FsAction.READ);
     assertAccessible(testFilePath, FsAction.READ);
@@ -207,8 +245,7 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test
   public void testFsActionWRITE() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test5.txt",
         FsAction.WRITE);
     assertAccessible(testFilePath, FsAction.WRITE);
@@ -223,8 +260,7 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test
   public void testFsActionREADEXECUTE() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test6.txt",
         FsAction.READ_EXECUTE);
     assertAccessible(testFilePath, FsAction.EXECUTE);
@@ -239,8 +275,7 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test
   public void testFsActionWRITEEXECUTE() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test7.txt",
         FsAction.WRITE_EXECUTE);
     assertAccessible(testFilePath, FsAction.EXECUTE);
@@ -255,8 +290,7 @@ public class ITestAzureBlobFileSystemCheckAccess
 
   @Test
   public void testFsActionALL() throws Exception {
-    assumeHNSAndCheckAccessEnabled();
+    checkPrerequisites();
-    setTestUserFs();
     Path testFilePath = setupTestDirectoryAndUserAccess("/test8.txt",
         FsAction.ALL);
     assertAccessible(testFilePath, FsAction.EXECUTE);
@@ -268,13 +302,22 @@ public class ITestAzureBlobFileSystemCheckAccess
     assertAccessible(testFilePath, FsAction.ALL);
   }
 
-  private void assumeHNSAndCheckAccessEnabled() {
+  private void checkPrerequisites() throws Exception {
+    setTestUserFs();
     Assume.assumeTrue(FS_AZURE_TEST_NAMESPACE_ENABLED_ACCOUNT + " is false",
         isHNSEnabled);
     Assume.assumeTrue(FS_AZURE_ENABLE_CHECK_ACCESS + " is false",
         isCheckAccessEnabled);
+    checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_ID);
+    checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_CLIENT_SECRET);
+    checkIfConfigIsSet(FS_AZURE_BLOB_FS_CHECKACCESS_TEST_USER_GUID);
+  }
 
-    Assume.assumeNotNull(getRawConfiguration().get(FS_AZURE_BLOB_FS_CLIENT_ID));
+  private void checkIfConfigIsSet(String configKey){
+    AbfsConfiguration conf = getConfiguration();
+    String value = conf.get(configKey);
+    Preconditions.checkArgument((value != null && value.trim().length() > 1),
+        configKey + " config is mandatory for the test to run");
   }
 
   private void assertAccessible(Path testFilePath, FsAction fsAction)