r/aws_msk_sasl_scram_secret: New Resource

aws/data_source_aws_msk_cluster.go

@@ -23,6 +23,10 @@ func dataSourceAwsMskCluster() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
+			"bootstrap_brokers_sasl_scram": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
 			"bootstrap_brokers_tls": {
 				Type:     schema.TypeString,
 				Computed: true,
@@ -100,6 +104,7 @@ func dataSourceAwsMskClusterRead(d *schema.ResourceData, meta interface{}) error
 
 	d.Set("arn", aws.StringValue(cluster.ClusterArn))
 	d.Set("bootstrap_brokers", aws.StringValue(bootstrapBrokersoOutput.BootstrapBrokerString))
+	d.Set("bootstrap_brokers_sasl_scram", aws.StringValue(bootstrapBrokersoOutput.BootstrapBrokerStringSaslScram))
 	d.Set("bootstrap_brokers_tls", aws.StringValue(bootstrapBrokersoOutput.BootstrapBrokerStringTls))
 	d.Set("cluster_name", aws.StringValue(cluster.ClusterName))
 	d.Set("kafka_version", aws.StringValue(cluster.CurrentBrokerSoftwareInfo.KafkaVersion))

aws/data_source_aws_msk_cluster_test.go

@@ -24,6 +24,7 @@ func TestAccAWSMskClusterDataSource_Name(t *testing.T) {
 				Check: resource.ComposeTestCheckFunc(
 					resource.TestCheckResourceAttrPair(resourceName, "arn", dataSourceName, "arn"),
 					resource.TestCheckResourceAttr(resourceName, "bootstrap_brokers", ""),
+					resource.TestCheckResourceAttrPair(resourceName, "bootstrap_brokers_sasl_scram", dataSourceName, "bootstrap_brokers_sasl_scram"),
 					resource.TestMatchResourceAttr(resourceName, "bootstrap_brokers_tls", regexp.MustCompile(`^(([-\w]+\.){1,}[\w]+:\d+,){2,}([-\w]+\.){1,}[\w]+:\d+$`)), // Hostname ordering not guaranteed between resource and data source reads
 					resource.TestCheckResourceAttrPair(resourceName, "cluster_name", dataSourceName, "cluster_name"),
 					resource.TestCheckResourceAttrPair(resourceName, "kafka_version", dataSourceName, "kafka_version"),

aws/internal/service/msk/finder/finder.go

@@ -0,0 +1,24 @@
+package finder
+
+import (
+	"github.com/aws/aws-sdk-go/aws"
+	"github.com/aws/aws-sdk-go/service/kafka"
+)
+
+// ScramSecrets returns the matching MSK Cluster's associated secrets
+func ScramSecrets(conn *kafka.Kafka, clusterArn string) ([]*string, error) {
+	input := &kafka.ListScramSecretsInput{
+		ClusterArn: aws.String(clusterArn),
+	}
+
+	var scramSecrets []*string
+	err := conn.ListScramSecretsPages(input, func(page *kafka.ListScramSecretsOutput, lastPage bool) bool {
+		if page == nil {
+			return !lastPage
+		}
+		scramSecrets = append(scramSecrets, page.SecretArnList...)
+		return !lastPage
+	})
+
+	return scramSecrets, err
+}

aws/provider.go

@@ -767,6 +767,7 @@ func Provider() *schema.Provider {
 			"aws_media_store_container_policy":                        resourceAwsMediaStoreContainerPolicy(),
 			"aws_msk_cluster":                                         resourceAwsMskCluster(),
 			"aws_msk_configuration":                                   resourceAwsMskConfiguration(),
+			"aws_msk_scram_secret_association":                        resourceAwsMskScramSecretAssociation(),
 			"aws_nat_gateway":                                         resourceAwsNatGateway(),
 			"aws_network_acl":                                         resourceAwsNetworkAcl(),
 			"aws_default_network_acl":                                 resourceAwsDefaultNetworkAcl(),

aws/resource_aws_msk_cluster.go

@@ -38,6 +38,10 @@ func resourceAwsMskCluster() *schema.Resource {
 				Type:     schema.TypeString,
 				Computed: true,
 			},
+			"bootstrap_brokers_sasl_scram": {
+				Type:     schema.TypeString,
+				Computed: true,
+			},
 			"bootstrap_brokers_tls": {
 				Type:     schema.TypeString,
 				Computed: true,
@@ -94,6 +98,22 @@ func resourceAwsMskCluster() *schema.Resource {
 				MaxItems: 1,
 				Elem: &schema.Resource{
 					Schema: map[string]*schema.Schema{
+						"sasl": {
+							Type:     schema.TypeList,
+							Optional: true,
+							ForceNew: true,
+							MaxItems: 1,
+							Elem: &schema.Resource{
+								Schema: map[string]*schema.Schema{
+									"scram": {
+										Type:     schema.TypeBool,
+										Optional: true,
+										ForceNew: true,
+									},
+								},
+							},
+							ConflictsWith: []string{"client_authentication.0.tls"},
+						},
 						"tls": {
 							Type:     schema.TypeList,
 							Optional: true,
@@ -444,6 +464,7 @@ func resourceAwsMskClusterRead(d *schema.ResourceData, meta interface{}) error {
 
 	d.Set("arn", aws.StringValue(cluster.ClusterArn))
 	d.Set("bootstrap_brokers", aws.StringValue(brokerOut.BootstrapBrokerString))
+	d.Set("bootstrap_brokers_sasl_scram", aws.StringValue(brokerOut.BootstrapBrokerStringSaslScram))
 	d.Set("bootstrap_brokers_tls", aws.StringValue(brokerOut.BootstrapBrokerStringTls))
 
 	if err := d.Set("broker_node_group_info", flattenMskBrokerNodeGroupInfo(cluster.BrokerNodeGroupInfo)); err != nil {
@@ -663,7 +684,8 @@ func expandMskClusterClientAuthentication(l []interface{}) *kafka.ClientAuthenti
 	m := l[0].(map[string]interface{})
 
 	ca := &kafka.ClientAuthentication{
-		Tls: expandMskClusterTls(m["tls"].([]interface{})),
+		Sasl: expandMskClusterScram(m["sasl"].([]interface{})),
+		Tls:  expandMskClusterTls(m["tls"].([]interface{})),
 	}
 
 	return ca
@@ -719,6 +741,25 @@ func expandMskClusterEncryptionInTransit(l []interface{}) *kafka.EncryptionInTra
 	return eit
 }
 
+func expandMskClusterScram(l []interface{}) *kafka.Sasl {
+	if len(l) == 0 || l[0] == nil {
+		return nil
+	}
+
+	tfMap, ok := l[0].(map[string]interface{})
+	if !ok {
+		return nil
+	}
+
+	sasl := &kafka.Sasl{
+		Scram: &kafka.Scram{
+			Enabled: aws.Bool(tfMap["scram"].(bool)),
+		},
+	}
+
+	return sasl
+}
+
 func expandMskClusterTls(l []interface{}) *kafka.Tls {
 	if len(l) == 0 || l[0] == nil {
 		return nil
@@ -892,7 +933,8 @@ func flattenMskClientAuthentication(ca *kafka.ClientAuthentication) []map[string
 	}
 
 	m := map[string]interface{}{
-		"tls": flattenMskTls(ca.Tls),
+		"sasl": flattenMskSasl(ca.Sasl),
+		"tls":  flattenMskTls(ca.Tls),
 	}
 
 	return []map[string]interface{}{m}
@@ -937,6 +979,26 @@ func flattenMskEncryptionInTransit(eit *kafka.EncryptionInTransit) []map[string]
 	return []map[string]interface{}{m}
 }
 
+func flattenMskSasl(sasl *kafka.Sasl) []map[string]interface{} {
+	if sasl == nil {
+		return []map[string]interface{}{}
+	}
+
+	m := map[string]interface{}{
+		"scram": flattenMskScram(sasl.Scram),
+	}
+
+	return []map[string]interface{}{m}
+}
+
+func flattenMskScram(scram *kafka.Scram) bool {
+	if scram == nil {
+		return false
+	}
+
+	return aws.BoolValue(scram.Enabled)
+}
+
 func flattenMskTls(tls *kafka.Tls) []map[string]interface{} {
 	if tls == nil {
 		return []map[string]interface{}{}

aws/resource_aws_msk_cluster_test.go

@@ -71,6 +71,7 @@ func TestAccAWSMskCluster_basic(t *testing.T) {
 					testAccCheckMskClusterExists(resourceName, &cluster),
 					testAccMatchResourceAttrRegionalARN(resourceName, "arn", "kafka", regexp.MustCompile(`cluster/.+`)),
 					resource.TestCheckResourceAttr(resourceName, "bootstrap_brokers", ""),
+					resource.TestCheckResourceAttr(resourceName, "bootstrap_brokers_sasl_scram", ""),
 					resource.TestMatchResourceAttr(resourceName, "bootstrap_brokers_tls", regexp.MustCompile(`^(([-\w]+\.){1,}[\w]+:\d+,){2,}([-\w]+\.){1,}[\w]+:\d+$`)),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.#", "1"),
 					resource.TestCheckResourceAttr(resourceName, "broker_node_group_info.0.az_distribution", kafka.BrokerAZDistributionDefault),
@@ -151,6 +152,59 @@ func TestAccAWSMskCluster_BrokerNodeGroupInfo_EbsVolumeSize(t *testing.T) {
 	})
 }
 
+func TestAccAWSMskCluster_ClientAuthentication_Sasl_Scram(t *testing.T) {
+	var cluster1, cluster2 kafka.ClusterInfo
+	rName := acctest.RandomWithPrefix("tf-acc-test")
+	resourceName := "aws_msk_cluster.test"
+
+	resource.ParallelTest(t, resource.TestCase{
+		PreCheck:     func() { testAccPreCheck(t); testAccPreCheckAWSMsk(t) },
+		Providers:    testAccProviders,
+		CheckDestroy: testAccCheckMskClusterDestroy,
+		Steps: []resource.TestStep{
+			{
+				Config: testAccMskClusterConfigClientAuthenticationSaslScram(rName, true),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckMskClusterExists(resourceName, &cluster1),
+					resource.TestMatchResourceAttr(resourceName, "bootstrap_brokers_sasl_scram", regexp.MustCompile(`^(([-\w]+\.){1,}[\w]+:\d+,){2,}([-\w]+\.){1,}[\w]+:\d+$`)),
+					resource.TestCheckResourceAttr(resourceName, "client_authentication.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "client_authentication.0.sasl.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "client_authentication.0.sasl.0.scram", "true"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"bootstrap_brokers",     // API may mutate ordering and selection of brokers to return
+					"bootstrap_brokers_tls", // API may mutate ordering and selection of brokers to return
+				},
+			},
+			{
+				Config: testAccMskClusterConfigClientAuthenticationSaslScram(rName, false),
+				Check: resource.ComposeTestCheckFunc(
+					testAccCheckMskClusterExists(resourceName, &cluster2),
+					testAccCheckMskClusterRecreated(&cluster1, &cluster2),
+					resource.TestCheckResourceAttr(resourceName, "bootstrap_brokers_sasl_scram", ""),
+					resource.TestCheckResourceAttr(resourceName, "client_authentication.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "client_authentication.0.sasl.#", "1"),
+					resource.TestCheckResourceAttr(resourceName, "client_authentication.0.sasl.0.scram", "false"),
+				),
+			},
+			{
+				ResourceName:      resourceName,
+				ImportState:       true,
+				ImportStateVerify: true,
+				ImportStateVerifyIgnore: []string{
+					"bootstrap_brokers",     // API may mutate ordering and selection of brokers to return
+					"bootstrap_brokers_tls", // API may mutate ordering and selection of brokers to return
+				},
+			},
+		},
+	})
+}
+
 func TestAccAWSMskCluster_ClientAuthentication_Tls_CertificateAuthorityArns(t *testing.T) {
 	TestAccSkip(t, "Requires the aws_acmpca_certificate_authority resource to support importing the root CA certificate")
 
@@ -924,6 +978,29 @@ resource "aws_msk_cluster" "test" {
 `, rName)
 }
 
+func testAccMskClusterConfigClientAuthenticationSaslScram(rName string, enabled bool) string {
+	return testAccMskClusterBaseConfig() + fmt.Sprintf(`
+resource "aws_msk_cluster" "test" {
+  cluster_name           = %[1]q
+  kafka_version          = "2.6.0"
+  number_of_broker_nodes = 3
+
+  broker_node_group_info {
+    client_subnets  = [aws_subnet.example_subnet_az1.id, aws_subnet.example_subnet_az2.id, aws_subnet.example_subnet_az3.id]
+    ebs_volume_size = 10
+    instance_type   = "kafka.m5.large"
+    security_groups = [aws_security_group.example_sg.id]
+  }
+
+  client_authentication {
+    sasl {
+      scram = %t
+    }
+  }
+}
+`, rName, enabled)
+}
+
 func testAccMskClusterConfigConfigurationInfoRevision1(rName string) string {
 	return testAccMskClusterBaseConfig() + fmt.Sprintf(`
 resource "aws_msk_configuration" "test" {