feat(cloudformation): Adding Sse property to SQS Encryption check

checkov/cloudformation/checks/resource/aws/SQSQueueEncryption.py

@@ -1,23 +1,23 @@
-from typing import Any
+from checkov.common.models.enums import CheckCategories, CheckResult
+from checkov.cloudformation.checks.resource.base_resource_value_check import BaseResourceCheck
 
-from checkov.common.models.enums import CheckCategories
-from checkov.cloudformation.checks.resource.base_resource_value_check import BaseResourceValueCheck
-from checkov.common.models.consts import ANY_VALUE
 
-
-class SQSQueueEncryption(BaseResourceValueCheck):
+class SQSQueueEncryption(BaseResourceCheck):
     def __init__(self) -> None:
         name = "Ensure all data stored in the SQS queue is encrypted"
         id = "CKV_AWS_27"
         supported_resources = ['AWS::SQS::Queue']
         categories = [CheckCategories.ENCRYPTION]
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
-    def get_inspected_key(self) -> str:
-        return 'Properties/KmsMasterKeyId'
+    def scan_resource_conf(self, conf):
+        properties = conf.get("Properties", {})
+        kms_key_id_value = properties.get("KmsMasterKeyId")
+        sse_enabled_value = properties.get("SqsManagedSseEnabled")
 
-    def get_expected_value(self) -> Any:
-        return ANY_VALUE
+        if kms_key_id_value or sse_enabled_value:
+            return CheckResult.PASSED
+        return CheckResult.FAILED
 
 
 check = SQSQueueEncryption()

tests/cloudformation/checks/resource/aws/example_SQSQueueEncryption/test_SQSQueueEncryption-FAILED3.yml

@@ -0,0 +1,9 @@
+AWSTemplateFormatVersion: "2010-09-09"
+Resources:
+  MySourceQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      RedrivePolicy:
+        deadLetterTargetArn: "example_arn"
+        maxReceiveCount: 5
+        SqsManagedSseEnabled: false

tests/cloudformation/checks/resource/aws/example_SQSQueueEncryption/test_SQSQueueEncryption-PASSED2.yml

@@ -0,0 +1,9 @@
+AWSTemplateFormatVersion: "2010-09-09"
+Resources:
+  MySourceQueue:
+    Type: AWS::SQS::Queue
+    Properties:
+      RedrivePolicy:
+        deadLetterTargetArn: "example_arn"
+        maxReceiveCount: 5
+      SqsManagedSseEnabled: true

tests/cloudformation/checks/resource/aws/test_SQSQueueEncryption.py

@@ -16,8 +16,8 @@ def test_summary(self):
         report = runner.run(root_folder=test_files_dir, runner_filter=RunnerFilter(checks=[check.id]))
         summary = report.get_summary()
 
-        self.assertEqual(summary['passed'], 1)
-        self.assertEqual(summary['failed'], 2)
+        self.assertEqual(summary['passed'], 2)
+        self.assertEqual(summary['failed'], 3)
         self.assertEqual(summary['skipped'], 0)
         self.assertEqual(summary['parsing_errors'], 0)