Merge branch 'main' into dapr-state-store-clickhouse

Author: dapr-bot

.build-tools/go.mod

@@ -1,6 +1,6 @@
 module github.com/dapr/components-contrib/build-tools
 
-go 1.24.1
+go 1.24.4
 
 require (
 	github.com/dapr/components-contrib v0.0.0

bindings/aws/s3/s3.go

@@ -24,17 +24,20 @@ import (
 	"net/http"
 	"os"
 	"reflect"
+	"slices"
 	"strings"
 	"time"
 
+	"github.com/aws/aws-sdk-go-v2/service/s3/types"
 	"github.com/aws/aws-sdk-go/aws"
-	"github.com/aws/aws-sdk-go/aws/awserr"
-	"github.com/aws/aws-sdk-go/service/s3"
-	"github.com/aws/aws-sdk-go/service/s3/s3manager"
+	awsCommon "github.com/dapr/components-contrib/common/aws"
+	awsCommonAuth "github.com/dapr/components-contrib/common/aws/auth"
+
+	"github.com/aws/aws-sdk-go-v2/feature/s3/manager"
+	"github.com/aws/aws-sdk-go-v2/service/s3"
 	"github.com/google/uuid"
 
 	"github.com/dapr/components-contrib/bindings"
-	awsAuth "github.com/dapr/components-contrib/common/authentication/aws"
 	commonutils "github.com/dapr/components-contrib/common/utils"
 	"github.com/dapr/components-contrib/metadata"
 	"github.com/dapr/kit/logger"
@@ -60,9 +63,12 @@ const (
 
 // AWSS3 is a binding for an AWS S3 storage bucket.
 type AWSS3 struct {
-	metadata     *s3Metadata
-	authProvider awsAuth.Provider
-	logger       logger.Logger
+	metadata        *s3Metadata
+	logger          logger.Logger
+	s3Client        *s3.Client
+	s3Uploader      *manager.Uploader
+	s3Downloader    *manager.Downloader
+	s3PresignClient *s3.PresignClient
 }
 
 type s3Metadata struct {
@@ -106,10 +112,42 @@ func NewAWSS3(logger logger.Logger) bindings.OutputBinding {
 	return &AWSS3{logger: logger}
 }
 
-func (s *AWSS3) getAWSConfig(opts awsAuth.Options) *aws.Config {
-	cfg := awsAuth.GetConfig(opts).WithS3ForcePathStyle(s.metadata.ForcePathStyle).WithDisableSSL(s.metadata.DisableSSL)
+// Init does metadata parsing and connection creation.
+func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
+	m, err := s.parseMetadata(metadata)
+	if err != nil {
+		return err
+	}
+	s.metadata = m
+
+	authOpts := awsCommonAuth.Options{
+		Logger: s.logger,
+
+		Properties: metadata.Properties,
+
+		Region:       m.Region,
+		Endpoint:     m.Endpoint,
+		AccessKey:    m.AccessKey,
+		SecretKey:    m.SecretKey,
+		SessionToken: m.SessionToken,
+	}
+
+	var configOptions []awsCommon.ConfigOption
+
+	var s3Options []func(options *s3.Options)
+
+	if s.metadata.DisableSSL {
+		s3Options = append(s3Options, func(options *s3.Options) {
+			options.EndpointOptions.DisableHTTPS = true
+		})
+	}
+
+	if !s.metadata.ForcePathStyle {
+		s3Options = append(s3Options, func(options *s3.Options) {
+			options.UsePathStyle = true
+		})
+	}
 
-	// Use a custom HTTP client to allow self-signed certs
 	if s.metadata.InsecureSSL {
 		customTransport := http.DefaultTransport.(*http.Transport).Clone()
 		customTransport.TLSClientConfig = &tls.Config{
@@ -119,44 +157,27 @@ func (s *AWSS3) getAWSConfig(opts awsAuth.Options) *aws.Config {
 		client := &http.Client{
 			Transport: customTransport,
 		}
-		cfg = cfg.WithHTTPClient(client)
+		configOptions = append(configOptions, awsCommon.WithHTTPClient(client))
 
 		s.logger.Infof("aws s3: you are using 'insecureSSL' to skip server config verify which is unsafe!")
 	}
-	return cfg
-}
 
-// Init does metadata parsing and connection creation.
-func (s *AWSS3) Init(ctx context.Context, metadata bindings.Metadata) error {
-	m, err := s.parseMetadata(metadata)
+	awsConfig, err := awsCommon.NewConfig(ctx, authOpts, configOptions...)
 	if err != nil {
-		return err
+		return fmt.Errorf("s3 binding error: failed to create AWS config: %w", err)
 	}
-	s.metadata = m
 
-	opts := awsAuth.Options{
-		Logger:       s.logger,
-		Properties:   metadata.Properties,
-		Region:       m.Region,
-		Endpoint:     m.Endpoint,
-		AccessKey:    m.AccessKey,
-		SecretKey:    m.SecretKey,
-		SessionToken: m.SessionToken,
-	}
-	// extra configs needed per component type
-	provider, err := awsAuth.NewProvider(ctx, opts, s.getAWSConfig(opts))
-	if err != nil {
-		return err
-	}
-	s.authProvider = provider
+	s.s3Client = s3.NewFromConfig(awsConfig, s3Options...)
+
+	s.s3Uploader = manager.NewUploader(s.s3Client)
+	s.s3Downloader = manager.NewDownloader(s.s3Client)
+
+	s.s3PresignClient = s3.NewPresignClient(s.s3Client)
 
 	return nil
 }
 
 func (s *AWSS3) Close() error {
-	if s.authProvider != nil {
-		return s.authProvider.Close()
-	}
 	return nil
 }
 
@@ -215,19 +236,25 @@ func (s *AWSS3) create(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 		r = b64.NewDecoder(b64.StdEncoding, r)
 	}
 
-	var storageClass *string
+	var storageClass types.StorageClass
 	if metadata.StorageClass != "" {
-		storageClass = aws.String(metadata.StorageClass)
+		// assert storageclass exists in the types.storageclass.values() slice
+		storageClass = types.StorageClass(strings.ToUpper(metadata.StorageClass))
+		if !slices.Contains(storageClass.Values(), storageClass) {
+			return nil, fmt.Errorf("s3 binding error: invalid storage class '%s' provided", metadata.StorageClass)
+		}
 	}
 
-	resultUpload, err := s.authProvider.S3().Uploader.UploadWithContext(ctx, &s3manager.UploadInput{
+	s3UploaderPutObjectInput := &s3.PutObjectInput{
 		Bucket:       ptr.Of(metadata.Bucket),
 		Key:          ptr.Of(key),
 		Body:         r,
 		ContentType:  contentType,
 		StorageClass: storageClass,
 		Tagging:      tagging,
-	})
+	}
+
+	resultUpload, err := s.s3Uploader.Upload(ctx, s3UploaderPutObjectInput)
 	if err != nil {
 		return nil, fmt.Errorf("s3 binding error: uploading failed: %w", err)
 	}
@@ -296,16 +323,21 @@ func (s *AWSS3) presignObject(ctx context.Context, bucket, key, ttl string) (str
 	if err != nil {
 		return "", fmt.Errorf("s3 binding error: cannot parse duration %s: %w", ttl, err)
 	}
-	objReq, _ := s.authProvider.S3().S3.GetObjectRequest(&s3.GetObjectInput{
+	s3GetObjectInput := &s3.GetObjectInput{
 		Bucket: ptr.Of(bucket),
 		Key:    ptr.Of(key),
-	})
-	url, err := objReq.Presign(d)
+	}
+
+	presignedObjectRequest, err := s.s3PresignClient.PresignGetObject(
+		ctx,
+		s3GetObjectInput,
+		s3.WithPresignExpires(d),
+	)
 	if err != nil {
 		return "", fmt.Errorf("s3 binding error: failed to presign URL: %w", err)
 	}
 
-	return url, nil
+	return presignedObjectRequest.URL, nil
 }
 
 func (s *AWSS3) get(ctx context.Context, req *bindings.InvokeRequest) (*bindings.InvokeResponse, error) {
@@ -320,16 +352,16 @@ func (s *AWSS3) get(ctx context.Context, req *bindings.InvokeRequest) (*bindings
 	}
 
 	buff := &aws.WriteAtBuffer{}
-	_, err = s.authProvider.S3().Downloader.DownloadWithContext(ctx,
+	_, err = s.s3Downloader.Download(ctx,
 		buff,
 		&s3.GetObjectInput{
 			Bucket: ptr.Of(s.metadata.Bucket),
 			Key:    ptr.Of(key),
 		},
 	)
 	if err != nil {
-		var awsErr awserr.Error
-		if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey {
+		var awsErr *types.NoSuchKey
+		if errors.As(err, &awsErr) {
 			return nil, errors.New("object not found")
 		}
 		return nil, fmt.Errorf("s3 binding error: error downloading S3 object: %w", err)
@@ -354,16 +386,16 @@ func (s *AWSS3) delete(ctx context.Context, req *bindings.InvokeRequest) (*bindi
 	if key == "" {
 		return nil, fmt.Errorf("s3 binding error: required metadata '%s' missing", metadataKey)
 	}
-	_, err := s.authProvider.S3().S3.DeleteObjectWithContext(
+	_, err := s.s3Client.DeleteObject(
 		ctx,
 		&s3.DeleteObjectInput{
 			Bucket: ptr.Of(s.metadata.Bucket),
 			Key:    ptr.Of(key),
 		},
 	)
 	if err != nil {
-		var awsErr awserr.Error
-		if errors.As(err, &awsErr) && awsErr.Code() == s3.ErrCodeNoSuchKey {
+		var awsErr *types.NoSuchKey
+		if errors.As(err, &awsErr) {
 			return nil, errors.New("object not found")
 		}
 		return nil, fmt.Errorf("s3 binding error: delete operation failed: %w", err)
@@ -383,9 +415,9 @@ func (s *AWSS3) list(ctx context.Context, req *bindings.InvokeRequest) (*binding
 	if payload.MaxResults < 1 {
 		payload.MaxResults = defaultMaxResults
 	}
-	result, err := s.authProvider.S3().S3.ListObjectsWithContext(ctx, &s3.ListObjectsInput{
+	result, err := s.s3Client.ListObjects(ctx, &s3.ListObjectsInput{
 		Bucket:    ptr.Of(s.metadata.Bucket),
-		MaxKeys:   ptr.Of(int64(payload.MaxResults)),
+		MaxKeys:   ptr.Of(payload.MaxResults),
 		Marker:    ptr.Of(payload.Marker),
 		Prefix:    ptr.Of(payload.Prefix),
 		Delimiter: ptr.Of(payload.Delimiter),

bindings/kafka/metadata.yaml

@@ -372,3 +372,10 @@ metadata:
       - "range"
       - "sticky"
       - "roundrobin"
+  - name: excludeHeaderMetaRegex
+    type: string
+    required: false
+    description: |
+      A regular expression to exclude keys from being converted to/from headers from/to metadata to avoid unwanted downstream side effects.
+    example: '"^rawPayload|valueSchemaType$"'
+    default: '""'

common/aws/auth/auth.go

@@ -0,0 +1,50 @@
+package auth
+
+import (
+	"context"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+
+	"github.com/dapr/kit/logger"
+)
+
+type ProviderType int
+
+const (
+	StaticProviderTypeStatic ProviderType = iota
+	StaticProviderTypeAssumeRole
+	X509ProviderType
+	ProviderTypeUnknown // Or default
+)
+
+type Options struct {
+	Logger     logger.Logger
+	Properties map[string]string
+
+	Region          string `json:"region" mapstructure:"region" mapstructurealiases:"awsRegion"`
+	AccessKey       string `json:"accessKey" mapstructure:"accessKey"`
+	SecretKey       string `json:"secretKey" mapstructure:"secretKey"`
+	SessionToken    string `json:"sessionToken" mapstructure:"sessionToken"`
+	AssumeRoleArn   string `json:"assumeRoleArn" mapstructure:"assumeRoleArn"`
+	TrustAnchorArn  string `json:"trustAnchorArn" mapstructure:"trustAnchorArn"`
+	TrustProfileArn string `json:"trustProfileArn" mapstructure:"trustProfileArn"`
+
+	Endpoint string `json:"endpoint" mapstructure:"endpoint"`
+}
+
+// CredentialProvider provides an interface for retrieving AWS credentials.
+type CredentialProvider interface {
+	Retrieve(ctx context.Context) (aws.Credentials, error)
+	Type() ProviderType
+}
+
+func NewCredentialProvider(ctx context.Context, opts Options, configOpts []func(*config.LoadOptions) error) (CredentialProvider,
+	error,
+) {
+	// TODO: Refactor this to search the opts structure for the right fields rather than the metadata map
+	if isX509Auth(opts.Properties) {
+		return newAuthX509(ctx, opts)
+	}
+	return newAuthStatic(ctx, opts, configOpts)
+}

common/aws/auth/auth_static.go

@@ -0,0 +1,76 @@
+package auth
+
+import (
+	"context"
+	"errors"
+
+	"github.com/aws/aws-sdk-go-v2/aws"
+	"github.com/aws/aws-sdk-go-v2/config"
+	"github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/credentials/stscreds"
+	"github.com/aws/aws-sdk-go-v2/service/sts"
+
+	"github.com/dapr/kit/logger"
+)
+
+type Static struct {
+	ProviderType  ProviderType
+	Logger        logger.Logger
+	AccessKey     string
+	SecretKey     string
+	SessionToken  string
+	Region        string
+	Endpoint      string
+	AssumeRoleArn string
+
+	CredentialProvider aws.CredentialsProvider
+}
+
+func (a *Static) Retrieve(ctx context.Context) (aws.Credentials, error) {
+	if a.CredentialProvider == nil {
+		return aws.Credentials{}, errors.New("credential provider is not set")
+	}
+	return a.CredentialProvider.Retrieve(ctx)
+}
+
+func (a *Static) Type() ProviderType {
+	return a.ProviderType
+}
+
+func newAuthStatic(ctx context.Context, opts Options, configOpts []func(*config.LoadOptions) error) (CredentialProvider, error) {
+	static := &Static{
+		Logger:        opts.Logger,
+		AccessKey:     opts.AccessKey,
+		SecretKey:     opts.SecretKey,
+		SessionToken:  opts.SessionToken,
+		Region:        opts.Region,
+		Endpoint:      opts.Endpoint,
+		AssumeRoleArn: opts.AssumeRoleArn,
+	}
+
+	switch {
+	case static.AccessKey != "" && static.SecretKey != "" && static.SessionToken != "":
+		static.ProviderType = StaticProviderTypeStatic
+		static.CredentialProvider = credentials.NewStaticCredentialsProvider(opts.AccessKey, opts.SecretKey,
+			opts.SessionToken)
+		static.Logger.Debug("using static credentials provider")
+
+	case static.AssumeRoleArn != "":
+		awsCfg, err := config.LoadDefaultConfig(ctx, configOpts...)
+		if err != nil {
+			return nil, err
+		}
+
+		stsSvc := sts.NewFromConfig(awsCfg)
+		stsProvider := stscreds.NewAssumeRoleProvider(stsSvc, static.AssumeRoleArn)
+		static.ProviderType = StaticProviderTypeAssumeRole
+		static.CredentialProvider = stsProvider
+		static.Logger.Debug("using AssumeRole credentials provider")
+
+	default:
+		static.Logger.Debug("using an undefined credentials provider, this may lead to unexpected behavior")
+		static.ProviderType = ProviderTypeUnknown
+	}
+
+	return static, nil
+}

common/aws/auth/auth_static_test.go

@@ -0,0 +1 @@
+package auth