Repository: bridgecrewio/checkov

Issue:
False positive for check 'MSKClusterEncryption' (CKV_AWS_81)
**Describe the bug**
CKV_AWS_81 is reporting MSK clusters as unencrypted at rest while they are encrypted at rest.

**To Reproduce**
Example Terraform code producing a cluster with encryption (it will use the default KMS key as documented in https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_cluster#encryption_at_rest_kms_key_arn)

(slightly adapted from example on the msk_cluster resource docs)
<details>
  <summary>Code</summary>

```
resource "aws_vpc" "vpc" {
  cidr_block = "192.168.0.0/22"
}

data "aws_availability_zones" "azs" {
  state = "available"
}

resource "aws_subnet" "subnet_az1" {
  availability_zone = data.aws_availability_zones.azs.names[0]
  cidr_block        = "192.168.0.0/24"
  vpc_id            = aws_vpc.vpc.id
}

resource "aws_subnet" "subnet_az2" {
  availability_zone = data.aws_availability_zones.azs.names[1]
  cidr_block        = "192.168.1.0/24"
  vpc_id            = aws_vpc.vpc.id
}

resource "aws_subnet" "subnet_az3" {
  availability_zone = data.aws_availability_zones.azs.names[2]
  cidr_block        = "192.168.2.0/24"
  vpc_id            = aws_vpc.vpc.id
}

resource "aws_security_group" "sg" {
  vpc_id = aws_vpc.vpc.id
}

resource "aws_kms_key" "kms" {
  description = "example"
}

resource "aws_cloudwatch_log_group" "test" {
  name = "msk_broker_logs"
}

resource "aws_s3_bucket" "bucket" {
  bucket = "msk-broker-logs-bucket"
  acl    = "private"
}

resource "aws_iam_role" "firehose_role" {
  name = "firehose_test_role"

  assume_role_policy = <<EOF
{
"Version": "2012-10-17",
"Statement": [
  {
    "Action": "sts:AssumeRole",
    "Principal": {
      "Service": "firehose.amazonaws.com"
    },
    "Effect": "Allow",
    "Sid": ""
  }
  ]
}
EOF
}

resource "aws_kinesis_firehose_delivery_stream" "test_stream" {
  name        = "terraform-kinesis-firehose-msk-broker-logs-stream"
  destination = "s3"

  s3_configuration {
    role_arn   = aws_iam_role.firehose_role.arn
    bucket_arn = aws_s3_bucket.bucket.arn
  }

  tags = {
    LogDeliveryEnabled = "placeholder"
  }

  lifecycle {
    ignore_changes = [
      tags["LogDeliveryEnabled"],
    ]
  }
}

resource "aws_msk_cluster" "example" {
  cluster_name           = "example"
  kafka_version          = "2.4.1"
  number_of_broker_nodes = 3

  broker_node_group_info {
    instance_type   = "kafka.m5.large"
    ebs_volume_size = 1000
    client_subnets = [
      aws_subnet.subnet_az1.id,
      aws_subnet.subnet_az2.id,
      aws_subnet.subnet_az3.id,
    ]
    security_groups = [aws_security_group.sg.id]
  }

  encryption_info {
    encryption_in_transit {                                                                                                                                        
      client_broker = "TLS"
      in_cluster    = true                                                                                                                                         
    }
  }

  open_monitoring {
    prometheus {
      jmx_exporter {
        enabled_in_broker = true
      }
      node_exporter {
        enabled_in_broker = true
      }
    }
  }

  logging_info {
    broker_logs {
      cloudwatch_logs {
        enabled   = true
        log_group = aws_cloudwatch_log_group.test.name
      }
      firehose {
        enabled         = true
        delivery_stream = aws_kinesis_firehose_delivery_stream.test_stream.name
      }
      s3 {
        enabled = true
        bucket  = aws_s3_bucket.bucket.id
        prefix  = "logs/msk-"
      }
    }
  }

  tags = {
    foo = "bar"
  }
}

output "zookeeper_connect_string" {
  value = aws_msk_cluster.example.zookeeper_connect_string
}

output "bootstrap_brokers_tls" {
  description = "TLS connection host:port pairs"
  value       = aws_msk_cluster.example.bootstrap_brokers_tls
}
```

</details>

**Expected behavior**
No error is thrown as long as an encryption_info block is defined


 Checkov Version 1.0.669


Discussion:

Patch:
diff --git a/checkov/terraform/checks/resource/aws/MSKClusterEncryption.py b/checkov/terraform/checks/resource/aws/MSKClusterEncryption.py
index 8e7bfee51..d0aa56f3e 100644
--- a/checkov/terraform/checks/resource/aws/MSKClusterEncryption.py
+++ b/checkov/terraform/checks/resource/aws/MSKClusterEncryption.py
@@ -11,16 +11,18 @@ class MSKClusterEncryption(BaseResourceCheck):
         super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
 
     def scan_resource_conf(self, conf):
+        # Note: As long as the 'encryption_info' block is specified, the cluster
+        # will be encrypted at rest even if 'encryption_at_rest_kms_key_arn' is not specified
+        # See https://registry.terraform.io/providers/hashicorp/aws/latest/docs/resources/msk_cluster#encryption_at_rest_kms_key_arn
         if 'encryption_info' in conf.keys():
             encryption = conf['encryption_info'][0]
-            if 'encryption_at_rest_kms_key_arn' in encryption:
-                if 'encryption_in_transit' in encryption:
-                    transit = encryption['encryption_in_transit'][0]
-                    if 'client_broker' in transit and transit['client_broker'][0] != 'TLS' or \
-                            'in_cluster' in transit and transit['in_cluster'][0] is False:
-                        return CheckResult.FAILED
-                    return CheckResult.PASSED
+            if 'encryption_in_transit' in encryption:
+                transit = encryption['encryption_in_transit'][0]
+                if 'client_broker' in transit and transit['client_broker'][0] != 'TLS' or \
+                        'in_cluster' in transit and transit['in_cluster'][0] is False:
+                    return CheckResult.FAILED
                 return CheckResult.PASSED
+            return CheckResult.PASSED
         return CheckResult.FAILED
 
 


Test patch:
diff --git a/tests/terraform/checks/resource/aws/test_MSKClusterEncryption.py b/tests/terraform/checks/resource/aws/test_MSKClusterEncryption.py
index 23fd0bdf1..0cfd55402 100644
--- a/tests/terraform/checks/resource/aws/test_MSKClusterEncryption.py
+++ b/tests/terraform/checks/resource/aws/test_MSKClusterEncryption.py
@@ -13,23 +13,6 @@ class TestMSKClusterEncryption(unittest.TestCase):
         scan_result = check.scan_resource_conf(conf=resource_conf)
         self.assertEqual(CheckResult.FAILED, scan_result)
 
-    def test_failure_no_kms(self):
-        resource_conf = {
-            "name": "test-project",
-            "encryption_info": [
-                {
-                    "encryption_in_transit": [
-                        {
-                            "client_broker": ["TLS"],
-                            "in_cluster": ["true"],
-                        }
-                    ],
-                }
-            ],
-        }
-        scan_result = check.scan_resource_conf(conf=resource_conf)
-        self.assertEqual(CheckResult.FAILED, scan_result)
-
     def test_failure_non_tls(self):
         resource_conf = {
             "name": "test-project",
@@ -96,6 +79,33 @@ class TestMSKClusterEncryption(unittest.TestCase):
         scan_result = check.scan_resource_conf(conf=resource_conf)
         self.assertEqual(CheckResult.PASSED, scan_result)
 
+    # Regression test for https://github.com/bridgecrewio/checkov/issues/747
+    def test_success_no_encryption_at_rest_kms_key_arn_specified(self):
+        resource_conf = {
+            "name": "test-project",
+            "encryption_info": [{}],
+        }
+        scan_result = check.scan_resource_conf(conf=resource_conf)
+        self.assertEqual(CheckResult.PASSED, scan_result)
+
+    # Regression test for https://github.com/bridgecrewio/checkov/issues/747
+    def test_success_encryption_in_transit_and_no_encryption_at_rest_kms_key_arn_specified(self):
+        resource_conf = {
+            "name": "test-project",
+            "encryption_info": [
+                {
+                    "encryption_in_transit": [
+                        {
+                            "client_broker": ["TLS"],
+                            "in_cluster": [True],
+                        }
+                    ],
+                }
+            ],
+        }
+        scan_result = check.scan_resource_conf(conf=resource_conf)
+        self.assertEqual(CheckResult.PASSED, scan_result)
+
 
 if __name__ == '__main__':
     unittest.main()