fix(terraform): correctly evaluate CKV_AWS_37 when there's a dynamic … (#6792)

* fix(terraform): correctly evaluate CKV_AWS_37 when there's a dynamic block

* Update EKSControlPlaneLogging.py

fix linting and access pattern

---------

Co-authored-by: Tj <[email protected]>

Author: Alex-Waring

checkov/terraform/checks/resource/aws/EKSControlPlaneLogging.py

@@ -19,10 +19,15 @@ def scan_resource_conf(self, conf):
         :return: <CheckResult>
         """
         log_types = ["api", "audit", "authenticator", "controllerManager", "scheduler"]
-        if "enabled_cluster_log_types" in conf.keys() and conf["enabled_cluster_log_types"] and \
-                conf["enabled_cluster_log_types"][0] is not None \
-                and all(elem in conf["enabled_cluster_log_types"][0] for elem in log_types):
-            return CheckResult.PASSED
+        enabled_cluster_log_types = conf.get("enabled_cluster_log_types")
+        if enabled_cluster_log_types and enabled_cluster_log_types[0] is not None:
+            enabled_cluster_log_types = enabled_cluster_log_types[0]
+            if isinstance(enabled_cluster_log_types[0], str):
+                if all(elem in enabled_cluster_log_types for elem in log_types):
+                    return CheckResult.PASSED
+            elif isinstance(enabled_cluster_log_types[0], list):
+                if all([elem] in enabled_cluster_log_types for elem in log_types):
+                    return CheckResult.PASSED
         return CheckResult.FAILED
 
     def get_evaluated_keys(self) -> List[str]:

tests/terraform/checks/resource/aws/example_EKSControlPlaneLogging/main.tf

@@ -0,0 +1,55 @@
+# pass
+
+resource "aws_eks_cluster" "fully_enabled" {
+  name     = "example"
+  role_arn = "aws_iam_role.arn"
+
+  enabled_cluster_log_types = [
+    "api",
+    "audit",
+    "authenticator",
+    "controllerManager",
+    "scheduler"
+  ]
+}
+
+resource "aws_eks_cluster" "fully_enabled_with_dynamic_block" {
+  name     = "example"
+  role_arn = "aws_iam_role.arn"
+
+  enabled_cluster_log_types = [
+    "api",
+    "audit",
+    "authenticator",
+    "controllerManager",
+    "scheduler"
+  ]
+
+  dynamic "encryption_config" {
+    for_each = [1]
+
+    content {
+      provider {
+        key_arn = "aws/kms/key"
+      }
+      resources = ["secrets"]
+    }
+  }
+}
+
+# fail
+
+resource "aws_eks_cluster" "partially_enabled" {
+  name     = "example"
+  role_arn = "aws_iam_role.arn"
+
+  enabled_cluster_log_types = [
+    "api",
+    "audit"
+  ]
+}
+
+resource "aws_eks_cluster" "not_configured" {
+  name     = "example"
+  role_arn = "aws_iam_role.arn"
+}

tests/terraform/checks/resource/aws/test_EKSControlPlaneLogging.py

@@ -1,6 +1,9 @@
 import unittest
+from pathlib import Path
 
+from checkov.runner_filter import RunnerFilter
 from checkov.terraform.checks.resource.aws.EKSControlPlaneLogging import check
+from checkov.terraform.runner import Runner
 from checkov.common.models.enums import CheckResult
 
 
@@ -23,12 +26,43 @@ def test_success(self):
         scan_result = check.scan_resource_conf(conf=resource_conf)
         self.assertEqual(CheckResult.PASSED, scan_result)
 
-    def test_success(self):
+    def test_failure_not_enabled(self):
         resource_conf = {'name': ['testcluster'], 'enabled_cluster_log_types': []}
 
         scan_result = check.scan_resource_conf(conf=resource_conf)
         self.assertEqual(CheckResult.FAILED, scan_result)
 
+    def test_file(self):
+        # given
+        test_files_dir = Path(__file__).parent / "example_EKSControlPlaneLogging"
+
+        # when
+        report = Runner().run(root_folder=str(test_files_dir), runner_filter=RunnerFilter(checks=[check.id]))
+
+        # then
+        summary = report.get_summary()
+
+        passing_resources = {
+            "aws_eks_cluster.fully_enabled",
+            "aws_eks_cluster.fully_enabled_with_dynamic_block"
+        }
+        failing_resources = {
+            "aws_eks_cluster.partially_enabled",
+            "aws_eks_cluster.not_configured"
+        }
+
+        passed_check_resources = {c.resource for c in report.passed_checks}
+        failed_check_resources = {c.resource for c in report.failed_checks}
+
+        self.assertEqual(summary["passed"], 2)
+        self.assertEqual(summary["failed"], 2)
+        self.assertEqual(summary["skipped"], 0)
+        self.assertEqual(summary["parsing_errors"], 0)
+
+        self.assertEqual(passing_resources, passed_check_resources)
+        self.assertEqual(failing_resources, failed_check_resources)
+
+
 
 if __name__ == '__main__':
     unittest.main()