HADOOP-19471. [ABFS] Support Fixed SAS Token at Container Level (apache#7461) (apache#7597)

Contributed by Manika Joshi
Reviewed by Anuj Modi, Anmol Asrani, Manish Bhatt

Signed off by Anuj Modi <anujmodi@apache.org>

Author: manika137

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

@@ -75,6 +75,7 @@
 import org.apache.hadoop.util.ReflectionUtils;
 
 import static org.apache.hadoop.fs.FileSystem.FS_DEFAULT_NAME_KEY;
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DOT;
 import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.*;
 import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.*;
@@ -89,6 +90,7 @@ public class AbfsConfiguration{
 
   private final Configuration rawConfig;
   private final String accountName;
+  private String fsName;
   // Service type identified from URL used to initialize FileSystem.
   private final AbfsServiceType fsConfiguredServiceType;
   private final boolean isSecure;
@@ -485,6 +487,24 @@ public AbfsConfiguration(final Configuration rawConfig,
     }
   }
 
+  /**
+   * Constructor for AbfsConfiguration for retrieve the FsName.
+   * @param rawConfig used to initialize the configuration.
+   * @param accountName the name of the azure storage account.
+   * @param fsName the name of the file system (container name).
+   * @param fsConfiguredServiceType service type configured for the file system.
+   * @throws IllegalAccessException if the field is not accessible.
+   * @throws IOException if an I/O error occurs.
+   */
+  public AbfsConfiguration(final Configuration rawConfig,
+      String accountName,
+      String fsName,
+      AbfsServiceType fsConfiguredServiceType)
+      throws IllegalAccessException, IOException {
+    this(rawConfig, accountName, fsConfiguredServiceType);
+    this.fsName = fsName;
+  }
+
   /**
    * Constructor for AbfsConfiguration for default service type i.e. DFS.
    * @param rawConfig used to initialize the configuration.
@@ -584,7 +604,17 @@ public String getClientCorrelationId() {
    * @return Account-specific configuration key
    */
   public String accountConf(String key) {
-    return key + "." + accountName;
+    return key + DOT + accountName;
+  }
+
+  /**
+   * Appends the container, account name to a configuration key yielding the
+   * container-specific form.
+   * @param key Account-agnostic configuration key
+   * @return Container-specific configuration key
+   */
+  public String containerConf(String key) {
+    return key + DOT + fsName + DOT + accountName;
   }
 
   /**
@@ -642,17 +672,19 @@ public int getInt(String key, int defaultValue) {
   }
 
   /**
-   * Returns the account-specific password in string form if it exists, then
+   * Returns the container-specific password if it exists,
+   * else searches for the account-specific password, else finally
    * looks for an account-agnostic value.
    * @param key Account-agnostic configuration key
    * @return value in String form if one exists, else null
    * @throws IOException if parsing fails.
    */
   public String getPasswordString(String key) throws IOException {
-    char[] passchars = rawConfig.getPassword(accountConf(key));
-    if (passchars == null) {
-      passchars = rawConfig.getPassword(key);
-    }
+    char[] passchars = rawConfig.getPassword(containerConf(key)) != null
+        ? rawConfig.getPassword(containerConf(key))
+        : rawConfig.getPassword(accountConf(key)) != null
+            ? rawConfig.getPassword(accountConf(key))
+            : rawConfig.getPassword(key);
     if (passchars != null) {
       return new String(passchars);
     }

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

@@ -227,7 +227,7 @@ public AzureBlobFileSystemStore(
 
     try {
       this.abfsConfiguration = new AbfsConfiguration(abfsStoreBuilder.configuration,
-          accountName, getAbfsServiceTypeFromUrl());
+          accountName, fileSystemName, getAbfsServiceTypeFromUrl());
     } catch (IllegalAccessException exception) {
       throw new FileSystemOperationUnhandledException(exception);
     }

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

@@ -22,6 +22,8 @@
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.fs.FileSystem;
 
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.DOT;
+
 /**
  * Responsible to keep all the Azure Blob File System configurations keys in Hadoop configuration file.
  */
@@ -319,7 +321,11 @@ public final class ConfigurationKeys {
   public static final String FS_AZURE_ABFS_ENABLE_CHECKSUM_VALIDATION = "fs.azure.enable.checksum.validation";
 
   public static String accountProperty(String property, String account) {
-    return property + "." + account;
+    return property + DOT + account;
+  }
+
+  public static String containerProperty(String property, String fsName, String account) {
+    return property + DOT + fsName + DOT + account;
   }
 
   public static final String FS_AZURE_ENABLE_DELEGATION_TOKEN = "fs.azure.enable.delegation.token";

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

@@ -32,11 +32,14 @@
 import org.apache.hadoop.fs.azurebfs.services.AuthType;
 import org.apache.hadoop.fs.azurebfs.services.FixedSASTokenProvider;
 import org.apache.hadoop.fs.azurebfs.utils.AccountSASGenerator;
+import org.apache.hadoop.fs.azurebfs.utils.ServiceSASGenerator;
 import org.apache.hadoop.fs.azurebfs.utils.Base64;
 
+import static org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants.EMPTY_STRING;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_SAS_FIXED_TOKEN;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.FS_AZURE_SAS_TOKEN_PROVIDER_TYPE;
 import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.accountProperty;
+import static org.apache.hadoop.fs.azurebfs.constants.ConfigurationKeys.containerProperty;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_TEST_APP_ID;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_TEST_APP_SECRET;
 import static org.apache.hadoop.fs.azurebfs.constants.TestConfigurationKeys.FS_AZURE_TEST_APP_SERVICE_PRINCIPAL_OBJECT_ID;
@@ -50,6 +53,8 @@
 public class ITestAzureBlobFileSystemChooseSAS extends AbstractAbfsIntegrationTest{
 
   private String accountSAS = null;
+  private String containerSAS = null;
+  private String accountAgnosticSAS = null;
   private static final String TEST_PATH = "testPath";
 
   /**
@@ -69,6 +74,8 @@ public void setup() throws Exception {
     super.setup();
     createFilesystemWithTestFileForSASTests(new Path(TEST_PATH));
     generateAccountSAS();
+    generateAccountAgnosticSAS();
+    generateContainerSAS();
   }
 
   /**
@@ -85,6 +92,37 @@ private void generateAccountSAS() throws AzureBlobFileSystemException {
     accountSAS = configAccountSASGenerator.getAccountSAS(getAccountName());
   }
 
+  /**
+   * Generates an Account SAS Token (for account-agnostic config) using the Account Shared Key to
+   * be used as a fixed SAS Token.
+   * Account SAS used here will  only have write permissions to resources.
+   * This will be used by individual tests to set in the configurations.
+   * @throws AzureBlobFileSystemException
+   */
+  private void generateAccountAgnosticSAS() throws AzureBlobFileSystemException {
+    final String accountKey = getConfiguration().getStorageAccountKey();
+    AccountSASGenerator configAccountSASGenerator = new AccountSASGenerator(Base64.decode(accountKey));
+    // Setting only write permissions.
+    configAccountSASGenerator.setPermissions("w");
+    accountAgnosticSAS = configAccountSASGenerator.getAccountSAS(getAccountName());
+  }
+
+  /**
+   * Generates a Container SAS Token using the Account Shared Key to be used as a fixed SAS Token.
+   * Container SAS used here will have only read permissions to resources.
+   * This will be used by individual tests to set in the configurations.
+   * @throws AzureBlobFileSystemException
+   */
+  private void generateContainerSAS() throws AzureBlobFileSystemException {
+    final byte[] accountKey = Base64.decode(
+        getConfiguration().getStorageAccountKey());
+    ServiceSASGenerator configServiceSASGenerator = new ServiceSASGenerator(
+        accountKey);
+    // Setting only read permissions.
+    configServiceSASGenerator.setPermissions("r");
+    containerSAS = configServiceSASGenerator.getContainerSASWithFullControl(
+        getAccountName(), getFileSystemName());
+  }
   /**
    * Tests the scenario where both the custom SASTokenProvider and a fixed SAS token are configured.
    * Custom implementation of SASTokenProvider class should be chosen and User Delegation SAS should be used.
@@ -126,6 +164,58 @@ public void testBothProviderFixedTokenConfigured() throws Exception {
     }
   }
 
+  /**
+   * Helper method to get the Fixed SAS token value
+   */
+  private String getFixedSASToken(AbfsConfiguration config) throws Exception {
+    return config.getSASTokenProvider()
+        .getSASToken(this.getAccountName(), this.getFileSystemName(),
+            getMethodName(),
+            EMPTY_STRING);
+  }
+
+  /**
+   * Tests the implementation sequence if all fixed SAS configs are set.
+   * The expected sequence is Container Specific Fixed SAS, Account Specific Fixed SAS, Account Agnostic Fixed SAS.
+   * @throws IOException
+   */
+  @Test
+  public void testFixedSASTokenProviderPreference() throws Exception {
+    AbfsConfiguration testAbfsConfig = new AbfsConfiguration(
+        getRawConfiguration(), this.getAccountName(), this.getFileSystemName(),
+        getAbfsServiceType());
+
+    // setting all types of Fixed SAS configs (container-specific, account-specific, account-agnostic)
+    removeAnyPresetConfiguration(testAbfsConfig);
+    testAbfsConfig.set(
+        containerProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getFileSystemName(),
+            this.getAccountName()), containerSAS);
+    testAbfsConfig.set(
+        accountProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getAccountName()),
+        accountSAS);
+    testAbfsConfig.set(FS_AZURE_SAS_FIXED_TOKEN, accountAgnosticSAS);
+
+    // Assert that Container Specific Fixed SAS is used
+    Assertions.assertThat(getFixedSASToken(testAbfsConfig))
+        .describedAs("Container-specific fixed SAS should've been used.")
+        .isEqualTo(containerSAS);
+
+    // Assert that Account Specific Fixed SAS is used if container SAS isn't set
+    testAbfsConfig.unset(
+        containerProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getFileSystemName(),
+            this.getAccountName()));
+    Assertions.assertThat(getFixedSASToken(testAbfsConfig))
+        .describedAs("Account-specific fixed SAS should've been used.")
+        .isEqualTo(accountSAS);
+
+    //Assert that Account-Agnostic fixed SAS is used if no other fixed SAS configs are set.
+    testAbfsConfig.unset(
+        accountProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getAccountName()));
+    Assertions.assertThat(getFixedSASToken(testAbfsConfig))
+        .describedAs("Account-agnostic fixed SAS should've been used.")
+        .isEqualTo(accountAgnosticSAS);
+  }
+
   /**
    * Tests the scenario where only the fixed token is configured, and no token provider class is set.
    * Account SAS Token configured as fixed SAS should be used.
@@ -189,5 +279,6 @@ private void removeAnyPresetConfiguration(AbfsConfiguration testAbfsConfig) {
     testAbfsConfig.unset(FS_AZURE_SAS_FIXED_TOKEN);
     testAbfsConfig.unset(accountProperty(FS_AZURE_SAS_TOKEN_PROVIDER_TYPE, this.getAccountName()));
     testAbfsConfig.unset(accountProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getAccountName()));
+    testAbfsConfig.unset(containerProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getFileSystemName(), this.getAccountName()));
   }
 }

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

@@ -37,10 +37,11 @@ public ServiceSASGenerator(byte[] accountKey) {
     super(accountKey);
   }
 
+  private String permissions = "racwdl";
   public String getContainerSASWithFullControl(String accountName, String containerName) throws
       InvalidConfigurationValueException {
     accountName = getCanonicalAccountName(accountName);
-    String sp = "rcwdl";
+    String sp = permissions;
     String sv = AuthenticationVersion.Feb20.toString();
     String sr = "c";
     String st = ISO_8601_FORMATTER.format(Instant.now().minus(FIVE_MINUTES));
@@ -96,4 +97,13 @@ private String computeSignatureForSAS(String sp, String st, String se, String sv
     LOG.debug("Service SAS stringToSign: " + stringToSign.replace("\n", "."));
     return computeHmac256(stringToSign);
   }
+
+  /**
+   * By default, Container SAS has all the available permissions. Use this to
+   * override the default permissions and set as per the requirements.
+   * @param permissions
+   */
+  public void setPermissions(final String permissions) {
+    this.permissions = permissions;
+  }
 }