feat(AWS MSK): Add support for SASL/SCRAM authentication (#11060)

docs/providers/aws/events/msk.md

@@ -81,6 +81,21 @@ functions:
           enabled: false
 ```
 
+## Enabling authentication
+
+In order to authenticate to the `msk` you can set the `saslScram512`, which sets the authentication protocol.
+
+```yml
+functions:
+  compute:
+    handler: handler.compute
+    events:
+      - msk:
+          arn: arn:aws:kafka:region:XXXXXX:cluster/MyCluster/xxxx-xxxxx-xxxx
+          topic: mytopic
+          saslScram512: arn:aws:secretsmanager:region:XXXXXX:secret:AmazonMSK_xxxxxx
+```
+
 ## IAM Permissions
 
 The Serverless Framework will automatically configure the most minimal set of IAM permissions for you. However you can still add additional permissions if you need to. Read the official [AWS documentation](https://docs.aws.amazon.com/lambda/latest/dg/with-msk.html) for more information about IAM Permissions for MSK events.

docs/providers/aws/guide/serverless.yml.md

@@ -968,6 +968,8 @@ functions:
           startingPosition: LATEST
           # (default: true)
           enabled: false
+          # Optional, arn of the secret key for authenticating with the brokers in your MSK cluster.
+          saslScram512: arn:aws:secretsmanager:region:XXXXXX:secret:AmazonMSK_xxxxxx
 ```
 
 ### ActiveMQ

lib/plugins/aws/package/compile/events/msk/index.js

@@ -41,6 +41,7 @@ class AwsCompileMSKEvents {
         topic: {
           type: 'string',
         },
+        saslScram512: { $ref: '#/definitions/awsArnString' },
       },
       additionalProperties: false,
       required: ['arn', 'topic'],
@@ -79,6 +80,7 @@ class AwsCompileMSKEvents {
           const maximumBatchingWindow = event.msk.maximumBatchingWindow;
           const enabled = event.msk.enabled;
           const startingPosition = event.msk.startingPosition || 'TRIM_HORIZON';
+          const saslScram512 = event.msk.saslScram512;
 
           const mskClusterNameToken = getMskClusterNameToken(eventSourceArn);
           const mskEventLogicalId = this.provider.naming.getMSKEventLogicalId(
@@ -116,6 +118,16 @@ class AwsCompileMSKEvents {
             mskResource.Properties.Enabled = enabled;
           }
 
+          if (saslScram512 != null) {
+            const secureAccessConfigurations = [
+              {
+                Type: 'SASL_SCRAM_512_AUTH',
+                URI: saslScram512,
+              },
+            ];
+            mskResource.Properties.SourceAccessConfigurations = secureAccessConfigurations;
+          }
+
           mskStatement.Resource.push(eventSourceArn);
 
           cfTemplate.Resources[mskEventLogicalId] = mskResource;

test/unit/lib/plugins/aws/package/compile/events/msk/index.test.js

@@ -14,6 +14,14 @@ describe('AwsCompileMSKEvents', () => {
   const startingPosition = 'LATEST';
   const batchSize = 5000;
   const maximumBatchingWindow = 10;
+  const saslScram512 =
+    'arn:aws:secretsmanager:us-east-1:111111111111:secret:AmazonMSK_a1a1a1a1a1a1a1a1';
+  const sourceAccessConfigurations = [
+    {
+      Type: 'SASL_SCRAM_512_AUTH',
+      URI: saslScram512,
+    },
+  ];
 
   describe('when there are msk events defined', () => {
     let minimalEventSourceMappingResource;
@@ -46,6 +54,7 @@ describe('AwsCompileMSKEvents', () => {
                     maximumBatchingWindow,
                     enabled,
                     startingPosition,
+                    saslScram512,
                   },
                 },
               ],
@@ -108,6 +117,7 @@ describe('AwsCompileMSKEvents', () => {
         Enabled: enabled,
         EventSourceArn: arn,
         StartingPosition: startingPosition,
+        SourceAccessConfigurations: sourceAccessConfigurations,
         Topics: [topic],
         FunctionName: {
           'Fn::GetAtt': [naming.getLambdaLogicalId('other'), 'Arn'],