feat(kafka): iam roles anywhere + assume role auth profiles (#3606)

Signed-off-by: Samantha Coyle <[email protected]>
Co-authored-by: Yaron Schneider <[email protected]>

Author: sicoyle

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

@@ -3,7 +3,7 @@ aws:
     description: |
       Authenticate using an Access Key ID and Secret Access Key included in the metadata
     metadata:
-      - name: awsRegion
+      - name: region
         type: string
         required: true
         description: |
@@ -25,8 +25,29 @@ aws:
         description: |
           AWS session token to use. A session token is only required if you are using
           temporary security credentials.
-        example: '"TOKEN"'
+  - title: "AWS: Assume IAM Role"
+    description: |
+      Assume a specific IAM role. Note: This is only supported for Kafka and PostgreSQL.
+    metadata:
+      - name: region
+        type: string
+        required: true
+        description: |
+          The AWS Region where the AWS resource is deployed to.
+        example: '"us-east-1"'
+      - name: assumeRoleArn
+        type: string
+        required: false
+        description: |
+          IAM role that has access to AWS resource.
+          This is another option to authenticate with MSK and RDS Aurora aside from the AWS Credentials.
+        example: '"arn:aws:iam::123456789:role/mskRole"'
+      - name: sessionName
         type: string
+        description: |
+          The session name for assuming a role.
+        example: '"MyAppSession"'
+        default: '"DaprDefaultSession"'
   - 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"

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

@@ -32,14 +32,46 @@ func ParseBuiltinAuthenticationProfile(bi BuiltinAuthenticationProfile, componen
 	for i, profile := range profiles {
 		res[i] = profile
 
-		res[i].Metadata = mergedMetadata(bi.Metadata, res[i].Metadata...)
+		// convert slice to a slice of pointers to update in place for required -> non-required fields
+		metadataPtr := make([]*Metadata, len(profile.Metadata))
+		for j := range profile.Metadata {
+			metadataPtr[j] = &profile.Metadata[j]
+		}
+
+		if componentTitle == "Apache Kafka" {
+			removeRequiredOnSomeAWSFields(&metadataPtr)
+		}
+
+		// convert back to value slices for merging
+		updatedMetadata := make([]Metadata, 0, len(metadataPtr))
+		for _, ptr := range metadataPtr {
+			if ptr != nil {
+				updatedMetadata = append(updatedMetadata, *ptr)
+			}
+		}
+
+		merged := mergedMetadata(bi.Metadata, updatedMetadata...)
+
+		// Note: We must apply the removal of deprecated fields after the merge!!
 
-		// If component is PostgreSQL, filter out duplicated aws profile fields
-		if strings.ToLower(componentTitle) == "postgresql" && bi.Name == "aws" {
-			res[i].Metadata = filterOutDuplicateFields(res[i].Metadata)
+		// 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" {
+			merged = removeSomeDeprecatedFieldsOnUnrelatedAuthProfiles(merged)
 		}
 
+		// Here, there are no metadata fields that need deprecating
+		if profile.Title == "AWS: Credentials from Environment Variables" && componentTitle == "Apache Kafka" {
+			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" {
+			merged = removeAllDeprecatedFieldsOnUnrelatedAuthProfiles(merged)
+		}
+
+		res[i].Metadata = merged
 	}
+
 	return res, nil
 }
 
@@ -54,26 +86,49 @@ func mergedMetadata(base []Metadata, add ...Metadata) []Metadata {
 	return res
 }
 
-// filterOutDuplicateFields removes specific duplicated fields from the metadata
-func filterOutDuplicateFields(metadata []Metadata) []Metadata {
-	duplicateFields := map[string]int{
-		"awsRegion": 0,
-		"accessKey": 0,
-		"secretKey": 0,
+// removeRequiredOnSomeAWSFields needs to be removed in Dapr 1.17 as duplicated AWS IAM fields get removed,
+// and we standardize on these fields.
+// Currently, there are: awsAccessKey, accessKey and awsSecretKey, secretKey, and awsRegion and region fields.
+// We normally have accessKey, secretKey, and region fields marked required as it is part of the builtin AWS auth profile fields.
+// However, as we rm the aws prefixed ones, we need to then mark the normally required ones as not required only for postgres and kafka.
+// This way we do not break existing users, and transition them to the standardized fields.
+func removeRequiredOnSomeAWSFields(metadata *[]*Metadata) {
+	if metadata == nil {
+		return
 	}
 
-	filteredMetadata := []Metadata{}
+	for _, field := range *metadata {
+		if field == nil {
+			continue
+		}
+
+		if field.Name == "accessKey" || field.Name == "secretKey" || field.Name == "region" {
+			field.Required = false
+		}
+	}
+}
 
+func removeAllDeprecatedFieldsOnUnrelatedAuthProfiles(metadata []Metadata) []Metadata {
+	filteredMetadata := []Metadata{}
 	for _, field := range metadata {
-		if _, exists := duplicateFields[field.Name]; !exists {
+		if strings.HasPrefix(field.Name, "aws") {
+			continue
+		} else {
 			filteredMetadata = append(filteredMetadata, field)
+		}
+	}
+
+	return filteredMetadata
+}
+
+func removeSomeDeprecatedFieldsOnUnrelatedAuthProfiles(metadata []Metadata) []Metadata {
+	filteredMetadata := []Metadata{}
+
+	for _, field := range metadata {
+		if field.Name == "awsAccessKey" || field.Name == "awsSecretKey" || field.Name == "awsSessionToken" {
+			continue
 		} else {
-			if field.Name == "awsRegion" && duplicateFields["awsRegion"] == 0 {
-				filteredMetadata = append(filteredMetadata, field)
-				duplicateFields["awsRegion"]++
-			} else if field.Name != "awsRegion" {
-				continue
-			}
+			filteredMetadata = append(filteredMetadata, field)
 		}
 	}
 

bindings/aws/sns/sns.go

@@ -43,6 +43,7 @@ type snsMetadata struct {
 	SessionToken string `json:"sessionToken" mapstructure:"sessionToken" mdignore:"true"`
 
 	TopicArn string `json:"topicArn"`
+	// TODO: in Dapr 1.17 rm the alias on region as we remove the aws prefix on these fields
 	Region   string `json:"region" mapstructure:"region" mapstructurealiases:"awsRegion" mdignore:"true"`
 	Endpoint string `json:"endpoint"`
 }

bindings/kafka/metadata.yaml

@@ -14,6 +14,75 @@ binding:
   operations:
     - name: create
       description: "Publish a new message in the topic."
+# This auth profile has duplicate fields intentionally as we maintain backwards compatibility,
+# but also move Kafka to utilize the noramlized AWS fields in the builtin auth profiles.
+# TODO: rm the duplicate aws prefixed fields in Dapr 1.17.
+builtinAuthenticationProfiles:
+  - name: "aws"
+    metadata:
+      - name: authType
+        type: string
+        required: true
+        description: |
+          Authentication type.
+          This must be set to "awsiam" for this authentication profile.
+        example: '"awsiam"'
+        allowedValues:
+          - "awsiam"
+      - 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 Relational Database Service is deployed to.
+        example: '"us-east-1"'
+      - name: awsAccessKey
+        type: string
+        required: false
+        description: |
+          This maintains backwards compatibility with existing fields. 
+          It will be deprecated as of Dapr 1.17. Use 'accessKey' instead.
+          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: false
+        sensitive: true
+        description: |
+          This maintains backwards compatibility with existing fields. 
+          It will be deprecated as of Dapr 1.17. Use 'secretKey' instead.
+          If both fields are set, then 'secretKey' value will be used.
+          The secret key associated with the access key.
+        example: '"wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"'
+      - name: awsSessionToken
+        type: string
+        sensitive: true
+        description: |
+          This maintains backwards compatibility with existing fields. 
+          It will be deprecated as of Dapr 1.17. Use 'sessionToken' instead.
+          If both fields are set, then 'sessionToken' value will be used.
+          AWS session token to use. A session token is only required if you are using temporary security credentials.
+        example: '"TOKEN"'
+      - name: awsIamRoleArn
+        type: string
+        required: false
+        description: |
+          This maintains backwards compatibility with existing fields. 
+          It will be deprecated as of Dapr 1.17. Use 'assumeRoleArn' instead.
+          If both fields are set, then 'assumeRoleArn' value will be used.
+          IAM role that has access to MSK. This is another option to authenticate with MSK aside from the AWS Credentials.
+        example: '"arn:aws:iam::123456789:role/mskRole"'
+      - name: awsStsSessionName
+        type: string
+        description: |
+          This maintains backwards compatibility with existing fields. 
+          It will be deprecated as of Dapr 1.17. Use 'sessionName' instead.
+          If both fields are set, then 'sessionName' value will be used.
+          Represents the session name for assuming a role.
+        example: '"MyAppSession"'
+        default: '"MSKSASLDefaultSession"'
 authenticationProfiles:
   - title: "OIDC Authentication"
     description: |
@@ -139,55 +208,6 @@ authenticationProfiles:
         example: '"none"'
         allowedValues:
           - "none"
-  - title: "AWS IAM"
-    description: "Authenticate using AWS IAM credentials or role for AWS MSK"
-    metadata:
-      - name: authType
-        type: string
-        required: true
-        description: |
-          Authentication type.
-          This must be set to "awsiam" for this authentication profile.
-        example: '"awsiam"'
-        allowedValues:
-          - "awsiam"
-      - 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"'
-      - name: awsIamRoleArn
-        type: string
-        required: true
-        description: |
-          IAM role that has access to MSK. This is another option to authenticate with MSK aside from the AWS Credentials.
-        example: '"arn:aws:iam::123456789:role/mskRole"'
-      - name: awsStsSessionName
-        type: string
-        description: |
-          Represents the session name for assuming a role.
-        example: '"MyAppSession"'
-        default: '"MSKSASLDefaultSession"'
 metadata:
   - name: topics
     type: string

common/authentication/aws/aws.go

@@ -44,6 +44,17 @@ type AWSIAM struct {
 	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 {
+	Region         string `json:"awsRegion" mapstructure:"awsRegion"`
+	AccessKey      string `json:"awsAccessKey" mapstructure:"awsAccessKey"`
+	SecretKey      string `json:"awsSecretKey" mapstructure:"awsSecretKey"`
+	SessionToken   string `json:"awsSessionToken" mapstructure:"awsSessionToken"`
+	IamRoleArn     string `json:"awsIamRoleArn" mapstructure:"awsIamRoleArn"`
+	StsSessionName string `json:"awsStsSessionName" mapstructure:"awsStsSessionName"`
+}
+
 type AWSIAMAuthOptions struct {
 	PoolConfig       *pgxpool.Config `json:"poolConfig" mapstructure:"poolConfig"`
 	ConnectionString string          `json:"connectionString" mapstructure:"connectionString"`
@@ -59,9 +70,13 @@ type Options 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"`
+	// TODO: in Dapr 1.17 rm the alias on regions as we rm the aws prefixed one.
+	// Docs have it just as region, but most metadata fields show the aws prefix...
+	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"`
 
 	Endpoint     string
 	SessionToken string
@@ -80,6 +95,7 @@ func GetConfig(opts Options) *aws.Config {
 	return cfg
 }
 
+//nolint:interfacebloat
 type Provider interface {
 	S3() *S3Clients
 	DynamoDB() *DynamoDBClients
@@ -91,6 +107,8 @@ type Provider interface {
 	Kinesis() *KinesisClients
 	Ses() *SesClients
 
+	Kafka(KafkaOptions) (*KafkaClients, error)
+
 	Close() error
 }
 
@@ -179,3 +197,14 @@ func (opts *Options) InitiateAWSIAMAuth() error {
 
 	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 {
+	for _, v := range values {
+		if v != "" {
+			return v
+		}
+	}
+	return ""
+}