HADOOP-19384. S3A: Add support for ProfileCredentialsProvider

hadoop-common-project/hadoop-common/src/main/resources/core-default.xml

@@ -1442,6 +1442,7 @@
     token binding it may be used
     to communicate wih the STS endpoint to request session/role
     credentials.
+    org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider is also supported, but is not enabled by default.
   </description>
 </property>
 

hadoop-tools/hadoop-aws/src/main/java/org/apache/hadoop/fs/s3a/auth/ProfileAWSCredentialsProvider.java

@@ -0,0 +1,109 @@
+/**
+ * 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
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * 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.s3a.auth;
+
+import java.net.URI;
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
+import software.amazon.awssdk.profiles.ProfileFile;
+
+import org.apache.commons.lang3.SystemUtils;
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+import org.apache.hadoop.conf.Configuration;
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class ProfileAWSCredentialsProvider extends AbstractAWSCredentialProvider {
+    private static final Logger LOG = LoggerFactory.getLogger(ProfileAWSCredentialsProvider.class);
+
+    public static final String NAME
+            = "org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider";
+
+    /** Conf setting for credentials file path*/
+    public static final String PROFILE_FILE = "fs.s3a.auth.profile.file";
+
+    /** Conf setting for profile name*/
+    public static final String PROFILE_NAME = "fs.s3a.auth.profile.name";
+
+    /** Environment variable for credentials file path*/
+    public static final String CREDENTIALS_FILE_ENV = "AWS_SHARED_CREDENTIALS_FILE";
+    /** Environment variable for profile name*/
+    public static final String PROFILE_ENV = "AWS_PROFILE";
+
+    private final ProfileCredentialsProvider pcp;
+
+    private static Path getCredentialsPath(Configuration conf) {
+        String credentialsFile = conf.get(PROFILE_FILE, null);
+        if (credentialsFile == null) {
+            credentialsFile = SystemUtils.getEnvironmentVariable(CREDENTIALS_FILE_ENV, null);
+            if (credentialsFile != null) {
+                LOG.info("Fetched credentials file path from environment variable");
+            }
+        }
+        else {
+            LOG.info("Fetched credentials file path from conf");
+        }
+        if (credentialsFile == null) {
+            LOG.info("Using default credentials file path");
+            return FileSystems.getDefault().getPath(SystemUtils.getUserHome().getPath(),".aws", "credentials");
+        }
+        else {
+            return FileSystems.getDefault().getPath(credentialsFile);
+        }
+    }
+
+    private static String getCredentialsName(Configuration conf) {
+        String profileName = conf.get(PROFILE_NAME, null);
+        if (profileName == null) {
+            profileName = SystemUtils.getEnvironmentVariable(PROFILE_ENV, null);
+            if (profileName == null) {
+                profileName = "default";
+                LOG.info("Using default profile name");
+            }
+            else {
+                LOG.info("Fetched profile name from environment variable");
+            }
+        }
+        else {
+            LOG.info("Fetched profile name from conf");
+        }
+        return profileName;
+    }
+
+    public ProfileAWSCredentialsProvider(URI uri, Configuration conf) {
+        super(uri, conf);
+        ProfileCredentialsProvider.Builder builder = ProfileCredentialsProvider.builder();
+        builder.profileName(getCredentialsName(conf))
+                .profileFile(ProfileFile.builder()
+                .content(getCredentialsPath(conf))
+                .type(ProfileFile.Type.CREDENTIALS)
+                .build());
+        pcp = builder.build();
+    }
+
+    public AwsCredentials resolveCredentials() {
+        return pcp.resolveCredentials();
+    }
+}

hadoop-tools/hadoop-aws/src/test/java/org/apache/hadoop/fs/s3a/TestS3AAWSCredentialsProvider.java

@@ -18,8 +18,11 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import java.io.IOException;
+import java.io.BufferedWriter;
+import java.io.File;
+import java.io.FileWriter;
 import java.io.InterruptedIOException;
+import java.io.IOException;
 import java.net.URI;
 import java.nio.file.AccessDeniedException;
 import java.util.ArrayList;
@@ -52,6 +55,7 @@
 import org.apache.hadoop.fs.s3a.auth.CredentialProviderListFactory;
 import org.apache.hadoop.fs.s3a.auth.IAMInstanceCredentialsProvider;
 import org.apache.hadoop.fs.s3a.auth.NoAuthWithAWSException;
+import org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider;
 import org.apache.hadoop.fs.s3a.auth.delegation.CountInvocationsProvider;
 import org.apache.hadoop.fs.s3a.impl.InstantiationIOException;
 import org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils;
@@ -139,6 +143,34 @@ public void testInstantiationChain() throws Throwable {
     assertCredentialProviders(expectedClasses, list);
   }
 
+  @Test
+  public void testProfileAWSCredentialsProvider() throws Throwable {
+    Configuration conf = new Configuration(false);
+    conf.set(AWS_CREDENTIALS_PROVIDER, ProfileAWSCredentialsProvider.NAME);
+    File tempFile = File.createTempFile("testcred", ".conf", new File("target"));
+    tempFile.deleteOnExit();
+    try (FileWriter fileWriter = new FileWriter(tempFile); BufferedWriter bufferedWriter = new BufferedWriter(fileWriter)) {
+      bufferedWriter.write("[default]\n"
+      + "aws_access_key_id = defaultaccesskeyid\n"
+      + "aws_secret_access_key = defaultsecretkeyid\n");
+      bufferedWriter.write("[nondefault]\n"
+              + "aws_access_key_id = nondefaultaccesskeyid\n"
+              + "aws_secret_access_key = nondefaultsecretkeyid\n");
+    }
+    conf.set(ProfileAWSCredentialsProvider.PROFILE_FILE, tempFile.getAbsolutePath());
+    URI testUri = new URI("s3a://bucket1");
+    AWSCredentialProviderList list = createAWSCredentialProviderList(testUri, conf);
+    assertCredentialProviders(Collections.singletonList(ProfileAWSCredentialsProvider.class), list);
+    AwsCredentials credentials = list.resolveCredentials();
+    Assertions.assertThat(credentials.accessKeyId()).isEqualTo("defaultaccesskeyid");
+    Assertions.assertThat(credentials.secretAccessKey()).isEqualTo("defaultsecretkeyid");
+    conf.set(ProfileAWSCredentialsProvider.PROFILE_NAME, "nondefault");
+    list = createAWSCredentialProviderList(testUri, conf);
+    credentials = list.resolveCredentials();
+    Assertions.assertThat(credentials.accessKeyId()).isEqualTo("nondefaultaccesskeyid");
+    Assertions.assertThat(credentials.secretAccessKey()).isEqualTo("nondefaultsecretkeyid");
+  }
+
   @Test
   public void testDefaultChain() throws Exception {
     URI uri1 = new URI("s3a://bucket1"), uri2 = new URI("s3a://bucket2");