Enable creating AWS sessions without providing explicit credentials f… (

#300)

* Enable creating AWS sessions without providing explicit credentials for AWS Gateways

AWS Gateways running on pods configured with IAM roles can use event sources without specifying access credentials.
The AWS Session is created using the IAM role present on the pod and has permissions associated with the IAM role.

* Add appropriate comments for example without credentials event source in aws sns

examples/event-sources/aws-sns.yaml

@@ -56,3 +56,13 @@ data:
       name: aws-secret
       key: secret
     region: "us-east-1"
+
+  example-without-credentials: |-
+    # If AWS access credentials are already present on the Pod's IAM role running the Gateway, 
+    # the AWS session will utilize the existing config and hence we do not need to provide explicit credentials.
+    topicArn: "topic-arn"
+    hook:
+     endpoint: "/"
+     port: "13000"
+     url: "http://mysecondfakeurl.fake"
+    region: "us-east-1"

examples/event-sources/aws-sqs.yaml

@@ -40,3 +40,8 @@ data:
     region: "us-east-1"
     queue: "my-fake-queue-2"
     waitTimeSeconds: 20
+
+  example-3: |-
+    region: "us-east-1"
+    queue: "my-fake-queue-2"
+    waitTimeSeconds: 20

gateways/common/aws.go

@@ -48,3 +48,9 @@ func GetAWSSession(creds *credentials.Credentials, region string) (*session.Sess
 		Credentials: creds,
 	})
 }
+
+func GetAWSSessionWithoutCreds(region string) (*session.Session, error) {
+	return session.NewSession(&aws.Config{
+		Region: &region,
+	})
+}

gateways/common/aws_test.go

@@ -75,4 +75,12 @@ func TestAWS(t *testing.T) {
 			convey.So(session, convey.ShouldNotBeNil)
 		})
 	})
+
+	convey.Convey("create AWS credential using already present config/IAM role", t, func() {
+		convey.Convey("Get a new aws session", func() {
+			session, err := GetAWSSessionWithoutCreds("mock-region")
+			convey.So(err, convey.ShouldBeNil)
+			convey.So(session, convey.ShouldNotBeNil)
+		})
+	})
 }

gateways/community/aws-sns/config_test.go

@@ -17,8 +17,9 @@ limitations under the License.
 package aws_sns
 
 import (
-	"github.com/smartystreets/goconvey/convey"
 	"testing"
+
+	"github.com/smartystreets/goconvey/convey"
 )
 
 var es = `
@@ -36,6 +37,15 @@ secretKey:
     name: sns
 `
 
+var esWithoutCreds = `
+hook:
+ endpoint: "/test"
+ port: "8080"
+ url: "myurl/test"
+topicArn: "test-arn"
+region: "us-east-1"
+`
+
 func TestParseConfig(t *testing.T) {
 	convey.Convey("Given a aws-sns event source, parse it", t, func() {
 		ps, err := parseEventSource(es)
@@ -44,4 +54,12 @@ func TestParseConfig(t *testing.T) {
 		_, ok := ps.(*snsEventSource)
 		convey.So(ok, convey.ShouldEqual, true)
 	})
+
+	convey.Convey("Given a aws-sns event source without credentials, parse it", t, func() {
+		ps, err := parseEventSource(esWithoutCreds)
+		convey.So(err, convey.ShouldBeNil)
+		convey.So(ps, convey.ShouldNotBeNil)
+		_, ok := ps.(*snsEventSource)
+		convey.So(ok, convey.ShouldEqual, true)
+	})
 }

gateways/community/aws-sns/start.go

@@ -24,6 +24,7 @@ import (
 	"github.com/argoproj/argo-events/common"
 	"github.com/argoproj/argo-events/gateways"
 	gwcommon "github.com/argoproj/argo-events/gateways/common"
+	"github.com/aws/aws-sdk-go/aws/session"
 	snslib "github.com/aws/aws-sdk-go/service/sns"
 	"github.com/ghodss/yaml"
 )
@@ -113,14 +114,27 @@ func (rc *RouteConfig) PostStart() error {
 	logger.Info("subscribing to sns topic")
 
 	sc := rc.snses
-	creds, err := gwcommon.GetAWSCreds(rc.clientset, rc.namespace, sc.AccessKey, sc.SecretKey)
-	if err != nil {
-		return fmt.Errorf("failed to get aws credentials. err: %+v", err)
-	}
+	var awsSession *session.Session
 
-	awsSession, err := gwcommon.GetAWSSession(creds, sc.Region)
-	if err != nil {
-		return fmt.Errorf("failed to create aws session. err: %+v", err)
+	if sc.AccessKey == nil && sc.SecretKey == nil {
+		awsSessionWithoutCreds, err := gwcommon.GetAWSSessionWithoutCreds(sc.Region)
+		if err != nil {
+			return fmt.Errorf("failed to create aws session. err: %+v", err)
+		}
+
+		awsSession = awsSessionWithoutCreds
+	} else {
+		creds, err := gwcommon.GetAWSCreds(rc.clientset, rc.namespace, sc.AccessKey, sc.SecretKey)
+		if err != nil {
+			return fmt.Errorf("failed to create aws session. err: %+v", err)
+		}
+
+		awsSessionWithCreds, err := gwcommon.GetAWSSession(creds, sc.Region)
+		if err != nil {
+			return fmt.Errorf("failed to create aws session. err: %+v", err)
+		}
+
+		awsSession = awsSessionWithCreds
 	}
 
 	rc.session = snslib.New(awsSession)

gateways/community/aws-sns/start_test.go

@@ -106,5 +106,15 @@ func TestAWSSNS(t *testing.T) {
 			err = rc.PostStop()
 			convey.So(err, convey.ShouldNotBeNil)
 		})
+
+		psWithoutCreds, err2 := parseEventSource(esWithoutCreds)
+		convey.So(err2, convey.ShouldBeNil)
+
+		rc.snses = psWithoutCreds.(*snsEventSource)
+
+		convey.Convey("Run post activate on event source without credentials", func() {
+			err := rc.PostStart()
+			convey.So(err, convey.ShouldNotBeNil)
+		})
 	})
 }

gateways/community/aws-sns/validate.go

@@ -19,6 +19,7 @@ package aws_sns
 import (
 	"context"
 	"fmt"
+
 	"github.com/argoproj/argo-events/gateways"
 	gwcommon "github.com/argoproj/argo-events/gateways/common"
 )
@@ -39,11 +40,5 @@ func validateSNSConfig(config interface{}) error {
 	if sc.Region == "" {
 		return fmt.Errorf("must specify region")
 	}
-	if sc.AccessKey == nil {
-		return fmt.Errorf("must specify access key")
-	}
-	if sc.SecretKey == nil {
-		return fmt.Errorf("must specify secret key")
-	}
 	return gwcommon.ValidateWebhook(sc.Hook)
 }

gateways/community/aws-sqs/config_test.go

@@ -34,6 +34,12 @@ queue: "test-queue"
 waitTimeSeconds: 10
 `
 
+var esWithoutCreds = `
+region: "us-east-1"
+queue: "test-queue"
+waitTimeSeconds: 10
+`
+
 func TestParseConfig(t *testing.T) {
 	convey.Convey("Given a aws-sqsEventSource event source, parse it", t, func() {
 		ps, err := parseEventSource(es)
@@ -42,4 +48,12 @@ func TestParseConfig(t *testing.T) {
 		_, ok := ps.(*sqsEventSource)
 		convey.So(ok, convey.ShouldEqual, true)
 	})
+
+	convey.Convey("Given a aws-sqsEventSource event source without AWS credentials, parse it", t, func() {
+		ps, err := parseEventSource(esWithoutCreds)
+		convey.So(err, convey.ShouldBeNil)
+		convey.So(ps, convey.ShouldNotBeNil)
+		_, ok := ps.(*sqsEventSource)
+		convey.So(ok, convey.ShouldEqual, true)
+	})
 }

gateways/community/aws-sqs/start.go

@@ -21,6 +21,7 @@ import (
 	"github.com/argoproj/argo-events/gateways"
 	gwcommon "github.com/argoproj/argo-events/gateways/common"
 	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/aws/session"
 	sqslib "github.com/aws/aws-sdk-go/service/sqs"
 )
 
@@ -48,16 +49,30 @@ func (ese *SQSEventSourceExecutor) StartEventSource(eventSource *gateways.EventS
 
 // listenEvents fires an event when interval completes and item is processed from queue.
 func (ese *SQSEventSourceExecutor) listenEvents(s *sqsEventSource, eventSource *gateways.EventSource, dataCh chan []byte, errorCh chan error, doneCh chan struct{}) {
-	creds, err := gwcommon.GetAWSCreds(ese.Clientset, ese.Namespace, s.AccessKey, s.SecretKey)
-	if err != nil {
-		errorCh <- err
-		return
-	}
+	var awsSession *session.Session
 
-	awsSession, err := gwcommon.GetAWSSession(creds, s.Region)
-	if err != nil {
-		errorCh <- err
-		return
+	if s.AccessKey == nil && s.SecretKey == nil {
+		awsSessionWithoutCreds, err := gwcommon.GetAWSSessionWithoutCreds(s.Region)
+		if err != nil {
+			errorCh <- err
+			return
+		}
+
+		awsSession = awsSessionWithoutCreds
+	} else {
+		creds, err := gwcommon.GetAWSCreds(ese.Clientset, ese.Namespace, s.AccessKey, s.SecretKey)
+		if err != nil {
+			errorCh <- err
+			return
+		}
+
+		awsSessionWithCreds, err := gwcommon.GetAWSSession(creds, s.Region)
+		if err != nil {
+			errorCh <- err
+			return
+		}
+
+		awsSession = awsSessionWithCreds
 	}
 
 	sqsClient := sqslib.New(awsSession)

gateways/community/aws-sqs/start_test.go

@@ -56,4 +56,35 @@ func TestListenEvents(t *testing.T) {
 		err = <-errorCh2
 		convey.So(err, convey.ShouldNotBeNil)
 	})
+
+	convey.Convey("Given an event source without AWS credentials, listen to events", t, func() {
+		ps, err := parseEventSource(esWithoutCreds)
+		convey.So(err, convey.ShouldBeNil)
+		convey.So(ps, convey.ShouldNotBeNil)
+
+		ese := &SQSEventSourceExecutor{
+			Clientset: fake.NewSimpleClientset(),
+			Namespace: "fake",
+			Log:       common.NewArgoEventsLogger(),
+		}
+
+		dataCh := make(chan []byte)
+		errorCh := make(chan error)
+		doneCh := make(chan struct{}, 1)
+		errorCh2 := make(chan error)
+
+		go func() {
+			err := <-errorCh
+			errorCh2 <- err
+		}()
+
+		ese.listenEvents(ps.(*sqsEventSource), &gateways.EventSource{
+			Name: "fake",
+			Data: es,
+			Id:   "1234",
+		}, dataCh, errorCh, doneCh)
+
+		err = <-errorCh2
+		convey.So(err, convey.ShouldNotBeNil)
+	})
 }

gateways/community/aws-sqs/validate.go

@@ -40,12 +40,6 @@ func validateSQSConfig(config interface{}) error {
 	if sc.Region == "" {
 		return fmt.Errorf("must specify region")
 	}
-	if sc.AccessKey == nil {
-		return fmt.Errorf("must specify access key")
-	}
-	if sc.SecretKey == nil {
-		return fmt.Errorf("must specify secret key")
-	}
 	if sc.Queue == "" {
 		return fmt.Errorf("must specify queue name")
 	}