Add validation for SecretsManagerKmsKeyId

samtranslator/model/eventsources/pull.py

@@ -188,6 +188,13 @@ def _validate_filter_criteria(self):
         if list(self.FilterCriteria.keys()) not in [[], ["Filters"]]:
             raise InvalidEventException(self.relative_id, "FilterCriteria field has a wrong format")
 
+    def validate_secrets_manager_kms_key_id(self):
+        if self.SecretsManagerKmsKeyId and not isinstance(self.SecretsManagerKmsKeyId, str):
+            raise InvalidEventException(
+                self.relative_id,
+                "Provided SecretsManagerKmsKeyId should be of type str.",
+            )
+
 
 class Kinesis(PullEventSource):
     """Kinesis event source."""
@@ -304,6 +311,7 @@ def get_policy_statements(self):
             },
         }
         if self.SecretsManagerKmsKeyId:
+            self.validate_secrets_manager_kms_key_id()
             kms_policy = {
                 "Action": "kms:Decrypt",
                 "Effect": "Allow",
@@ -367,6 +375,7 @@ def generate_policy_document(self):
             statements.append(vpc_permissions)
 
         if self.SecretsManagerKmsKeyId:
+            self.validate_secrets_manager_kms_key_id()
             kms_policy = self.get_kms_policy()
             statements.append(kms_policy)
 

tests/model/eventsources/test_mq_event_source.py

@@ -1,5 +1,7 @@
 from unittest import TestCase
 from samtranslator.model.eventsources.pull import MQ
+from samtranslator.model.exceptions import InvalidEventException
+from parameterized import parameterized
 
 
 class MQEventSource(TestCase):
@@ -40,3 +42,57 @@ def test_get_policy_statements(self):
             }
         ]
         self.assertEqual(policy_statements, expected_policy_document)
+
+    @parameterized.expand(
+        [
+            (1,),
+            (True,),
+            (["1abc23d4-567f-8ab9-cde0-1fab234c5d67"],),
+            ({"KmsKeyId": "1abc23d4-567f-8ab9-cde0-1fab234c5d67"},),
+        ]
+    )
+    def test_must_validate_secrets_manager_kms_key_id(self, kms_key_id_value):
+        self.mq_event_source.SourceAccessConfigurations = [{"Type": "BASIC_AUTH", "URI": "SECRET_URI"}]
+        self.mq_event_source.Broker = "BROKER_ARN"
+        self.mq_event_source.SecretsManagerKmsKeyId = kms_key_id_value
+        error_message = "(None, 'Provided SecretsManagerKmsKeyId should be of type str.')"
+        with self.assertRaises(InvalidEventException) as error:
+            self.mq_event_source.get_policy_statements()
+        self.assertEqual(error_message, str(error.exception))
+
+    def test_get_policy_statements_with_secrets_manager_kms_key_id(self):
+        self.mq_event_source.SourceAccessConfigurations = [{"Type": "BASIC_AUTH", "URI": "SECRET_URI"}]
+        self.mq_event_source.Broker = "BROKER_ARN"
+        self.mq_event_source.SecretsManagerKmsKeyId = "1abc23d4-567f-8ab9-cde0-1fab234c5d67"
+        policy_statements = self.mq_event_source.get_policy_statements()
+        expected_policy_document = [
+            {
+                "PolicyName": "SamAutoGeneratedAMQPolicy",
+                "PolicyDocument": {
+                    "Statement": [
+                        {
+                            "Action": [
+                                "secretsmanager:GetSecretValue",
+                            ],
+                            "Effect": "Allow",
+                            "Resource": "SECRET_URI",
+                        },
+                        {
+                            "Action": [
+                                "mq:DescribeBroker",
+                            ],
+                            "Effect": "Allow",
+                            "Resource": "BROKER_ARN",
+                        },
+                        {
+                            "Action": "kms:Decrypt",
+                            "Effect": "Allow",
+                            "Resource": {
+                                "Fn::Sub": "arn:${AWS::Partition}:kms:${AWS::Region}:${AWS::AccountId}:key/1abc23d4-567f-8ab9-cde0-1fab234c5d67"
+                            },
+                        },
+                    ]
+                },
+            }
+        ]
+        self.assertEqual(policy_statements, expected_policy_document)

tests/model/eventsources/test_self_managed_kafka_event_source.py

@@ -1,6 +1,7 @@
 from unittest import TestCase
 from samtranslator.model.eventsources.pull import SelfManagedKafka
 from samtranslator.model.exceptions import InvalidEventException
+from parameterized import parameterized
 
 
 class SelfManagedKafkaEventSource(TestCase):
@@ -294,3 +295,27 @@ def test_must_raise_for_wrong_source_access_configurations_uri(self):
             self.kafka_event_source.SourceAccessConfigurations = config
             with self.assertRaises(InvalidEventException):
                 self.kafka_event_source.get_policy_statements()
+
+    @parameterized.expand(
+        [
+            (1,),
+            (True,),
+            (["1abc23d4-567f-8ab9-cde0-1fab234c5d67"],),
+            ({"KmsKeyId": "1abc23d4-567f-8ab9-cde0-1fab234c5d67"},),
+        ]
+    )
+    def test_must_validate_secrets_manager_kms_key_id(self, kms_key_id_value):
+        self.kafka_event_source.SourceAccessConfigurations = [
+            {"Type": "SASL_SCRAM_256_AUTH", "URI": "SECRET_URI"},
+            {"Type": "VPC_SUBNET", "URI": "SECRET_URI"},
+            {"Type": "VPC_SECURITY_GROUP", "URI": "SECRET_URI"},
+        ]
+        self.kafka_event_source.Topics = ["Topics"]
+        self.kafka_event_source.KafkaBootstrapServers = ["endpoint1", "endpoint2"]
+        self.kafka_event_source.Enabled = True
+        self.kafka_event_source.BatchSize = 1
+        self.kafka_event_source.SecretsManagerKmsKeyId = kms_key_id_value
+        error_message = "(None, 'Provided SecretsManagerKmsKeyId should be of type str.')"
+        with self.assertRaises(InvalidEventException) as error:
+            self.kafka_event_source.get_policy_statements()
+        self.assertEqual(error_message, str(error.exception))

tests/translator/input/error_function_with_invalid_kms_type_for_self_managed_kafka.yaml

@@ -0,0 +1,31 @@
+AWSTemplateFormatVersion: '2010-09-09'
+Parameter:
+  SecretsManagerKmsKeyIdValue:
+    Type: String
+    Default: 1abc23d4-567f-8ab9-cde0-1fab234c5d67
+Resources:
+  KafkaFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      CodeUri: s3://sam-demo-bucket/kafka.zip
+      Handler: index.kafka_handler
+      Runtime: python3.9
+      Events:
+        MyKafkaCluster:
+          Type: SelfManagedKafka
+          Properties:
+            KafkaBootstrapServers:
+              - "abc.xyz.com:9092"
+              - "123.45.67.89:9096"
+            Topics:
+              - "Topic1"
+            SourceAccessConfigurations:
+              - Type: SASL_SCRAM_512_AUTH
+                URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+              - Type: VPC_SUBNET
+                URI: subnet:subnet-12345
+              - Type: VPC_SECURITY_GROUP
+                URI: security_group:sg-67890 
+            SecretsManagerKmsKeyId:
+              Ref: SecretsManagerKmsKeyIdValue
+

tests/translator/input/error_function_with_mq_kms_invalid_type.yaml

@@ -0,0 +1,24 @@
+Parameter:
+  SecretsManagerKmsKeyIdValue:
+    Type: String
+    Default: 1abc23d4-567f-8ab9-cde0-1fab234c5d67
+
+Resources:
+  MQFunction:
+    Type: 'AWS::Serverless::Function'
+    Properties:
+      CodeUri: s3://sam-demo-bucket/queues.zip
+      Handler: queue.mq_handler
+      Runtime: python2.7
+      Events:
+        MyMQQueue:
+          Type: MQ
+          Properties:
+            Broker: arn:aws:mq:us-east-2:123456789012:broker:MyBroker:b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9
+            Queues:
+              - "Queue1"
+            SourceAccessConfigurations:
+              - Type: BASIC_AUTH
+                URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+            SecretsManagerKmsKeyId:
+              Ref: SecretsManagerKmsKeyIdValue

tests/translator/output/error_function_with_invalid_kms_type_for_self_managed_kafka.json

@@ -0,0 +1,8 @@
+{
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [KafkaFunction] is invalid. Event with id [MyKafkaCluster] is invalid. Provided SecretsManagerKmsKeyId should be of type str.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [KafkaFunction] is invalid. Event with id [MyKafkaCluster] is invalid. Provided SecretsManagerKmsKeyId should be of type str."
+    }
+  ]
+}

tests/translator/output/error_function_with_mq_kms_invalid_type.json

@@ -0,0 +1,8 @@
+{
+  "errorMessage": "Invalid Serverless Application Specification document. Number of errors found: 1. Resource with id [MQFunction] is invalid. Event with id [MyMQQueue] is invalid. Provided SecretsManagerKmsKeyId should be of type str.",
+  "errors": [
+    {
+      "errorMessage": "Resource with id [MQFunction] is invalid. Event with id [MyMQQueue] is invalid. Provided SecretsManagerKmsKeyId should be of type str."
+    }
+  ]
+}