fix(msk): allow both sasl/scram and iam auth (#31743)

Pointed out [here](#14647 (comment)) and verified in the Console, both `SASL/SCRAM` and `IAM` can be enabled together.

Closes #32779

It's a little confusing because [CloudFormation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-msk-cluster-sasl.html) groups `Iam` and `Scram` together under `Sasl`, 
but the Console separates the two and allows both at the same time. I'd like to refactor this further but 
this change unblocks the issue where `SASL/SCRAM` and `IAM` cannot be enabled together.

![image](https://github.com/user-attachments/assets/60dd662e-4db9-4bbb-bcc0-67bd059f6870)

Author: msambol

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

@@ -124,7 +124,7 @@ const cluster = new msk.Cluster(this, 'cluster', {
 });
 ```
 
-### SASL/IAM
+### IAM
 
 Enable client authentication with [IAM](https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html):
 

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

@@ -444,11 +444,7 @@ export class Cluster extends ClusterBase {
   /**
    * Reference an existing cluster, defined outside of the CDK code, by name.
    */
-  public static fromClusterArn(
-    scope: constructs.Construct,
-    id: string,
-    clusterArn: string,
-  ): ICluster {
+  public static fromClusterArn(scope: constructs.Construct, id: string, clusterArn: string): ICluster {
     class Import extends ClusterBase {
       public readonly clusterArn = clusterArn;
       public readonly clusterName = core.Fn.select(1, core.Fn.split('/', clusterArn)); // ['arn:partition:kafka:region:account-id', clusterName, clusterId]
@@ -465,9 +461,7 @@ export class Cluster extends ClusterBase {
   private _clusterBootstrapBrokers?: cr.AwsCustomResource;
 
   constructor(scope: constructs.Construct, id: string, props: ClusterProps) {
-    super(scope, id, {
-      physicalName: props.clusterName,
-    });
+    super(scope, id, { physicalName: props.clusterName });
     // Enhanced CDK Analytics Telemetry
     addConstructMetadata(this, props);
 
@@ -483,26 +477,11 @@ export class Cluster extends ClusterBase {
     });
 
     if (subnetSelection.subnets.length < 2) {
-      throw Error(
-        `Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length}`,
-      );
+      throw Error(`Cluster requires at least 2 subnets, got ${subnetSelection.subnets.length}`);
     }
 
-    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 &&
-      props.clientAuthentication
-    ) {
-      throw Error(
-        'To enable client authentication, you must enabled TLS-encrypted traffic between clients and brokers.',
-      );
+    if (props.encryptionInTransit?.clientBroker === ClientBrokerEncryption.PLAINTEXT && props.clientAuthentication) {
+      throw Error('To enable client authentication, you must enabled TLS-encrypted traffic between clients and brokers.');
     } else if (
       props.encryptionInTransit?.clientBroker ===
         ClientBrokerEncryption.TLS_PLAINTEXT &&
@@ -514,13 +493,10 @@ export class Cluster extends ClusterBase {
       );
     }
 
-    const volumeSize =
-      props.ebsStorageInfo?.volumeSize ?? 1000;
+    const volumeSize = props.ebsStorageInfo?.volumeSize ?? 1000;
     // Minimum: 1 GiB, maximum: 16384 GiB
     if (volumeSize < 1 || volumeSize > 16384) {
-      throw Error(
-        'EBS volume size should be in the range 1-16384',
-      );
+      throw Error('EBS volume size should be in the range 1-16384');
     }
 
     const instanceType = props.instanceType
@@ -642,13 +618,9 @@ export class Cluster extends ClusterBase {
       },
     };
 
-    if (
-      props.clientAuthentication?.saslProps?.scram &&
-      props.clientAuthentication?.saslProps?.key === undefined
-    ) {
+    if (props.clientAuthentication?.saslProps?.scram && props.clientAuthentication?.saslProps?.key === undefined) {
       this.saslScramAuthenticationKey = new kms.Key(this, 'SASLKey', {
-        description:
-          'Used for encrypting MSK secrets for SASL/SCRAM authentication.',
+        description: 'Used for encrypting MSK secrets for SASL/SCRAM authentication.',
         alias: `msk/${props.clusterName}/sasl/scram`,
       });
 
@@ -678,37 +650,16 @@ export class Cluster extends ClusterBase {
     }
 
     let clientAuthentication: CfnCluster.ClientAuthenticationProperty | undefined;
-    if (props.clientAuthentication?.saslProps?.iam) {
+    if (props.clientAuthentication) {
+      const { saslProps, tlsProps } = props.clientAuthentication;
       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,
-          },
-        },
-      };
-    } else if (
-      props.clientAuthentication?.tlsProps?.certificateAuthorities !== undefined
-    ) {
-      clientAuthentication = {
-        tls: {
-          certificateAuthorityArnList: props.clientAuthentication?.tlsProps?.certificateAuthorities.map(
-            (ca) => ca.certificateAuthorityArn,
-          ),
-        },
+        sasl: saslProps ? {
+          iam: saslProps.iam ? { enabled: true }: undefined,
+          scram: saslProps.scram ? { enabled: true }: undefined,
+        } : undefined,
+        tls: tlsProps?.certificateAuthorities ? {
+          certificateAuthorityArnList: tlsProps.certificateAuthorities?.map((ca) => ca.certificateAuthorityArn),
+        } : undefined,
       };
     }
 

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 combinations of sasl/scram, iam, and tls Snapshot test with all values set (iam/scram/tls) 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",