HADOOP-16105. WASB in secure mode does not set connectingUsingSAS.

Contributed by Steve Loughran.

(cherry picked from commit 9cb2f470b759bbe7609a00e8f8f72779e2daae80)

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

@@ -907,15 +907,16 @@ public class AzureNativeFileSystemStore implements NativeFileSystemStore {
       String containerName, URI sessionUri)
           throws AzureException, StorageException, URISyntaxException {
 
+    LOG.debug("Connecting to Azure storage in Secure Mode");
     // Assertion: storageInteractionLayer instance has to be a SecureStorageInterfaceImpl
     if (!(this.storageInteractionLayer instanceof SecureStorageInterfaceImpl)) {
-      throw new AssertionError("connectToAzureStorageInSASKeyMode() should be called only"
+      throw new AssertionError("connectToAzureStorageInSecureMode() should be called only"
         + " for SecureStorageInterfaceImpl instances");
     }
 
     ((SecureStorageInterfaceImpl) this.storageInteractionLayer).
       setStorageAccountName(accountName);
-
+    connectingUsingSAS = true;
     container = storageInteractionLayer.getContainerReference(containerName);
     rootDirectory = container.getDirectoryReference("");
 

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

@@ -43,6 +43,8 @@ import com.microsoft.azure.storage.StorageException;
 import com.microsoft.azure.storage.blob.CloudBlobClient;
 import com.microsoft.azure.storage.blob.CloudBlobContainer;
 import com.microsoft.azure.storage.blob.CloudBlockBlob;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /***
  * Local SAS Key Generation implementation. This class resides in
@@ -54,6 +56,9 @@ import com.microsoft.azure.storage.blob.CloudBlockBlob;
 
 public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
 
+  public static final Logger LOG = LoggerFactory.getLogger(
+      LocalSASKeyGeneratorImpl.class);
+
   /**
    * Map to cache CloudStorageAccount instances.
    */
@@ -75,6 +80,7 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
   public URI getContainerSASUri(String accountName, String container)
       throws SASKeyGenerationException {
 
+    LOG.debug("Retrieving Container SAS URI For {}@{}", container, accountName);
     try {
 
       CachedSASKeyEntry cacheKey = new CachedSASKeyEntry(accountName, container, "/");
@@ -113,7 +119,7 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
    */
   private CloudStorageAccount getSASKeyBasedStorageAccountInstance(
       String accountName) throws SASKeyGenerationException {
-
+    LOG.debug("Creating SAS key from account instance {}", accountName);
     try {
 
       String accountNameWithoutDomain =
@@ -223,6 +229,10 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
       String accountKey) throws SASKeyGenerationException {
 
     if (!storageAccountMap.containsKey(accountName)) {
+      if (accountKey == null || accountKey.isEmpty()) {
+        throw new SASKeyGenerationException(
+            "No key for Storage account " + accountName);
+      }
 
       CloudStorageAccount account = null;
       try {
@@ -282,4 +292,4 @@ public class LocalSASKeyGeneratorImpl extends SASKeyGeneratorImpl {
         SharedAccessAccountPermissions.UPDATE,
         SharedAccessAccountPermissions.WRITE);
   }
-}
+}

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

@@ -83,8 +83,10 @@ public class SecureStorageInterfaceImpl extends StorageInterface {
       Configuration conf) throws SecureModeException {
 
     if (useLocalSASKeyMode) {
+      LOG.debug("Authenticating with SecureStorage and local SAS key");
       this.sasKeyGenerator = new LocalSASKeyGeneratorImpl(conf);
     } else {
+      LOG.debug("Authenticating with SecureStorage and remote SAS key generation");
       RemoteSASKeyGeneratorImpl remoteSasKeyGenerator =
           new RemoteSASKeyGeneratorImpl(conf);
       try {
@@ -96,6 +98,8 @@ public class SecureStorageInterfaceImpl extends StorageInterface {
       this.sasKeyGenerator = remoteSasKeyGenerator;
     }
     this.useContainerSasKeyForAllAccess = conf.getBoolean(KEY_USE_CONTAINER_SASKEY_FOR_ALL_ACCESS, true);
+    LOG.debug("Container SAS key {} be used for all access",
+        useContainerSasKeyForAllAccess ? "will" : "will not");
   }
 
   @Override

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

@@ -590,7 +590,10 @@ public final class AzureBlobStorageTestAccount implements AutoCloseable,
       }
       // Remove the account key from the configuration to make sure we don't
       // cheat and use that.
-      conf.set(ACCOUNT_KEY_PROPERTY_NAME + accountName, "");
+      // but only if not in secure mode, which requires that login
+      if (!conf.getBoolean(AzureNativeFileSystemStore.KEY_USE_SECURE_MODE, false)) {
+        conf.set(ACCOUNT_KEY_PROPERTY_NAME + accountName, "");
+      }
       // Set the SAS key.
       conf.set(SAS_PROPERTY_NAME + containerName + "." + accountName, sas);
     }

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

@@ -19,6 +19,8 @@
 package org.apache.hadoop.fs.azure;
 
 import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.FS_DEFAULT_NAME_KEY;
+import static org.apache.hadoop.test.LambdaTestUtils.intercept;
+import static org.junit.Assume.assumeFalse;
 import static org.junit.Assume.assumeNotNull;
 
 import java.io.ByteArrayInputStream;
@@ -30,8 +32,10 @@ import java.net.URI;
 import java.util.Date;
 import java.util.EnumSet;
 import java.io.File;
+import java.util.NoSuchElementException;
 
 import org.apache.hadoop.fs.azure.integration.AzureTestUtils;
+import org.apache.hadoop.fs.contract.ContractTestUtils;
 import org.apache.hadoop.security.ProviderUtils;
 import org.apache.hadoop.security.alias.CredentialProvider;
 import org.apache.hadoop.security.alias.CredentialProviderFactory;
@@ -42,6 +46,8 @@ import org.apache.hadoop.fs.FileSystem;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.fs.UnsupportedFileSystemException;
 import org.apache.hadoop.fs.azure.AzureBlobStorageTestAccount.CreateOptions;
+import org.apache.hadoop.test.GenericTestUtils;
+
 import org.junit.After;
 import org.junit.Assert;
 import org.junit.Assume;
@@ -50,8 +56,10 @@ import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.TemporaryFolder;
 
+import com.microsoft.azure.storage.StorageException;
 import com.microsoft.azure.storage.blob.CloudBlobContainer;
 import com.microsoft.azure.storage.blob.CloudBlockBlob;
+import com.microsoft.azure.storage.core.SR;
 
 public class ITestWasbUriAndConfiguration extends AbstractWasbTestWithTimeout {
 
@@ -173,6 +181,57 @@ public class ITestWasbUriAndConfiguration extends AbstractWasbTestWithTimeout {
     assertEquals(3, obtained[2]);
   }
 
+  /**
+   * Use secure mode, which will automatically switch to SAS,
+   */
+  @Test
+  public void testConnectUsingSecureSAS() throws Exception {
+    // Create the test account with SAS credentials.
+    Configuration conf = new Configuration();
+    conf.setBoolean(AzureNativeFileSystemStore.KEY_USE_SECURE_MODE, true);
+    testAccount = AzureBlobStorageTestAccount.create("",
+        EnumSet.of(CreateOptions.UseSas),
+        conf);
+    assumeNotNull(testAccount);
+    NativeAzureFileSystem fs = testAccount.getFileSystem();
+
+    AzureException ex = intercept(AzureException.class,
+        SR.ENUMERATION_ERROR,
+        () -> ContractTestUtils.writeTextFile(fs,
+            new Path("/testConnectUsingSecureSAS"),
+            "testConnectUsingSecureSAS",
+            true));
+
+    StorageException cause = getCause(StorageException.class,
+        getCause(NoSuchElementException.class, ex));
+    GenericTestUtils.assertExceptionContains(
+        "The specified container does not exist", cause);
+  }
+
+  /**
+   * Get an inner cause of an exception; require it to be of the given
+   * type.
+   * If there is a problem, an AssertionError is thrown, containing the
+   * outer or inner exception.
+   * @param clazz required class
+   * @param t exception
+   * @param <E> type of required exception
+   * @return the retrieved exception
+   * @throws AssertionError if there is no cause or it is of the wrong type.
+   */
+  private <E extends Throwable> E getCause(
+      Class<E> clazz, Throwable t) {
+    Throwable e = t.getCause();
+    if (e == null) {
+      throw new AssertionError("No cause", t);
+    }
+    if (!clazz.isAssignableFrom(e.getClass())) {
+      throw new AssertionError("Wrong inner class", e);
+    } else {
+      return (E) e;
+    }
+  }
+
   @Test
   public void testConnectUsingAnonymous() throws Exception {
 
@@ -324,7 +383,7 @@ public class ITestWasbUriAndConfiguration extends AbstractWasbTestWithTimeout {
   @Test
   public void testCredsFromCredentialProvider() throws Exception {
 
-    Assume.assumeFalse(runningInSASMode);
+    assumeFalse(runningInSASMode);
     String account = "testacct";
     String key = "testkey";
     // set up conf to have a cred provider