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

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

@@ -88,6 +88,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 +485,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.
@@ -589,6 +608,16 @@ public String accountConf(String key) {
     return key + "." + 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 + "." + fsName + "." + accountName;
+  }
+
   /**
    * Returns the account-specific value if it exists, then looks for an
    * account-agnostic value.
@@ -644,17 +673,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

@@ -322,6 +322,10 @@ public static String accountProperty(String property, String account) {
     return property + "." + account;
   }
 
+  public static String containerProperty(String property, String fsName, String account) {
+    return property + "." + fsName + "." + account;
+  }
+
   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";
 

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

@@ -32,11 +32,13 @@
 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.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 +52,7 @@
 public class ITestAzureBlobFileSystemChooseSAS extends AbstractAbfsIntegrationTest{
 
   private String accountSAS = null;
+  private String containerSAS = null;
   private static final String TEST_PATH = "testPath";
 
   /**
@@ -69,6 +72,7 @@ public void setup() throws Exception {
     super.setup();
     createFilesystemWithTestFileForSASTests(new Path(TEST_PATH));
     generateAccountSAS();
+    generateContainerSAS();
   }
 
   /**
@@ -85,6 +89,22 @@ private void generateAccountSAS() throws AzureBlobFileSystemException {
     accountSAS = 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 +146,44 @@ public void testBothProviderFixedTokenConfigured() throws Exception {
     }
   }
 
+  /**
+   * Helper method to get the Fixed SAS token value
+   */
+  private String getFixedSASToken(AbfsConfiguration config) throws Exception {
+    String readPermission = "read";
+    return config.getSASTokenProvider().getSASToken(this.getAccountName(), this.getFileSystemName(), getMethodName(),
+        readPermission);
+  }
+
+  /**
+   * 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 testFixedTokenPreference() 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, accountSAS);
+
+    // Assert that Container Specific Fixed SAS is used
+    Assertions.assertThat(getFixedSASToken(testAbfsConfig)).contains("sr=c");
+
+    // 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)).contains("ss=bf");
+
+    //Assert that Account-Agnostic fixed SAS is used if no other fixed SAS configs are set.
+    // The token is the same as the Account Specific Fixed SAS.
+    testAbfsConfig.unset(accountProperty(FS_AZURE_SAS_FIXED_TOKEN, this.getAccountName()));
+    Assertions.assertThat(getFixedSASToken(testAbfsConfig)).contains("ss=bf");
+  }
+
   /**
    * 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 +247,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;
+  }
 }