From e9efed85c266427778e01bbd3f8aeb5b53c9e6c7 Mon Sep 17 00:00:00 2001
From: Cody Rose <rosecodym@gmail.com>
Date: Wed, 27 Sep 2023 16:57:47 -0400
Subject: [PATCH] Use S3 credentials waterfall (#1823)

This PR updates the S3 source to use explicitly configured credentials if they're available and follow the normal AWS credentials waterfall if they're not. This is irrespective of whether role assumption is configured. This changes the previous behavior, which was to use waterfall credentials only if role assumption was configured and explicitly configured credentials only when it was not.
---
 pkg/sources/s3/s3.go                  | 27 +++++++++++++--------------
 pkg/sources/s3/s3_integration_test.go |  7 ++-----
 2 files changed, 15 insertions(+), 19 deletions(-)

diff --git a/pkg/sources/s3/s3.go b/pkg/sources/s3/s3.go
index d95746378..f27f54501 100644
--- a/pkg/sources/s3/s3.go
+++ b/pkg/sources/s3/s3.go
@@ -130,20 +130,19 @@ func (s *Source) newClient(region, roleArn string) (*s3.S3, error) {
 	cfg.CredentialsChainVerboseErrors = aws.Bool(true)
 	cfg.Region = aws.String(region)
 
-	if roleArn == "" {
-		switch cred := s.conn.GetCredential().(type) {
-		case *sourcespb.S3_SessionToken:
-			cfg.Credentials = credentials.NewStaticCredentials(cred.SessionToken.Key, cred.SessionToken.Secret, cred.SessionToken.SessionToken)
-		case *sourcespb.S3_AccessKey:
-			cfg.Credentials = credentials.NewStaticCredentials(cred.AccessKey.Key, cred.AccessKey.Secret, "")
-		case *sourcespb.S3_Unauthenticated:
-			cfg.Credentials = credentials.AnonymousCredentials
-		case *sourcespb.S3_CloudEnvironment:
-			// Nothing needs to be done!
-		default:
-			return nil, errors.Errorf("invalid configuration given for %s source", s.name)
-		}
-	} else {
+	switch cred := s.conn.GetCredential().(type) {
+	case *sourcespb.S3_SessionToken:
+		cfg.Credentials = credentials.NewStaticCredentials(cred.SessionToken.Key, cred.SessionToken.Secret, cred.SessionToken.SessionToken)
+	case *sourcespb.S3_AccessKey:
+		cfg.Credentials = credentials.NewStaticCredentials(cred.AccessKey.Key, cred.AccessKey.Secret, "")
+	case *sourcespb.S3_Unauthenticated:
+		cfg.Credentials = credentials.AnonymousCredentials
+	default:
+		// In all other cases, the AWS SDK will follow its normal waterfall logic to pick up credentials (i.e. they can
+		// come from the environment or the credentials file or whatever else AWS gets up to).
+	}
+
+	if roleArn != "" {
 		sess, err := session.NewSession(cfg)
 		if err != nil {
 			return nil, err
diff --git a/pkg/sources/s3/s3_integration_test.go b/pkg/sources/s3/s3_integration_test.go
index 3764caee5..12d72376c 100644
--- a/pkg/sources/s3/s3_integration_test.go
+++ b/pkg/sources/s3/s3_integration_test.go
@@ -141,14 +141,11 @@ func TestSource_Validate(t *testing.T) {
 			var cancelOnce sync.Once
 			defer cancelOnce.Do(cancel)
 
-			// These are used by the tests that assume roles
-			t.Setenv("AWS_ACCESS_KEY_ID", s3key)
-			t.Setenv("AWS_SECRET_ACCESS_KEY", s3secret)
-
 			s := &Source{}
 
+			// As of this writing, credentials set in the environment or the on-disk credentials file also work, but I
+			// couldn't figure out how to write automated tests for those cases that weren't ugly as sin.
 			conn, err := anypb.New(&sourcespb.S3{
-				// These are used by the tests that don't assume roles
 				Credential: &sourcespb.S3_AccessKey{
 					AccessKey: &credentialspb.KeySecret{
 						Key:    s3key,