Merge remote-tracking branch 'upstream/main' into fix-cert-test-deps

Signed-off-by: Bernd Verst <github@bernd.dev>

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

@@ -3,6 +3,12 @@ aws:
     description: |
       Authenticate using an Access Key ID and Secret Access Key included in the metadata
     metadata:
+      - name: awsRegion
+        type: string
+        required: true
+        description: |
+          The AWS Region where the AWS Relational Database Service is deployed to.
+        example: '"us-east-1"'
       - name: accessKey
         description: AWS access key associated with an IAM account
         required: true

bindings/azure/eventhubs/metadata.yaml

@@ -90,8 +90,8 @@ builtinAuthenticationProfiles:
           entity management is enabled.
 metadata:
   # Input-only metadata
-  # consumerGroup is an alias for consumerId, if both are defined consumerId takes precedence.
-  - name: consumerId
+  # consumerGroup is an alias for consumerID, if both are defined consumerID takes precedence.
+  - name: consumerID
     type: string
     required: true # consumerGroup is an alias for this field, let's promote this to default
     binding:
@@ -108,7 +108,7 @@ metadata:
       output: false
     description: |
       The name of the Event Hubs Consumer Group to listen on.
-      Alias to consumerId.
+      Alias to consumerID.
     example: '"group1"'
     deprecated: true
   - name: storageAccountKey

bindings/kafka/metadata.yaml

@@ -174,7 +174,7 @@ authenticationProfiles:
         type: string
         sensitive: true
         description: |
-          AWS session token to use. A session token is only required if you are using\ntemporary security credentials.
+          AWS session token to use. A session token is only required if you are using temporary security credentials.
         example: '"TOKEN"'
       - name: awsIamRoleArn
         type: string

bindings/postgres/metadata.go

@@ -14,26 +14,41 @@ limitations under the License.
 package postgres
 
 import (
+	"time"
+
+	"github.com/dapr/components-contrib/common/authentication/aws"
 	pgauth "github.com/dapr/components-contrib/common/authentication/postgresql"
 	kitmd "github.com/dapr/kit/metadata"
 )
 
+const (
+	defaultTimeout = 20 * time.Second // Default timeout for network requests
+)
+
 type psqlMetadata struct {
 	pgauth.PostgresAuthMetadata `mapstructure:",squash"`
+	aws.AWSIAM                  `mapstructure:",squash"`
+	Timeout                     time.Duration `mapstructure:"timeout" mapstructurealiases:"timeoutInSeconds"`
 }
 
 func (m *psqlMetadata) InitWithMetadata(meta map[string]string) error {
 	// Reset the object
 	m.PostgresAuthMetadata.Reset()
+	m.Timeout = defaultTimeout
 
 	err := kitmd.DecodeMetadata(meta, &m)
 	if err != nil {
 		return err
 	}
 
+	opts := pgauth.InitWithMetadataOpts{
+		AzureADEnabled: true,
+		AWSIAMEnabled:  true,
+	}
+
 	// Validate and sanitize input
-	// Azure AD auth is supported for this component
-	err = m.PostgresAuthMetadata.InitWithMetadata(meta, true)
+	// Azure AD & AWS IAM auth is supported for this component
+	err = m.PostgresAuthMetadata.InitWithMetadata(meta, opts)
 	if err != nil {
 		return err
 	}

bindings/postgres/metadata.yaml

@@ -38,6 +38,40 @@ builtinAuthenticationProfiles:
         example: |
           "host=mydb.postgres.database.azure.com user=myapplication port=5432 database=dapr_test sslmode=require"
         type: string
+  - name: "aws"
+    metadata:
+      - name: useAWSIAM
+        required: true
+        type: bool
+        example: '"true"'
+        description: |
+          Must be set to `true` to enable the component to retrieve access tokens from AWS IAM.
+          This authentication method only works with AWS Relational Database Service for PostgreSQL databases.
+      - name: awsRegion
+        type: string
+        required: true
+        description: |
+          The AWS Region where the MSK Kafka broker is deployed to.
+        example: '"us-east-1"'
+      - name: awsAccessKey
+        type: string
+        required: true
+        description: |
+          AWS access key associated with an IAM account.
+        example: '"AKIAIOSFODNN7EXAMPLE"'
+      - name: awsSecretKey
+        type: string
+        required: true
+        sensitive: true
+        description: |
+          The secret key associated with the access key.
+        example: '"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"'
+      - name: awsSessionToken
+        type: string
+        sensitive: true
+        description: |
+          AWS session token to use. A session token is only required if you are using temporary security credentials.
+        example: '"TOKEN"'
 authenticationProfiles:
   - title: "Connection string"
     description: "Authenticate using a Connection String"
@@ -54,6 +88,12 @@ authenticationProfiles:
           or "postgres://dapr:secret@dapr.example.com:5432/dapr?sslmode=verify-ca"
         type: string
 metadata:
+  - name: timeout
+    required: false
+    description: Timeout for all database operations. 
+    example:  "30s"
+    default: "20s"
+    type: duration
   - name: maxConns
     required: false
     description: |

bindings/postgres/postgres.go

@@ -66,9 +66,9 @@ func (p *Postgres) Init(ctx context.Context, meta bindings.Metadata) error {
 		return err
 	}
 
-	poolConfig, err := m.GetPgxPoolConfig()
+	poolConfig, err := m.GetPgxPoolConfig(ctx)
 	if err != nil {
-		return fmt.Errorf("error opening DB connection: %w", err)
+		return err
 	}
 
 	// This context doesn't control the lifetime of the connection pool, and is

common/authentication/aws/aws.go

@@ -14,14 +14,29 @@ limitations under the License.
 package aws
 
 import (
+	"context"
+	"errors"
+	"fmt"
+	"strconv"
+	"time"
+
+	"github.com/aws/aws-sdk-go-v2/config"
+	v2creds "github.com/aws/aws-sdk-go-v2/credentials"
+	"github.com/aws/aws-sdk-go-v2/feature/rds/auth"
 	"github.com/aws/aws-sdk-go/aws"
 	"github.com/aws/aws-sdk-go/aws/credentials"
 	"github.com/aws/aws-sdk-go/aws/request"
 	"github.com/aws/aws-sdk-go/aws/session"
+	"github.com/jackc/pgx/v5"
+	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/dapr/kit/logger"
 )
 
+type EnvironmentSettings struct {
+	Metadata map[string]string
+}
+
 func GetClient(accessKey string, secretKey string, sessionToken string, region string, endpoint string) (*session.Session, error) {
 	awsConfig := aws.NewConfig()
 
@@ -53,3 +68,95 @@ func GetClient(accessKey string, secretKey string, sessionToken string, region s
 
 	return awsSession, nil
 }
+
+// NewEnvironmentSettings returns a new EnvironmentSettings configured for a given AWS resource.
+func NewEnvironmentSettings(md map[string]string) (EnvironmentSettings, error) {
+	es := EnvironmentSettings{
+		Metadata: md,
+	}
+
+	return es, nil
+}
+
+type AWSIAM struct {
+	// Ignored by metadata parser because included in built-in authentication profile
+	// access key to use for accessing postgresql.
+	AWSAccessKey string `json:"awsAccessKey" mapstructure:"awsAccessKey"`
+	// secret key to use for accessing postgresql.
+	AWSSecretKey string `json:"awsSecretKey" mapstructure:"awsSecretKey"`
+	// aws session token to use.
+	AWSSessionToken string `mapstructure:"awsSessionToken"`
+	// aws region in which postgresql should create resources.
+	AWSRegion string `mapstructure:"awsRegion"`
+}
+
+type AWSIAMAuthOptions struct {
+	PoolConfig       *pgxpool.Config `json:"poolConfig" mapstructure:"poolConfig"`
+	ConnectionString string          `json:"connectionString" mapstructure:"connectionString"`
+	Region           string          `json:"region" mapstructure:"region"`
+	AccessKey        string          `json:"accessKey" mapstructure:"accessKey"`
+	SecretKey        string          `json:"secretKey" mapstructure:"secretKey"`
+}
+
+func (opts *AWSIAMAuthOptions) GetAccessToken(ctx context.Context) (string, error) {
+	dbEndpoint := opts.PoolConfig.ConnConfig.Host + ":" + strconv.Itoa(int(opts.PoolConfig.ConnConfig.Port))
+	var authenticationToken string
+
+	// https://docs.aws.amazon.com/AmazonRDS/latest/AuroraUserGuide/UsingWithRDS.IAMDBAuth.Connecting.Go.html
+	// Default to load default config through aws credentials file (~/.aws/credentials)
+	awsCfg, err := config.LoadDefaultConfig(ctx)
+	// Note: in the event of an error with invalid config or failed to load config,
+	// then we fall back to using the access key and secret key.
+	switch {
+	case errors.Is(err, config.SharedConfigAssumeRoleError{}.Err),
+		errors.Is(err, config.SharedConfigLoadError{}.Err),
+		errors.Is(err, config.SharedConfigProfileNotExistError{}.Err):
+		// Validate if access key and secret access key are provided
+		if opts.AccessKey == "" || opts.SecretKey == "" {
+			return "", fmt.Errorf("failed to load default configuration for AWS using accessKey and secretKey: %w", err)
+		}
+
+		// Set credentials explicitly
+		awsCfg := v2creds.NewStaticCredentialsProvider(opts.AccessKey, opts.SecretKey, "")
+		authenticationToken, err = auth.BuildAuthToken(
+			ctx, dbEndpoint, opts.Region, opts.PoolConfig.ConnConfig.User, awsCfg)
+		if err != nil {
+			return "", fmt.Errorf("failed to create AWS authentication token: %w", err)
+		}
+
+		return authenticationToken, nil
+	case err != nil:
+		return "", fmt.Errorf("failed to load default AWS authentication configuration")
+	}
+
+	authenticationToken, err = auth.BuildAuthToken(
+		ctx, dbEndpoint, opts.Region, opts.PoolConfig.ConnConfig.User, awsCfg.Credentials)
+	if err != nil {
+		return "", fmt.Errorf("failed to create AWS authentication token: %w", err)
+	}
+
+	return authenticationToken, nil
+}
+
+func (opts *AWSIAMAuthOptions) InitiateAWSIAMAuth(ctx context.Context) error {
+	// Set max connection lifetime to 8 minutes in postgres connection pool configuration.
+	// Note: this will refresh connections before the 15 min expiration on the IAM AWS auth token,
+	// while leveraging the BeforeConnect hook to recreate the token in time dynamically.
+	opts.PoolConfig.MaxConnLifetime = time.Minute * 8
+
+	// Setup connection pool config needed for AWS IAM authentication
+	opts.PoolConfig.BeforeConnect = func(ctx context.Context, pgConfig *pgx.ConnConfig) error {
+		// Manually reset auth token with aws and reset the config password using the new iam token
+		pwd, errGetAccessToken := opts.GetAccessToken(ctx)
+		if errGetAccessToken != nil {
+			return fmt.Errorf("failed to refresh access token for iam authentication with PostgreSQL: %w", errGetAccessToken)
+		}
+
+		pgConfig.Password = pwd
+		opts.PoolConfig.ConnConfig.Password = pwd
+
+		return nil
+	}
+
+	return nil
+}