feat(terraform): Add new checks for GCP and Azure

checkov/terraform/checks/resource/azure/SQLDatabaseThreatDetectionEnabled.py

@@ -0,0 +1,21 @@
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+
+
+class SQLDatabaseThreatDetectionEnabled(BaseResourceCheck):
+    def __init__(self):
+        name = "Ensure that Threat Detection is enabled for SQL Database"
+        id = "CKV_AZURE_101"
+        supported_resources = ['azurerm_sql_database']
+        categories = [CheckCategories.LOGGING]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf):
+        if 'threat_detection_policy' in conf:
+            policy = conf['threat_detection_policy'][0]
+            if 'state' in policy and policy['state'][0].upper() == 'ENABLED':
+                return CheckResult.PASSED
+        return CheckResult.FAILED
+
+
+check = SQLDatabaseThreatDetectionEnabled()

checkov/terraform/checks/resource/gcp/MemorystoreForRedisPublicNetworkAccess.py

@@ -0,0 +1,19 @@
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+
+
+class MemorystoreForRedisPublicNetworkAccess(BaseResourceCheck):
+    def __init__(self):
+        name = "Ensure Memorystore for Redis is not exposed to public internet"
+        id = "CKV_GCP_99"
+        supported_resources = ['google_redis_instance']
+        categories = [CheckCategories.NETWORKING]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf):
+        if 'connect_mode' in conf and conf['connect_mode'][0] == 'DIRECT_PEERING':
+            return CheckResult.FAILED
+        return CheckResult.PASSED
+
+
+check = MemorystoreForRedisPublicNetworkAccess()

tests/terraform/checks/resource/azure/example_SQLDatabaseThreatDetectionEnabled/main.tf

@@ -0,0 +1,32 @@
+
+resource "azurerm_resource_group" "example" {
+  name     = "example-resources"
+  location = "West Europe"
+}
+
+resource "azurerm_sql_server" "example" {
+  name                         = "examplesql"
+  resource_group_name          = azurerm_resource_group.example.name
+  location                     = azurerm_resource_group.example.location
+  version                      = "12.0"
+  administrator_login          = "4dm1n157r470r"
+  administrator_login_password = "Password1234!"
+}
+
+resource "azurerm_sql_database" "fail" {
+  name                = "sqldb-fail"
+  resource_group_name = azurerm_resource_group.example.name
+  location            = azurerm_resource_group.example.location
+  server_name         = azurerm_sql_server.example.name
+}
+
+resource "azurerm_sql_database" "pass" {
+  name                = "sqldb-pass"
+  resource_group_name = azurerm_resource_group.example.name
+  location            = azurerm_resource_group.example.location
+  server_name         = azurerm_sql_server.example.name
+
+  threat_detection_policy {
+    state = "Enabled"
+  }
+}

tests/terraform/checks/resource/azure/test_SQLDatabaseThreatDetectionEnabled.py

@@ -0,0 +1,40 @@
+import unittest
+from pathlib import Path
+
+from checkov.runner_filter import RunnerFilter
+from checkov.terraform.runner import Runner
+from checkov.terraform.checks.resource.azure.SQLDatabaseThreatDetectionEnabled import check
+
+
+class TestSQLDatabaseThreatDetectionEnabled(unittest.TestCase):
+    def test(self):
+        # given
+        test_files_dir = Path(__file__).parent / "example_SQLDatabaseThreatDetectionEnabled"
+
+        # when
+        report = Runner().run(root_folder=str(test_files_dir), runner_filter=RunnerFilter(checks=[check.id]))
+
+        # then
+        summary = report.get_summary()
+
+        passing_resources = {
+            "azurerm_sql_database.pass",
+        }
+        failing_resources = {
+            "azurerm_sql_database.fail",
+        }
+
+        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"], 1)
+        self.assertEqual(summary["failed"], 1)
+        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()

tests/terraform/checks/resource/gcp/example_MemorystoreForRedisPublicNetworkAccess/main.tf

@@ -0,0 +1,10 @@
+
+resource "google_redis_instance" "fail" {
+  name           = "fail-instance"
+  connect_mode   = "DIRECT_PEERING"
+}
+
+resource "google_redis_instance" "pass" {
+  name           = "pass-instance"
+  connect_mode   = "PRIVATE_SERVICE_ACCESS"
+}

tests/terraform/checks/resource/gcp/test_MemorystoreForRedisPublicNetworkAccess.py

@@ -0,0 +1,40 @@
+import unittest
+from pathlib import Path
+
+from checkov.runner_filter import RunnerFilter
+from checkov.terraform.runner import Runner
+from checkov.terraform.checks.resource.gcp.MemorystoreForRedisPublicNetworkAccess import check
+
+
+class TestMemorystoreForRedisPublicNetworkAccess(unittest.TestCase):
+    def test(self):
+        # given
+        test_files_dir = Path(__file__).parent / "example_MemorystoreForRedisPublicNetworkAccess"
+
+        # when
+        report = Runner().run(root_folder=str(test_files_dir), runner_filter=RunnerFilter(checks=[check.id]))
+
+        # then
+        summary = report.get_summary()
+
+        passing_resources = {
+            "google_redis_instance.pass",
+        }
+        failing_resources = {
+            "google_redis_instance.fail",
+        }
+
+        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"], 1)
+        self.assertEqual(summary["failed"], 1)
+        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()