feat(msk): IAM access control for clusters

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

@@ -76,6 +76,10 @@ const cluster = msk.Cluster.fromClusterArn(this, 'Cluster', 'arn:aws:kafka:us-we
 
 ## Client Authentication
 
+[MSK supports](https://docs.aws.amazon.com/msk/latest/developerguide/kafka_apis_iam.html) the following authentication mechanisms.
+
+> Only one authentication method can be enabled.
+
 ### TLS
 
 To enable client authentication with TLS set the `certificateAuthorityArns` property to reference your ACM Private CA. [More info on Private CAs.](https://docs.aws.amazon.com/msk/latest/developerguide/msk-authentication.html)
@@ -104,7 +108,7 @@ const cluster = new msk.Cluster(this, 'Cluster', {
 
 ### SASL/SCRAM
 
-Enable client authentication with SASL/SCRAM:
+Enable client authentication with [SASL/SCRAM](https://docs.aws.amazon.com/msk/latest/developerguide/msk-password.html):
 
 ```typescript
 import * as msk from "@aws-cdk/aws-msk"
@@ -119,3 +123,21 @@ const cluster = new msk.cluster(this, "cluster", {
   }),
 })
 ```
+
+### SASL/IAM
+
+Enable client authentication with [IAM](https://docs.aws.amazon.com/msk/latest/developerguide/iam-access-control.html):
+
+```typescript
+import * as msk from "@aws-cdk/aws-msk"
+
+const cluster = new msk.cluster(this, "cluster", {
+  ...
+  encryptionInTransit: {
+    clientBroker: msk.ClientBrokerEncryption.TLS,
+  },
+  clientAuthentication: msk.ClientAuthentication.sasl({
+    iam: true,
+  }),
+})
+```

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

@@ -311,6 +311,12 @@ export interface SaslAuthProps {
    * @default false
    */
   readonly scram?: boolean;
+  /**
+   * Enable IAM access control.
+   *
+   * @default false
+   */
+  readonly iam?: boolean;
   /**
    * KMS Key to encrypt SASL/SCRAM secrets.
    *
@@ -424,6 +430,13 @@ 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 &&
@@ -435,10 +448,11 @@ export class Cluster extends ClusterBase {
     } else if (
       props.encryptionInTransit?.clientBroker ===
         ClientBrokerEncryption.TLS_PLAINTEXT &&
-      props.clientAuthentication?.saslProps?.scram
+      (props.clientAuthentication?.saslProps?.scram ||
+        props.clientAuthentication?.saslProps?.iam)
     ) {
       throw Error(
-        'To enable SASL/SCRAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.',
+        'To enable SASL/SCRAM or IAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.',
       );
     }
 
@@ -544,24 +558,31 @@ export class Cluster extends ClusterBase {
         }),
       );
     }
-    const clientAuthentication = props.clientAuthentication
-      ? {
-        sasl: props.clientAuthentication?.saslProps?.scram
-          ? {
-            scram: {
-              enabled: props.clientAuthentication?.saslProps.scram,
-            },
-          }
-          : undefined,
-        tls: props.clientAuthentication?.tlsProps?.certificateAuthorities
-          ? {
-            certificateAuthorityArnList: props.clientAuthentication?.tlsProps?.certificateAuthorities.map(
-              (ca) => ca.certificateAuthorityArn,
-            ),
-          }
-          : undefined,
-      }
-      : undefined;
+
+    let clientAuthentication;
+    if (props.clientAuthentication?.saslProps?.iam) {
+      clientAuthentication = {
+        sasl: { iam: { enabled: props.clientAuthentication.saslProps.iam } },
+      };
+    } 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,
+          ),
+        },
+      };
+    }
 
     const resource = new CfnCluster(this, 'Resource', {
       clusterName: props.clusterName,

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

@@ -109,18 +109,79 @@ describe('MSK Cluster', () => {
       });
 
       test('fails if tls encryption is set to tls and plaintext', () => {
-        expect(() => new msk.Cluster(stack, 'Cluster', {
+        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({
+                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', () => {
+      test('iam enabled is true', () => {
+        new msk.Cluster(stack, 'Cluster', {
           clusterName: 'cluster',
           kafkaVersion: msk.KafkaVersion.V2_6_1,
           vpc,
           encryptionInTransit: {
-            clientBroker: msk.ClientBrokerEncryption.TLS_PLAINTEXT,
+            clientBroker: msk.ClientBrokerEncryption.TLS,
           },
           clientAuthentication: msk.ClientAuthentication.sasl({
-            scram: true,
+            iam: true,
           }),
-        })).toThrow(
-          'To enable SASL/SCRAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.',
+        });
+        expect(stack).toHaveResourceLike('AWS::MSK::Cluster', {
+          ClientAuthentication: {
+            Sasl: { Iam: { 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,
+              }),
+            }),
+        ).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,
+              }),
+            }),
+        ).toThrow(
+          'To enable SASL/SCRAM or IAM authentication, you must only allow TLS-encrypted traffic between clients and brokers.',
         );
       });
     });
@@ -226,6 +287,24 @@ 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', () => {