HADOOP-19471. ABFS: Support Fixed SAS Token at Container Level

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.*;
@@ -88,6 +89,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;
@@ -484,6 +486,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.
@@ -586,7 +606,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;
   }
 
   /**
@@ -644,17 +674,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

@@ -229,7 +229,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/site/markdown/abfs.md

@@ -742,7 +742,7 @@ ADLS Gen 2 storage accounts.
 #### Using Account/Service SAS with ABFS
 
 - **Description**: ABFS allows user to use Account/Service SAS for authenticating
-requests. User can specify them as fixed SAS Token to be used across all the requests.
+  requests. User can specify them as fixed SAS Token to be used across all the requests.
 
 - **Configuration**: To use this method with ABFS Driver, specify the following properties in your `core-site.xml` file:
 
@@ -754,22 +754,41 @@ requests. User can specify them as fixed SAS Token to be used across all the req
         </property>
         ```
 
-    1.  Fixed SAS Token:
+    2. Account SAS (Fixed SAS Token at Account Level):
+          ```xml
+          <property>
+            <name>fs.azure.sas.fixed.token.ACCOUNT_NAME</name>
+            <value>FIXED_ACCOUNT_SAS_TOKEN</value>
+          </property>
+          ```
+
+    - Replace `FIXED_ACCOUNT_SAS_TOKEN` with fixed Account/Service SAS. You can also
+      generate SAS from Azure portal. Account -> Security + Networking -> Shared Access Signature
+
+    3. Service  SAS (Fixed SAS Token at Container Level):
         ```xml
-        <property>
-          <name>fs.azure.sas.fixed.token</name>
-          <value>FIXED_SAS_TOKEN</value>
-        </property>
-        ```
+           <property>
+             <name>fs.azure.sas.fixed.token.CONTAINER_NAME.ACCOUNT_NAME</name>
+             <value>FIXED_SAS_TOKEN</value>
+           </property>
+           ```
+
+    - Replace `FIXED_SERVICE_SAS_TOKEN` with fixed Service SAS. You can also
+      generate SAS from Azure portal. Account -> Data storage -> Containers ->
+      right click on your container and select generate SAS ->
+      Give valid permissions and expiry time -> Click on generate SAS and copy
+      the SAS token.
 
-    Replace `FIXED_SAS_TOKEN` with fixed Account/Service SAS. You can also
-generate SAS from Azure portal. Account -> Security + Networking -> Shared Access Signature
 
 - **Security**: Account/Service SAS requires account keys to be used which makes
 them less secure. There is no scope of having delegated access to different users.
 
-*Note:* When `fs.azure.sas.token.provider.type` and `fs.azure.fixed.sas.token`
-are both configured, precedence will be given to the custom token provider implementation.
+*Note:*
+- Preference order for SAS will be:
+  - fs.azure.sas.token.provider.type
+  - fs.azure.sas.fixed.token.CONTAINER_NAME.ACCOUNT_NAME
+  - fs.azure.sas.fixed.token.ACCOUNT_NAME
+  - fs.azure.sas.fixed.token
 
 ## <a name="technical"></a> Technical notes
 

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;
+  }
 }