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,56 @@
+package org.apache.hadoop.fs.s3a.auth;
+
+import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import software.amazon.awssdk.auth.credentials.ProfileCredentialsProvider;
+import software.amazon.awssdk.profiles.ProfileFile;
+
+import org.apache.hadoop.classification.InterfaceAudience;
+import org.apache.hadoop.classification.InterfaceStability;
+
+import org.apache.commons.lang3.SystemUtils;
+import org.apache.hadoop.conf.Configuration;
+
+import java.net.URI;
+import java.nio.file.FileSystems;
+import java.nio.file.Path;
+
+@InterfaceAudience.Public
+@InterfaceStability.Evolving
+public class ProfileAWSCredentialsProvider extends AbstractAWSCredentialProvider {
+    public static final String NAME
+            = "org.apache.hadoop.fs.s3a.auth.ProfileAWSCredentialsProvider";
+    public static final String PROFILE_FILE = "fs.s3a.auth.profile.file";
+    public static final String PROFILE_NAME = "fs.s3a.auth.profile.name";
+
+    private final ProfileCredentialsProvider pcp;
+
+    private static Path getCredentialsPath(Configuration conf) {
+        String credentialsFile = conf.get(PROFILE_FILE, null);
+        if (credentialsFile == null) {
+            credentialsFile = SystemUtils.getEnvironmentVariable("AWS_SHARED_CREDENTIALS_FILE", null);
+        }
+        Path path = (credentialsFile == null) ?
+                FileSystems.getDefault().getPath(SystemUtils.getUserHome().getPath(),".aws","credentials")
+                : FileSystems.getDefault().getPath(credentialsFile);
+        return path;
+    }
+
+    private static String getCredentialsName(Configuration conf) {
+        String profileName = conf.get(PROFILE_NAME, null);
+        if (profileName == null) {
+            profileName = SystemUtils.getEnvironmentVariable("AWS_PROFILE", "default");
+        }
+        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,7 @@
 
 package org.apache.hadoop.fs.s3a;
 
-import java.io.IOException;
-import java.io.InterruptedIOException;
+import java.io.*;
 import java.net.URI;
 import java.nio.file.AccessDeniedException;
 import java.util.ArrayList;
@@ -35,6 +34,7 @@
 import java.util.stream.Collectors;
 import javax.annotation.Nullable;
 
+import org.apache.hadoop.fs.s3a.auth.*;
 import org.assertj.core.api.Assertions;
 import org.junit.Test;
 import org.slf4j.Logger;
@@ -47,11 +47,6 @@
 
 import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
-import org.apache.hadoop.fs.s3a.auth.AbstractSessionCredentialsProvider;
-import org.apache.hadoop.fs.s3a.auth.AssumedRoleCredentialProvider;
-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.delegation.CountInvocationsProvider;
 import org.apache.hadoop.fs.s3a.impl.InstantiationIOException;
 import org.apache.hadoop.fs.s3a.test.PublicDatasetTestUtils;
@@ -139,6 +134,32 @@ 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);
+    try (FileWriter fileWriter = new FileWriter("testcred"); 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, "testcred");
+    URI testUri = new URI("s3a://bucket1");
+    AWSCredentialProviderList list = createAWSCredentialProviderList(testUri, conf);
+    assertCredentialProviders(Collections.singletonList(ProfileAWSCredentialsProvider.class), list);
+    AwsCredentials credentials = list.resolveCredentials();
+    assertEquals("defaultaccesskeyid", credentials.accessKeyId());
+    assertEquals("defaultsecretkeyid", credentials.secretAccessKey());
+    conf.set(ProfileAWSCredentialsProvider.PROFILE_NAME, "nondefault");
+    list = createAWSCredentialProviderList(testUri, conf);
+    credentials = list.resolveCredentials();
+    assertEquals("nondefaultaccesskeyid", credentials.accessKeyId());
+    assertEquals("nondefaultsecretkeyid", credentials.secretAccessKey());
+  }
+
   @Test
   public void testDefaultChain() throws Exception {
     URI uri1 = new URI("s3a://bucket1"), uri2 = new URI("s3a://bucket2");