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

Author: middt

.build-tools/pkg/metadataschema/builtin-authentication-profiles.go

@@ -38,7 +38,7 @@ func ParseBuiltinAuthenticationProfile(bi BuiltinAuthenticationProfile, componen
 			metadataPtr[j] = &profile.Metadata[j]
 		}
 
-		if componentTitle == "Apache Kafka" {
+		if componentTitle == "Apache Kafka" || strings.ToLower(componentTitle) == "postgresql" {
 			removeRequiredOnSomeAWSFields(&metadataPtr)
 		}
 
@@ -55,17 +55,17 @@ func ParseBuiltinAuthenticationProfile(bi BuiltinAuthenticationProfile, componen
 		// Note: We must apply the removal of deprecated fields after the merge!!
 
 		// Here, we remove some deprecated fields as we support the transition to a new auth profile
-		if profile.Title == "AWS: Assume specific IAM Role" && componentTitle == "Apache Kafka" {
+		if profile.Title == "AWS: Assume IAM Role" && componentTitle == "Apache Kafka" || profile.Title == "AWS: Assume IAM Role" && strings.ToLower(componentTitle) == "postgresql" {
 			merged = removeSomeDeprecatedFieldsOnUnrelatedAuthProfiles(merged)
 		}
 
 		// Here, there are no metadata fields that need deprecating
-		if profile.Title == "AWS: Credentials from Environment Variables" && componentTitle == "Apache Kafka" {
+		if profile.Title == "AWS: Credentials from Environment Variables" && componentTitle == "Apache Kafka" || profile.Title == "AWS: Credentials from Environment Variables" && strings.ToLower(componentTitle) == "postgresql" {
 			merged = removeAllDeprecatedFieldsOnUnrelatedAuthProfiles(merged)
 		}
 
 		// Here, this is a new auth profile, so rm all deprecating fields as unrelated.
-		if profile.Title == "AWS: IAM Roles Anywhere" && componentTitle == "Apache Kafka" {
+		if profile.Title == "AWS: IAM Roles Anywhere" && componentTitle == "Apache Kafka" || profile.Title == "AWS: IAM Roles Anywhere" && strings.ToLower(componentTitle) == "postgresql" {
 			merged = removeAllDeprecatedFieldsOnUnrelatedAuthProfiles(merged)
 		}
 
@@ -125,7 +125,7 @@ func removeSomeDeprecatedFieldsOnUnrelatedAuthProfiles(metadata []Metadata) []Me
 	filteredMetadata := []Metadata{}
 
 	for _, field := range metadata {
-		if field.Name == "awsAccessKey" || field.Name == "awsSecretKey" || field.Name == "awsSessionToken" {
+		if field.Name == "awsAccessKey" || field.Name == "awsSecretKey" || field.Name == "awsSessionToken" || field.Name == "awsRegion" {
 			continue
 		} else {
 			filteredMetadata = append(filteredMetadata, field)

bindings/kafka/metadata.yaml

@@ -35,7 +35,7 @@ builtinAuthenticationProfiles:
         description: |
           This maintains backwards compatibility with existing fields. 
           It will be deprecated as of Dapr 1.17. Use 'region' instead.
-          The AWS Region where the AWS Relational Database Service is deployed to.
+          The AWS Region where the AWS service is deployed to.
         example: '"us-east-1"'
       - name: awsAccessKey
         type: string
@@ -82,7 +82,7 @@ builtinAuthenticationProfiles:
           If both fields are set, then 'sessionName' value will be used.
           Represents the session name for assuming a role.
         example: '"MyAppSession"'
-        default: '"MSKSASLDefaultSession"'
+        default: '"DaprDefaultSession"'
 authenticationProfiles:
   - title: "OIDC Authentication"
     description: |

bindings/postgres/metadata.go

@@ -28,7 +28,7 @@ const (
 
 type psqlMetadata struct {
 	pgauth.PostgresAuthMetadata `mapstructure:",squash"`
-	aws.AWSIAM                  `mapstructure:",squash"`
+	aws.DeprecatedPostgresIAM   `mapstructure:",squash"`
 	Timeout                     time.Duration `mapstructure:"timeout" mapstructurealiases:"timeoutInSeconds"`
 }
 

bindings/postgres/metadata.yaml

@@ -56,25 +56,31 @@ builtinAuthenticationProfiles:
         example: |
           "host=mydb.postgres.database.aws.com user=myapplication port=5432 dbname=dapr_test sslmode=require"
         type: string
-      - name: awsRegion
-        type: string
-        required: true
-        description: |
-          The AWS Region where the AWS Relational Database Service is deployed to.
-        example: '"us-east-1"'
       - name: awsAccessKey
         type: string
-        required: true
+        required: false
         description: |
+          Deprecated as of Dapr 1.17. Use 'accessKey' instead if using AWS IAM.
+          If both fields are set, then 'accessKey' value will be used.
           AWS access key associated with an IAM account.
         example: '"AKIAIOSFODNN7EXAMPLE"'
       - name: awsSecretKey
         type: string
-        required: true
+        required: false
         sensitive: true
         description: |
+          Deprecated as of Dapr 1.17. Use 'secretKey' instead if using AWS IAM.
+          If both fields are set, then 'secretKey' value will be used.
           The secret key associated with the access key.
         example: '"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"'
+      - name: awsRegion
+        type: string
+        required: false
+        description: |
+          This maintains backwards compatibility with existing fields. 
+          It will be deprecated as of Dapr 1.17. Use 'region' instead.
+          The AWS Region where the AWS service is deployed to.
+        example: '"us-east-1"'
 authenticationProfiles:
   - title: "Connection string"
     description: "Authenticate using a Connection String"

bindings/postgres/postgres.go

@@ -26,6 +26,8 @@ import (
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/dapr/components-contrib/bindings"
+	awsAuth "github.com/dapr/components-contrib/common/authentication/aws"
+	pgauth "github.com/dapr/components-contrib/common/authentication/postgresql"
 	"github.com/dapr/components-contrib/metadata"
 	"github.com/dapr/kit/logger"
 )
@@ -45,6 +47,11 @@ type Postgres struct {
 	logger logger.Logger
 	db     *pgxpool.Pool
 	closed atomic.Bool
+
+	enableAzureAD bool
+	enableAWSIAM  bool
+
+	awsAuthProvider awsAuth.Provider
 }
 
 // NewPostgres returns a new PostgreSQL output binding.
@@ -59,18 +66,36 @@ func (p *Postgres) Init(ctx context.Context, meta bindings.Metadata) error {
 	if p.closed.Load() {
 		return errors.New("cannot initialize a previously-closed component")
 	}
-
+	opts := pgauth.InitWithMetadataOpts{
+		AzureADEnabled: p.enableAzureAD,
+		AWSIAMEnabled:  p.enableAWSIAM,
+	}
 	m := psqlMetadata{}
-	err := m.InitWithMetadata(meta.Properties)
-	if err != nil {
+	if err := m.InitWithMetadata(meta.Properties); err != nil {
 		return err
 	}
 
+	var err error
 	poolConfig, err := m.GetPgxPoolConfig()
 	if err != nil {
 		return err
 	}
 
+	if opts.AWSIAMEnabled && m.UseAWSIAM {
+		opts, validateErr := m.BuildAwsIamOptions(p.logger, meta.Properties)
+		if validateErr != nil {
+			return fmt.Errorf("failed to validate AWS IAM authentication fields: %w", validateErr)
+		}
+
+		var provider awsAuth.Provider
+		provider, err = awsAuth.NewProvider(ctx, *opts, awsAuth.GetConfig(*opts))
+		if err != nil {
+			return err
+		}
+		p.awsAuthProvider = provider
+		p.awsAuthProvider.UpdatePostgres(ctx, poolConfig)
+	}
+
 	// This context doesn't control the lifetime of the connection pool, and is
 	// only scoped to postgres creating resources at init.
 	connCtx, connCancel := context.WithTimeout(ctx, m.Timeout)
@@ -186,7 +211,11 @@ func (p *Postgres) Close() error {
 	}
 	p.db = nil
 
-	return nil
+	errs := make([]error, 1)
+	if p.awsAuthProvider != nil {
+		errs[0] = p.awsAuthProvider.Close()
+	}
+	return errors.Join(errs...)
 }
 
 func (p *Postgres) query(ctx context.Context, sql string, args ...any) (result []byte, err error) {

common/authentication/aws/aws.go

@@ -15,16 +15,8 @@ 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/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/pgxpool"
 
 	"github.com/dapr/kit/logger"
@@ -34,16 +26,6 @@ type EnvironmentSettings struct {
 	Metadata map[string]string
 }
 
-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 region in which PostgreSQL is deployed.
-	AWSRegion string `json:"awsRegion" mapstructure:"awsRegion"`
-}
-
 // TODO: Delete in Dapr 1.17 so we can move all IAM fields to use the defaults of:
 // accessKey and secretKey and region as noted in the docs, and Options struct above.
 type DeprecatedKafkaIAM struct {
@@ -55,14 +37,6 @@ type DeprecatedKafkaIAM struct {
 	StsSessionName string `json:"awsStsSessionName" mapstructure:"awsStsSessionName"`
 }
 
-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"`
-}
-
 type Options struct {
 	Logger     logger.Logger
 	Properties map[string]string
@@ -75,11 +49,20 @@ type Options struct {
 	Region        string `json:"region" mapstructure:"region" mapstructurealiases:"awsRegion"`
 	AccessKey     string `json:"accessKey" mapstructure:"accessKey"`
 	SecretKey     string `json:"secretKey" mapstructure:"secretKey"`
-	SessionName   string `mapstructure:"sessionName"`
-	AssumeRoleARN string `mapstructure:"assumeRoleArn"`
+	SessionName   string `json:"sessionName" mapstructure:"sessionName"`
+	AssumeRoleARN string `json:"assumeRoleArn" mapstructure:"assumeRoleArn"`
+	SessionToken  string `json:"sessionToken" mapstructure:"sessionToken"`
 
-	Endpoint     string
-	SessionToken string
+	Endpoint string
+}
+
+// TODO: Delete in Dapr 1.17 so we can move all IAM fields to use the defaults of:
+// accessKey and secretKey and region as noted in the docs, and Options struct above.
+type DeprecatedPostgresIAM struct {
+	// Access key to use for accessing PostgreSQL.
+	AccessKey string `json:"awsAccessKey" mapstructure:"awsAccessKey"`
+	// Secret key to use for accessing PostgreSQL.
+	SecretKey string `json:"awsSecretKey" mapstructure:"awsSecretKey"`
 }
 
 func GetConfig(opts Options) *aws.Config {
@@ -106,9 +89,14 @@ type Provider interface {
 	ParameterStore() *ParameterStoreClients
 	Kinesis() *KinesisClients
 	Ses() *SesClients
-
 	Kafka(KafkaOptions) (*KafkaClients, error)
 
+	// Postgres is an outlier to the others in the sense that we can update only it's config,
+	// as we use a max connection time of 8 minutes.
+	// This means that we can just update the config session credentials,
+	// and then in 8 minutes it will update to a new session automatically for us.
+	UpdatePostgres(context.Context, *pgxpool.Config)
+
 	Close() error
 }
 
@@ -128,69 +116,6 @@ func NewEnvironmentSettings(md map[string]string) (EnvironmentSettings, error) {
 	return es, nil
 }
 
-func (opts *Options) 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 "", errors.New("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 *Options) InitiateAWSIAMAuth() 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
-}
-
 // Coalesce is a helper function to return the first non-empty string from the inputs
 // This helps us to migrate away from the deprecated duplicate aws auth profile metadata fields in Dapr 1.17.
 func Coalesce(values ...string) string {