feat(iam auth): allow iam roles anywhere auth profile

.build-tools/builtin-authentication-profiles.yaml

@@ -29,7 +29,25 @@ aws:
         type: string
   - title: "AWS: Credentials from Environment Variables"
     description: Use AWS_ACCESS_KEY_ID and AWS_SECRET_ACCESS_KEY from the environment
-
+  - title: "AWS: IAM Roles Anywhere"
+    description: Use X.509 certificates to establish trust between AWS and your AWS account and the Dapr cluster using AWS IAM Roles Anywhere.
+    metadata:
+      - name: trustAnchorArn
+        description: |
+          ARN of the AWS Trust Anchor in the AWS account granting trust to the Dapr Certificate Authority.
+        example: arn:aws:rolesanywhere:us-west-1:012345678910:trust-anchor/01234568-0123-0123-0123-012345678901
+        required: true
+      - name: trustProfileArn
+        description: |
+          ARN of the AWS IAM Profile in the trusting AWS account.
+        example: arn:aws:rolesanywhere:us-west-1:012345678910:profile/01234568-0123-0123-0123-012345678901
+        required: true
+      - name: assumeRoleArn
+        description: |
+          ARN of the AWS IAM role to assume in the trusting AWS account.
+        example: arn:aws:iam:012345678910:role/exampleIAMRoleName
+        required: true
+
 azuread:
   - title: "Azure AD: Managed identity"
     description: Authenticate using Azure AD and a managed identity.

bindings/aws/dynamodb/dynamodb.go

@@ -31,9 +31,9 @@ import (
 
 // DynamoDB allows performing stateful operations on AWS DynamoDB.
 type DynamoDB struct {
-	client *dynamodb.DynamoDB
-	table  string
-	logger logger.Logger
+	authProvider awsAuth.Provider
+	table        string
+	logger       logger.Logger
 }
 
 type dynamoDBMetadata struct {
@@ -51,18 +51,27 @@ func NewDynamoDB(logger logger.Logger) bindings.OutputBinding {
 }
 
 // Init performs connection parsing for DynamoDB.
-func (d *DynamoDB) Init(_ context.Context, metadata bindings.Metadata) error {
+func (d *DynamoDB) Init(ctx context.Context, metadata bindings.Metadata) error {
 	meta, err := d.getDynamoDBMetadata(metadata)
 	if err != nil {
 		return err
 	}
 
-	client, err := d.getClient(meta)
+	opts := awsAuth.Options{
+		Logger:       d.logger,
+		Properties:   metadata.Properties,
+		Region:       meta.Region,
+		Endpoint:     meta.Endpoint,
+		AccessKey:    meta.AccessKey,
+		SecretKey:    meta.SecretKey,
+		SessionToken: meta.SessionToken,
+	}
+
+	provider, err := awsAuth.NewProvider(ctx, opts, awsAuth.GetConfig(opts))
 	if err != nil {
 		return err
 	}
-
-	d.client = client
+	d.authProvider = provider
 	d.table = meta.Table
 
 	return nil
@@ -84,7 +93,7 @@ func (d *DynamoDB) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*bi
 		return nil, err
 	}
 
-	_, err = d.client.PutItemWithContext(ctx, &dynamodb.PutItemInput{
+	_, err = d.authProvider.DynamoDB().DynamoDB.PutItemWithContext(ctx, &dynamodb.PutItemInput{
 		Item:      item,
 		TableName: aws.String(d.table),
 	})
@@ -105,16 +114,6 @@ func (d *DynamoDB) getDynamoDBMetadata(spec bindings.Metadata) (*dynamoDBMetadat
 	return &meta, nil
 }
 
-func (d *DynamoDB) getClient(metadata *dynamoDBMetadata) (*dynamodb.DynamoDB, error) {
-	sess, err := awsAuth.GetClient(metadata.AccessKey, metadata.SecretKey, metadata.SessionToken, metadata.Region, metadata.Endpoint)
-	if err != nil {
-		return nil, err
-	}
-	c := dynamodb.New(sess)
-
-	return c, nil
-}
-
 // GetComponentMetadata returns the metadata of the component.
 func (d *DynamoDB) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 	metadataStruct := dynamoDBMetadata{}
@@ -123,5 +122,5 @@ func (d *DynamoDB) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 }
 
 func (d *DynamoDB) Close() error {
-	return nil
+	return d.authProvider.Close()
 }

bindings/aws/kinesis/kinesis.go

@@ -27,7 +27,6 @@ import (
 	"github.com/aws/aws-sdk-go/service/kinesis"
 	"github.com/cenkalti/backoff/v4"
 	"github.com/google/uuid"
-	"github.com/vmware/vmware-go-kcl/clientlibrary/config"
 	"github.com/vmware/vmware-go-kcl/clientlibrary/interfaces"
 	"github.com/vmware/vmware-go-kcl/clientlibrary/worker"
 
@@ -40,15 +39,16 @@ import (
 
 // AWSKinesis allows receiving and sending data to/from AWS Kinesis stream.
 type AWSKinesis struct {
-	client   *kinesis.Kinesis
-	metadata *kinesisMetadata
+	authProvider awsAuth.Provider
+	metadata     *kinesisMetadata
 
-	worker       *worker.Worker
-	workerConfig *config.KinesisClientLibConfiguration
+	worker *worker.Worker
 
-	streamARN   *string
-	consumerARN *string
-	logger      logger.Logger
+	streamName   string
+	consumerName string
+	consumerARN  *string
+	logger       logger.Logger
+	consumerMode string
 
 	closed  atomic.Bool
 	closeCh chan struct{}
@@ -112,30 +112,25 @@ func (a *AWSKinesis) Init(ctx context.Context, metadata bindings.Metadata) error
 		return fmt.Errorf("%s invalid \"mode\" field %s", "aws.kinesis", m.KinesisConsumerMode)
 	}
 
-	client, err := a.getClient(m)
-	if err != nil {
-		return err
-	}
+	a.consumerMode = m.KinesisConsumerMode
+	a.streamName = m.StreamName
+	a.consumerName = m.ConsumerName
+	a.metadata = m
 
-	streamName := aws.String(m.StreamName)
-	stream, err := client.DescribeStreamWithContext(ctx, &kinesis.DescribeStreamInput{
-		StreamName: streamName,
-	})
+	opts := awsAuth.Options{
+		Logger:       a.logger,
+		Properties:   metadata.Properties,
+		Region:       m.Region,
+		AccessKey:    m.AccessKey,
+		SecretKey:    m.SecretKey,
+		SessionToken: "",
+	}
+	// extra configs needed per component type
+	provider, err := awsAuth.NewProvider(ctx, opts, awsAuth.GetConfig(opts))
 	if err != nil {
 		return err
 	}
-
-	if m.KinesisConsumerMode == SharedThroughput {
-		kclConfig := config.NewKinesisClientLibConfigWithCredential(m.ConsumerName,
-			m.StreamName, m.Region, m.ConsumerName,
-			client.Config.Credentials)
-		a.workerConfig = kclConfig
-	}
-
-	a.streamARN = stream.StreamDescription.StreamARN
-	a.metadata = m
-	a.client = client
-
+	a.authProvider = provider
 	return nil
 }
 
@@ -148,7 +143,7 @@ func (a *AWSKinesis) Invoke(ctx context.Context, req *bindings.InvokeRequest) (*
 	if partitionKey == "" {
 		partitionKey = uuid.New().String()
 	}
-	_, err := a.client.PutRecordWithContext(ctx, &kinesis.PutRecordInput{
+	_, err := a.authProvider.Kinesis().Kinesis.PutRecordWithContext(ctx, &kinesis.PutRecordInput{
 		StreamName:   &a.metadata.StreamName,
 		Data:         req.Data,
 		PartitionKey: &partitionKey,
@@ -161,16 +156,15 @@ func (a *AWSKinesis) Read(ctx context.Context, handler bindings.Handler) (err er
 	if a.closed.Load() {
 		return errors.New("binding is closed")
 	}
-
 	if a.metadata.KinesisConsumerMode == SharedThroughput {
-		a.worker = worker.NewWorker(a.recordProcessorFactory(ctx, handler), a.workerConfig)
+		a.worker = worker.NewWorker(a.recordProcessorFactory(ctx, handler), a.authProvider.Kinesis().WorkerCfg(ctx, a.streamName, a.consumerName, a.consumerMode))
 		err = a.worker.Start()
 		if err != nil {
 			return err
 		}
 	} else if a.metadata.KinesisConsumerMode == ExtendedFanout {
 		var stream *kinesis.DescribeStreamOutput
-		stream, err = a.client.DescribeStream(&kinesis.DescribeStreamInput{StreamName: &a.metadata.StreamName})
+		stream, err = a.authProvider.Kinesis().Kinesis.DescribeStream(&kinesis.DescribeStreamInput{StreamName: &a.metadata.StreamName})
 		if err != nil {
 			return err
 		}
@@ -180,6 +174,10 @@ func (a *AWSKinesis) Read(ctx context.Context, handler bindings.Handler) (err er
 		}
 	}
 
+	stream, err := a.authProvider.Kinesis().Stream(ctx, a.streamName)
+	if err != nil {
+		return fmt.Errorf("failed to get kinesis stream arn: %v", err)
+	}
 	// Wait for context cancelation then stop
 	a.wg.Add(1)
 	go func() {
@@ -191,7 +189,7 @@ func (a *AWSKinesis) Read(ctx context.Context, handler bindings.Handler) (err er
 		if a.metadata.KinesisConsumerMode == SharedThroughput {
 			a.worker.Shutdown()
 		} else if a.metadata.KinesisConsumerMode == ExtendedFanout {
-			a.deregisterConsumer(a.streamARN, a.consumerARN)
+			a.deregisterConsumer(ctx, stream, a.consumerARN)
 		}
 	}()
 
@@ -226,8 +224,7 @@ func (a *AWSKinesis) Subscribe(ctx context.Context, streamDesc kinesis.StreamDes
 					return
 				default:
 				}
-
-				sub, err := a.client.SubscribeToShardWithContext(ctx, &kinesis.SubscribeToShardInput{
+				sub, err := a.authProvider.Kinesis().Kinesis.SubscribeToShardWithContext(ctx, &kinesis.SubscribeToShardInput{
 					ConsumerARN:      consumerARN,
 					ShardId:          s.ShardId,
 					StartingPosition: &kinesis.StartingPosition{Type: aws.String(kinesis.ShardIteratorTypeLatest)},
@@ -269,14 +266,14 @@ func (a *AWSKinesis) Close() error {
 		close(a.closeCh)
 	}
 	a.wg.Wait()
-	return nil
+	return a.authProvider.Close()
 }
 
 func (a *AWSKinesis) ensureConsumer(ctx context.Context, streamARN *string) (*string, error) {
 	// Only set timeout on consumer call.
 	conCtx, cancel := context.WithTimeout(ctx, 30*time.Second)
 	defer cancel()
-	consumer, err := a.client.DescribeStreamConsumerWithContext(conCtx, &kinesis.DescribeStreamConsumerInput{
+	consumer, err := a.authProvider.Kinesis().Kinesis.DescribeStreamConsumerWithContext(conCtx, &kinesis.DescribeStreamConsumerInput{
 		ConsumerName: &a.metadata.ConsumerName,
 		StreamARN:    streamARN,
 	})
@@ -288,7 +285,7 @@ func (a *AWSKinesis) ensureConsumer(ctx context.Context, streamARN *string) (*st
 }
 
 func (a *AWSKinesis) registerConsumer(ctx context.Context, streamARN *string) (*string, error) {
-	consumer, err := a.client.RegisterStreamConsumerWithContext(ctx, &kinesis.RegisterStreamConsumerInput{
+	consumer, err := a.authProvider.Kinesis().Kinesis.RegisterStreamConsumerWithContext(ctx, &kinesis.RegisterStreamConsumerInput{
 		ConsumerName: &a.metadata.ConsumerName,
 		StreamARN:    streamARN,
 	})
@@ -307,11 +304,11 @@ func (a *AWSKinesis) registerConsumer(ctx context.Context, streamARN *string) (*
 	return consumer.Consumer.ConsumerARN, nil
 }
 
-func (a *AWSKinesis) deregisterConsumer(streamARN *string, consumerARN *string) error {
+func (a *AWSKinesis) deregisterConsumer(ctx context.Context, streamARN *string, consumerARN *string) error {
 	if a.consumerARN != nil {
 		// Use a background context because the running context may have been canceled already
 		ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
-		_, err := a.client.DeregisterStreamConsumerWithContext(ctx, &kinesis.DeregisterStreamConsumerInput{
+		_, err := a.authProvider.Kinesis().Kinesis.DeregisterStreamConsumerWithContext(ctx, &kinesis.DeregisterStreamConsumerInput{
 			ConsumerARN:  consumerARN,
 			StreamARN:    streamARN,
 			ConsumerName: &a.metadata.ConsumerName,
@@ -342,7 +339,7 @@ func (a *AWSKinesis) waitUntilConsumerExists(ctx aws.Context, input *kinesis.Des
 				tmp := *input
 				inCpy = &tmp
 			}
-			req, _ := a.client.DescribeStreamConsumerRequest(inCpy)
+			req, _ := a.authProvider.Kinesis().Kinesis.DescribeStreamConsumerRequest(inCpy)
 			req.SetContext(ctx)
 			req.ApplyOptions(opts...)
 
@@ -354,16 +351,6 @@ func (a *AWSKinesis) waitUntilConsumerExists(ctx aws.Context, input *kinesis.Des
 	return w.WaitWithContext(ctx)
 }
 
-func (a *AWSKinesis) getClient(metadata *kinesisMetadata) (*kinesis.Kinesis, error) {
-	sess, err := awsAuth.GetClient(metadata.AccessKey, metadata.SecretKey, metadata.SessionToken, metadata.Region, metadata.Endpoint)
-	if err != nil {
-		return nil, err
-	}
-	k := kinesis.New(sess)
-
-	return k, nil
-}
-
 func (a *AWSKinesis) parseMetadata(meta bindings.Metadata) (*kinesisMetadata, error) {
 	var m kinesisMetadata
 	err := kitmd.DecodeMetadata(meta.Properties, &m)