HADOOP-16587. Make ABFS AAD endpoints configurable.

Contributed by Bilahari T H.

This also addresses HADOOP-16498: AzureADAuthenticator cannot authenticate
in China.

Change-Id: I2441dd48b50b59b912b0242f7f5a4418cf94a87c

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

@@ -25,9 +25,12 @@ import java.util.Map;
 
 import com.google.common.annotations.VisibleForTesting;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.hadoop.classification.InterfaceAudience;
 import org.apache.hadoop.classification.InterfaceStability;
 import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.azurebfs.constants.AbfsHttpConstants;
+import org.apache.hadoop.fs.azurebfs.constants.AuthConfigurations;
 import org.apache.hadoop.fs.azurebfs.contracts.annotations.ConfigurationValidationAnnotations.IntegerConfigurationValidatorAnnotation;
 import org.apache.hadoop.fs.azurebfs.contracts.annotations.ConfigurationValidationAnnotations.LongConfigurationValidatorAnnotation;
 import org.apache.hadoop.fs.azurebfs.contracts.annotations.ConfigurationValidationAnnotations.StringConfigurationValidatorAnnotation;
@@ -69,6 +72,7 @@ import static org.apache.hadoop.fs.azurebfs.constants.FileSystemConfigurations.*
 @InterfaceAudience.Private
 @InterfaceStability.Evolving
 public class AbfsConfiguration{
+
   private final Configuration rawConfig;
   private final String accountName;
   private final boolean isSecure;
@@ -486,13 +490,25 @@ public class AbfsConfiguration{
           String password = getPasswordString(FS_AZURE_ACCOUNT_OAUTH_USER_PASSWORD);
           tokenProvider = new UserPasswordTokenProvider(authEndpoint, username, password);
         } else if (tokenProviderClass == MsiTokenProvider.class) {
+          String authEndpoint = getTrimmedPasswordString(
+              FS_AZURE_ACCOUNT_OAUTH_MSI_ENDPOINT,
+              AuthConfigurations.DEFAULT_FS_AZURE_ACCOUNT_OAUTH_MSI_ENDPOINT);
           String tenantGuid = getPasswordString(FS_AZURE_ACCOUNT_OAUTH_MSI_TENANT);
           String clientId = getPasswordString(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ID);
-          tokenProvider = new MsiTokenProvider(tenantGuid, clientId);
+          String authority = getTrimmedPasswordString(
+              FS_AZURE_ACCOUNT_OAUTH_MSI_AUTHORITY,
+              AuthConfigurations.DEFAULT_FS_AZURE_ACCOUNT_OAUTH_MSI_AUTHORITY);
+          authority = appendSlashIfNeeded(authority);
+          tokenProvider = new MsiTokenProvider(authEndpoint, tenantGuid,
+              clientId, authority);
         } else if (tokenProviderClass == RefreshTokenBasedTokenProvider.class) {
+          String authEndpoint = getTrimmedPasswordString(
+              FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN_ENDPOINT,
+              AuthConfigurations.DEFAULT_FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN_ENDPOINT);
           String refreshToken = getPasswordString(FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN);
           String clientId = getPasswordString(FS_AZURE_ACCOUNT_OAUTH_CLIENT_ID);
-          tokenProvider = new RefreshTokenBasedTokenProvider(clientId, refreshToken);
+          tokenProvider = new RefreshTokenBasedTokenProvider(authEndpoint,
+              clientId, refreshToken);
         } else {
           throw new IllegalArgumentException("Failed to initialize " + tokenProviderClass);
         }
@@ -649,4 +665,19 @@ public class AbfsConfiguration{
     this.disableOutputStreamFlush = disableOutputStreamFlush;
   }
 
+  private String getTrimmedPasswordString(String key, String defaultValue) throws IOException {
+    String value = getPasswordString(key);
+    if (StringUtils.isBlank(value)) {
+      value = defaultValue;
+    }
+    return value.trim();
+  }
+
+  private String appendSlashIfNeeded(String authority) {
+    if (!authority.endsWith(AbfsHttpConstants.FORWARD_SLASH)) {
+      authority = authority + AbfsHttpConstants.FORWARD_SLASH;
+    }
+    return authority;
+  }
+
 }

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

@@ -0,0 +1,45 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ * <p>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p>
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.hadoop.fs.azurebfs.constants;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+/**
+ * Responsible to keep all the Azure Blob File System auth related
+ * configurations.
+ */
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public final class AuthConfigurations {
+
+  /** Default OAuth token end point for the MSI flow. */
+  public static final String DEFAULT_FS_AZURE_ACCOUNT_OAUTH_MSI_ENDPOINT =
+      "http://169.254.169.254/metadata/identity/oauth2/token";
+  /** Default value for authority for the MSI flow. */
+  public static final String DEFAULT_FS_AZURE_ACCOUNT_OAUTH_MSI_AUTHORITY =
+      "https://login.microsoftonline.com/";
+  /** Default OAuth token end point for the refresh token flow. */
+  public static final String
+      DEFAULT_FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN_ENDPOINT =
+      "https://login.microsoftonline.com/Common/oauth2/token";
+
+  private AuthConfigurations() {
+  }
+}

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

@@ -102,12 +102,18 @@ public final class ConfigurationKeys {
   public static final String FS_AZURE_ACCOUNT_OAUTH_CLIENT_ENDPOINT = "fs.azure.account.oauth2.client.endpoint";
   /** Key for oauth msi tenant id: {@value}. */
   public static final String FS_AZURE_ACCOUNT_OAUTH_MSI_TENANT = "fs.azure.account.oauth2.msi.tenant";
+  /** Key for oauth msi endpoint: {@value}. */
+  public static final String FS_AZURE_ACCOUNT_OAUTH_MSI_ENDPOINT = "fs.azure.account.oauth2.msi.endpoint";
+  /** Key for oauth msi Authority: {@value}. */
+  public static final String FS_AZURE_ACCOUNT_OAUTH_MSI_AUTHORITY = "fs.azure.account.oauth2.msi.authority";
   /** Key for oauth user name: {@value}. */
   public static final String FS_AZURE_ACCOUNT_OAUTH_USER_NAME = "fs.azure.account.oauth2.user.name";
   /** Key for oauth user password: {@value}. */
   public static final String FS_AZURE_ACCOUNT_OAUTH_USER_PASSWORD = "fs.azure.account.oauth2.user.password";
   /** Key for oauth refresh token: {@value}. */
   public static final String FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN = "fs.azure.account.oauth2.refresh.token";
+  /** Key for oauth AAD refresh token endpoint: {@value}. */
+  public static final String FS_AZURE_ACCOUNT_OAUTH_REFRESH_TOKEN_ENDPOINT = "fs.azure.account.oauth2.refresh.token.endpoint";
 
   public static String accountProperty(String property, String account) {
     return property + "." + account;

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

@@ -101,6 +101,9 @@ public final class AzureADAuthenticator {
    * an Azure VM with MSI extension
    * enabled.
    *
+   * @param authEndpoint the OAuth 2.0 token endpoint associated
+   *                     with the user's directory (obtain from
+   *                     Active Directory configuration)
    * @param tenantGuid  (optional) The guid of the AAD tenant. Can be {@code null}.
    * @param clientId    (optional) The clientId guid of the MSI service
    *                    principal to use. Can be {@code null}.
@@ -109,17 +112,16 @@ public final class AzureADAuthenticator {
    * @return {@link AzureADToken} obtained using the creds
    * @throws IOException throws IOException if there is a failure in obtaining the token
    */
-  public static AzureADToken getTokenFromMsi(String tenantGuid, String clientId,
+  public static AzureADToken getTokenFromMsi(final String authEndpoint,
-                                             boolean bypassCache) throws IOException {
+      final String tenantGuid, final String clientId, String authority,
-    String authEndpoint = "http://169.254.169.254/metadata/identity/oauth2/token";
+      boolean bypassCache) throws IOException {
-
     QueryParams qp = new QueryParams();
     qp.add("api-version", "2018-02-01");
     qp.add("resource", RESOURCE_NAME);
 
-
     if (tenantGuid != null && tenantGuid.length() > 0) {
-      String authority = "https://login.microsoftonline.com/" + tenantGuid;
+      authority = authority + tenantGuid;
+      LOG.debug("MSI authority : {}", authority);
       qp.add("authority", authority);
     }
 
@@ -141,14 +143,17 @@ public final class AzureADAuthenticator {
   /**
    * Gets Azure Active Directory token using refresh token.
    *
+   * @param authEndpoint the OAuth 2.0 token endpoint associated
+   *                     with the user's directory (obtain from
+   *                     Active Directory configuration)
    * @param clientId the client ID (GUID) of the client web app obtained from Azure Active Directory configuration
    * @param refreshToken the refresh token
    * @return {@link AzureADToken} obtained using the refresh token
    * @throws IOException throws IOException if there is a failure in connecting to Azure AD
    */
-  public static AzureADToken getTokenUsingRefreshToken(String clientId,
+  public static AzureADToken getTokenUsingRefreshToken(
-                                                       String refreshToken) throws IOException {
+      final String authEndpoint, final String clientId,
-    String authEndpoint = "https://login.microsoftonline.com/Common/oauth2/token";
+      final String refreshToken) throws IOException {
     QueryParams qp = new QueryParams();
     qp.add("grant_type", "refresh_token");
     qp.add("refresh_token", refreshToken);

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

@@ -28,21 +28,29 @@ import org.slf4j.LoggerFactory;
  */
 public class MsiTokenProvider extends AccessTokenProvider {
 
+  private final String authEndpoint;
+
+  private final String authority;
+
   private final String tenantGuid;
 
   private final String clientId;
 
   private static final Logger LOG = LoggerFactory.getLogger(AccessTokenProvider.class);
 
-  public MsiTokenProvider(final String tenantGuid, final String clientId) {
+  public MsiTokenProvider(final String authEndpoint, final String tenantGuid,
+      final String clientId, final String authority) {
+    this.authEndpoint = authEndpoint;
     this.tenantGuid = tenantGuid;
     this.clientId = clientId;
+    this.authority = authority;
   }
 
   @Override
   protected AzureADToken refreshToken() throws IOException {
     LOG.debug("AADToken: refreshing token from MSI");
-    AzureADToken token = AzureADAuthenticator.getTokenFromMsi(tenantGuid, clientId, false);
+    AzureADToken token = AzureADAuthenticator
+        .getTokenFromMsi(authEndpoint, tenantGuid, clientId, authority, false);
     return token;
   }
-}
+}

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

@@ -31,6 +31,8 @@ import org.slf4j.LoggerFactory;
 public class RefreshTokenBasedTokenProvider extends AccessTokenProvider {
   private static final Logger LOG = LoggerFactory.getLogger(AccessTokenProvider.class);
 
+  private final String authEndpoint;
+
   private final String clientId;
 
   private final String refreshToken;
@@ -41,9 +43,12 @@ public class RefreshTokenBasedTokenProvider extends AccessTokenProvider {
    * @param clientId the client ID (GUID) of the client web app obtained from Azure Active Directory configuration
    * @param refreshToken the refresh token
    */
-  public RefreshTokenBasedTokenProvider(String clientId, String refreshToken) {
+  public RefreshTokenBasedTokenProvider(final String authEndpoint,
+      String clientId, String refreshToken) {
+    Preconditions.checkNotNull(authEndpoint, "authEndpoint");
     Preconditions.checkNotNull(clientId, "clientId");
     Preconditions.checkNotNull(refreshToken, "refreshToken");
+    this.authEndpoint = authEndpoint;
     this.clientId = clientId;
     this.refreshToken = refreshToken;
   }
@@ -52,6 +57,7 @@ public class RefreshTokenBasedTokenProvider extends AccessTokenProvider {
   @Override
   protected AzureADToken refreshToken() throws IOException {
     LOG.debug("AADToken: refreshing refresh-token based token");
-    return AzureADAuthenticator.getTokenUsingRefreshToken(clientId, refreshToken);
+    return AzureADAuthenticator
+        .getTokenUsingRefreshToken(authEndpoint, clientId, refreshToken);
   }
-}
+}