fix(msk-alpha): support any combination of client auth mechanisms

packages/@aws-cdk/aws-msk-alpha/README.md

@@ -143,6 +143,25 @@ const cluster = new msk.Cluster(this, 'cluster', {
 });
 ```
 
+### SASL/SCRAM + SASL/IAM
+
+Enable client authentication with [IAM](https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html) and [SASL/SCRAM](https://docs.aws.amazon.com/msk/latest/developerguide/msk-password.html):
+
+```ts
+declare const vpc: ec2.Vpc;
+const cluster = new msk.Cluster(this, 'cluster', {
+  clusterName: 'myCluster',
+  kafkaVersion: msk.KafkaVersion.V2_8_1,
+  vpc,
+  encryptionInTransit: {
+    clientBroker: msk.ClientBrokerEncryption.TLS,
+  },
+  clientAuthentication: msk.ClientAuthentication.sasl({
+    iam: true,
+    scram: true,
+  }),
+});
+```
 
 ### SASL/IAM + TLS
 

packages/@aws-cdk/aws-msk-alpha/lib/cluster.ts

@@ -496,13 +496,6 @@ export class Cluster extends ClusterBase {
       );
     }
 
-    if (
-      props.clientAuthentication?.saslProps?.iam &&
-      props.clientAuthentication?.saslProps?.scram
-    ) {
-      throw Error('Only one client authentication method can be enabled.');
-    }
-
     if (
       props.encryptionInTransit?.clientBroker ===
         ClientBrokerEncryption.PLAINTEXT &&
@@ -685,38 +678,18 @@ export class Cluster extends ClusterBase {
       );
     }
 
-    let clientAuthentication;
-    if (props.clientAuthentication?.saslProps?.iam) {
-      clientAuthentication = {
-        sasl: { iam: { enabled: props.clientAuthentication.saslProps.iam } },
-      };
-      if (props.clientAuthentication?.tlsProps) {
-        clientAuthentication = {
-          sasl: { iam: { enabled: props.clientAuthentication.saslProps.iam } },
-          tls: {
-            certificateAuthorityArnList: props.clientAuthentication?.tlsProps?.certificateAuthorities?.map(
-              (ca) => ca.certificateAuthorityArn,
-            ),
-          },
-        };
-      }
-    } else if (props.clientAuthentication?.saslProps?.scram) {
-      clientAuthentication = {
-        sasl: {
-          scram: {
-            enabled: props.clientAuthentication.saslProps.scram,
-          },
-        },
+    let clientAuthentication: any = {};
+
+    if (props.clientAuthentication?.saslProps) {
+      clientAuthentication.sasl = {
+        ...(props.clientAuthentication?.saslProps?.iam ? { iam: { enabled: props.clientAuthentication.saslProps.iam } } : {}),
+        ...(props.clientAuthentication?.saslProps?.scram ? { scram: { enabled: props.clientAuthentication.saslProps.scram } } : {}),
       };
-    } else if (
-      props.clientAuthentication?.tlsProps?.certificateAuthorities !== undefined
-    ) {
-      clientAuthentication = {
-        tls: {
-          certificateAuthorityArnList: props.clientAuthentication?.tlsProps?.certificateAuthorities.map(
-            (ca) => ca.certificateAuthorityArn,
-          ),
-        },
+    }
+
+    if (props.clientAuthentication?.tlsProps) {
+      clientAuthentication.tls = {
+        certificateAuthorityArnList: props.clientAuthentication.tlsProps?.certificateAuthorities?.map((ca) => ca.certificateAuthorityArn),
       };
     }
 
@@ -831,7 +804,7 @@ export class Cluster extends ClusterBase {
    */
   private _bootstrapBrokers(responseField: string): string {
     if (!this._clusterBootstrapBrokers) {
-      this._clusterBootstrapBrokers = new cr.AwsCustomResource(this, `BootstrapBrokers${responseField}`, {
+      this._clusterBootstrapBrokers = new cr.AwsCustomResource(this, 'BootstrapBrokers', {
         onUpdate: {
           service: 'Kafka',
           action: 'getBootstrapBrokers',

packages/@aws-cdk/aws-msk-alpha/test/__snapshots__/cluster.test.ts.snap

@@ -493,7 +493,7 @@ exports[`MSK Cluster Snapshot test with all values set 1`] = `
 }
 `;
 
-exports[`MSK Cluster created with authentication enabled with sasl/iam auth and tls Snapshot test with all values set (iam/sasl) 1`] = `
+exports[`MSK Cluster created with authentication enabled with sasl/iam auth, sasl/scram auth and tls auth Snapshot test with all values set (iam/sasl) 1`] = `
 {
   "Resources": {
     "Vpc8378EB38": {
@@ -915,6 +915,9 @@ exports[`MSK Cluster created with authentication enabled with sasl/iam auth and
             "Iam": {
               "Enabled": true,
             },
+            "Scram": {
+              "Enabled": true,
+            },
           },
           "Tls": {
             "CertificateAuthorityArnList": [
@@ -965,6 +968,89 @@ exports[`MSK Cluster created with authentication enabled with sasl/iam auth and
       "Type": "AWS::MSK::Cluster",
       "UpdateReplacePolicy": "Retain",
     },
+    "kafkaSASLKey69FC3AFA": {
+      "DeletionPolicy": "Retain",
+      "Properties": {
+        "Description": "Used for encrypting MSK secrets for SASL/SCRAM authentication.",
+        "KeyPolicy": {
+          "Statement": [
+            {
+              "Action": "kms:*",
+              "Effect": "Allow",
+              "Principal": {
+                "AWS": {
+                  "Fn::Join": [
+                    "",
+                    [
+                      "arn:",
+                      {
+                        "Ref": "AWS::Partition",
+                      },
+                      ":iam::",
+                      {
+                        "Ref": "AWS::AccountId",
+                      },
+                      ":root",
+                    ],
+                  ],
+                },
+              },
+              "Resource": "*",
+            },
+            {
+              "Action": [
+                "kms:Encrypt",
+                "kms:Decrypt",
+                "kms:ReEncrypt*",
+                "kms:GenerateDataKey*",
+                "kms:CreateGrant",
+                "kms:DescribeKey",
+              ],
+              "Condition": {
+                "StringEquals": {
+                  "kms:CallerAccount": {
+                    "Ref": "AWS::AccountId",
+                  },
+                  "kms:ViaService": {
+                    "Fn::Join": [
+                      "",
+                      [
+                        "secretsmanager.",
+                        {
+                          "Ref": "AWS::Region",
+                        },
+                        ".amazonaws.com",
+                      ],
+                    ],
+                  },
+                },
+              },
+              "Effect": "Allow",
+              "Principal": {
+                "AWS": "*",
+              },
+              "Resource": "*",
+              "Sid": "Allow access through AWS Secrets Manager for all principals in the account that are authorized to use AWS Secrets Manager",
+            },
+          ],
+          "Version": "2012-10-17",
+        },
+      },
+      "Type": "AWS::KMS::Key",
+      "UpdateReplacePolicy": "Retain",
+    },
+    "kafkaSASLKeyAlias7A73E101": {
+      "Properties": {
+        "AliasName": "alias/msk/test-cluster/sasl/scram",
+        "TargetKeyId": {
+          "Fn::GetAtt": [
+            "kafkaSASLKey69FC3AFA",
+            "Arn",
+          ],
+        },
+      },
+      "Type": "AWS::KMS::Alias",
+    },
     "sg1fromsg32181E6F4C07E": {
       "Properties": {
         "Description": "from sg3:2181",

packages/@aws-cdk/aws-msk-alpha/test/cluster.test.ts

@@ -226,7 +226,71 @@ describe('MSK Cluster', () => {
       });
     });
 
-    describe('with sasl/iam auth and tls', () => {
+    describe('with sasl/iam and sasl/scram', () => {
+      test('iam and scram enabled is true', () => {
+        new msk.Cluster(stack, 'Cluster', {
+          clusterName: 'cluster',
+          kafkaVersion: msk.KafkaVersion.V3_6_0,
+          vpc,
+          encryptionInTransit: {
+            clientBroker: msk.ClientBrokerEncryption.TLS,
+          },
+          clientAuthentication: msk.ClientAuthentication.sasl({
+            iam: true,
+            scram: true,
+          }),
+        });
+
+        Template.fromStack(stack).hasResourceProperties('AWS::MSK::Cluster', {
+          ClientAuthentication: {
+            Sasl: {
+              Iam: { Enabled: true },
+              Scram: { Enabled: true },
+            },
+          },
+        });
+      });
+
+      test('fails if tls encryption is set to plaintext', () => {
+        expect(() => new msk.Cluster(stack, 'Cluster', {
+          clusterName: 'cluster',
+          kafkaVersion: msk.KafkaVersion.V2_6_1,
+          vpc,
+          encryptionInTransit: {
+            clientBroker: msk.ClientBrokerEncryption.PLAINTEXT,
+          },
+          clientAuthentication: msk.ClientAuthentication.sasl({
+            iam: true,
+            scram: true,
+          }),
+        }),
+        ).toThrow(
+          'To enable client authentication, you must enabled TLS-encrypted traffic between clients and brokers.',
+        );
+      });
+
+      test('fails if tls encryption is set to tls and plaintext', () => {
+        expect(
+          () =>
+            new msk.Cluster(stack, 'Cluster', {
+              clusterName: 'cluster',
+              kafkaVersion: msk.KafkaVersion.V2_6_1,
+              vpc,
+              encryptionInTransit: {
+                clientBroker: msk.ClientBrokerEncryption.TLS_PLAINTEXT,
+              },
+              clientAuthentication: msk.ClientAuthentication.sasl({
+                iam: true,
+                scram: true,
+              }),
+            }),
+        ).toThrow(
+          'To enable SASL/SCRAM or IAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.',
+        );
+      });
+    });
+
+    describe('with sasl/iam auth, sasl/scram auth and tls auth', () => {
       test('Snapshot test with all values set (iam/sasl)', () => {
         const cluster = new msk.Cluster(stack, 'kafka', {
           clusterName: 'test-cluster',
@@ -249,6 +313,7 @@ describe('MSK Cluster', () => {
           },
           clientAuthentication: msk.ClientAuthentication.saslTls({
             iam: true,
+            scram: true,
             certificateAuthorities: [
               acmpca.CertificateAuthority.fromCertificateAuthorityArn(
                 stack,
@@ -373,23 +438,6 @@ describe('MSK Cluster', () => {
       });
     });
 
-    test('fails if more than one authentication method is enabled', () => {
-      expect(
-        () =>
-          new msk.Cluster(stack, 'Cluster', {
-            clusterName: 'cluster',
-            kafkaVersion: msk.KafkaVersion.V2_6_1,
-            vpc,
-            encryptionInTransit: {
-              clientBroker: msk.ClientBrokerEncryption.TLS,
-            },
-            clientAuthentication: msk.ClientAuthentication.sasl({
-              iam: true,
-              scram: true,
-            }),
-          }),
-      ).toThrow('Only one client authentication method can be enabled.');
-    });
   });
 
   describe('created with an instance type set', () => {