feat(azure): add new policies for Azure Synapse

checkov/terraform/checks/graph_checks/azure/SynapseLogMonitoringEnabledForSQLPool.yaml

@@ -0,0 +1,39 @@
+metadata: 
+  id: "CKV2_AZURE_54"
+  name: "Ensure log monitoring is enabled for Synapse SQL Pool"
+  category: "GENERAL_SECURITY"
+
+definition:
+  and:
+    - cond_type: connection
+      resource_types:
+        - azurerm_synapse_sql_pool
+      connected_resource_types:
+        - azurerm_synapse_sql_pool_extended_auditing_policy
+      operator: exists
+    - cond_type: filter
+      attribute: resource_type
+      value:
+        - azurerm_synapse_sql_pool
+      operator: within
+
+    - or:
+      - and:
+        - cond_type: attribute
+          resource_types:
+            - azurerm_synapse_sql_pool_extended_auditing_policy
+          attribute: log_monitoring_enabled
+          operator: exists
+
+        - cond_type: attribute
+          resource_types:
+            - azurerm_synapse_sql_pool_extended_auditing_policy
+          attribute: 'log_monitoring_enabled'
+          operator: equals
+          value: true
+
+      - cond_type: attribute
+        resource_types:
+          - azurerm_synapse_sql_pool_extended_auditing_policy
+        attribute: log_monitoring_enabled
+        operator: not_exists

checkov/terraform/checks/graph_checks/azure/SynapseSQLPoolHasSecurityAlertPolicy.yaml

@@ -0,0 +1,24 @@
+metadata: 
+  id: "CKV2_AZURE_51"
+  name: "Ensure Synapse SQL Pool has a security alert policy"
+  category: "GENERAL_SECURITY"
+
+definition:
+  and:
+    - cond_type: connection
+      resource_types:
+        - azurerm_synapse_sql_pool
+      connected_resource_types:
+        - azurerm_synapse_sql_pool_security_alert_policy
+      operator: exists
+    - cond_type: attribute
+      resource_types:
+        - azurerm_synapse_sql_pool_security_alert_policy
+      attribute: 'policy_state'
+      operator: equals
+      value: 'Enabled'
+    - cond_type: filter
+      attribute: resource_type
+      value:
+        - azurerm_synapse_sql_pool
+      operator: within

checkov/terraform/checks/graph_checks/azure/SynapseSQLPoolHasVulnerabilityAssessment.yaml

@@ -0,0 +1,32 @@
+metadata: 
+  id: "CKV2_AZURE_52"
+  name: "Ensure Synapse SQL Pool has vulnerability assessment attached"
+  category: "GENERAL_SECURITY"
+
+definition:
+  and:
+    - resource_types:
+        - azurerm_synapse_sql_pool
+      connected_resource_types:
+        - azurerm_synapse_sql_pool_security_alert_policy
+      operator: exists
+      cond_type: connection
+    - resource_types:
+        - azurerm_synapse_sql_pool_security_alert_policy
+      connected_resource_types:
+        - azurerm_synapse_sql_pool_vulnerability_assessment
+      operator: exists
+      cond_type: connection
+    - cond_type: attribute
+      resource_types:
+        - azurerm_synapse_sql_pool_vulnerability_assessment
+      attribute: 'recurring_scans.*.enabled'
+      operator: equals
+      value: true
+    - cond_type: filter
+      attribute: resource_type
+      value:
+        - azurerm_synapse_sql_pool_security_alert_policy
+      operator: within
+
+

checkov/terraform/checks/graph_checks/azure/SynapseWorkspaceHasExtendedAuditLogs.yaml

@@ -0,0 +1,18 @@
+metadata: 
+  id: "CKV2_AZURE_53"
+  name: "Ensure Azure Synapse Workspace has extended audit logs"
+  category: "GENERAL_SECURITY"
+
+definition:
+  and:
+    - cond_type: filter
+      attribute: resource_type
+      value:
+        - azurerm_synapse_workspace
+      operator: within
+    - cond_type: connection
+      resource_types:
+        - azurerm_synapse_workspace
+      connected_resource_types:
+        - azurerm_synapse_workspace_extended_auditing_policy
+      operator: exists

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

@@ -0,0 +1,20 @@
+from checkov.common.models.enums import CheckResult, CheckCategories
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+from typing import List
+
+
+class SynapseWorkspaceAdministratorLoginPasswordHidden(BaseResourceCheck):
+    def __init__(self):
+        name = "Ensure Azure Synapse Workspace administrator login password is not exposed"
+        id = "CKV_AZURE_240"
+        supported_resources = ['azurerm_synapse_workspace']
+        categories = [CheckCategories.ENCRYPTION]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def scan_resource_conf(self, conf):
+        if 'sql_administrator_login_password' in conf:
+            return CheckResult.FAILED
+        return CheckResult.PASSED
+
+
+check = SynapseWorkspaceAdministratorLoginPasswordHidden()

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

@@ -0,0 +1,20 @@
+from checkov.common.models.consts import ANY_VALUE
+from checkov.terraform.checks.resource.base_resource_value_check import BaseResourceValueCheck
+from checkov.common.models.enums import CheckCategories
+
+
+class SynapseWorkspaceCMKEncryption(BaseResourceValueCheck):
+    def __init__(self):
+        name = "Ensure Azure Synapse Workspace is encrypted with a CMK"
+        id = "CKV_AZURE_241"
+        supported_resources = ['azurerm_synapse_workspace']
+        categories = [CheckCategories.ENCRYPTION]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources)
+
+    def get_inspected_key(self):
+        return "customer_managed_key/[0]/key_name"
+
+    def get_expected_value(self):
+        return ANY_VALUE
+
+check = SynapseWorkspaceCMKEncryption()

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

@@ -0,0 +1,29 @@
+resource "azurerm_resource_group" "example" {
+  name     = "example-resources"
+  location = "West Europe"
+}
+
+resource "azurerm_synapse_workspace" "azurerm_synapse_workspace_pass" {
+  name                                 = "MyAzureSynapseWorkspace"
+  resource_group_name                  = azurerm_resource_group.example.name
+  location                             = azurerm_resource_group.example.location
+  storage_data_lake_gen2_filesystem_id = azurerm_storage_data_lake_gen2_filesystem.example.id
+  sql_administrator_login              = "sqladminuser"
+
+  identity {
+    type = "SystemAssigned"
+  }
+}
+
+resource "azurerm_synapse_workspace" "azurerm_synapse_workspace_fail" {
+  name                                 = "MyAzureSynapseWorkspace"
+  resource_group_name                  = azurerm_resource_group.example.name
+  location                             = azurerm_resource_group.example.location
+  storage_data_lake_gen2_filesystem_id = azurerm_storage_data_lake_gen2_filesystem.example.id
+  sql_administrator_login              = "sqladminuser"
+  sql_administrator_login_password     = "H@Sh1CoR3!"
+
+  identity {
+    type = "SystemAssigned"
+  }
+}

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

@@ -0,0 +1,34 @@
+resource "azurerm_resource_group" "example" {
+  name     = "example-resources"
+  location = "West Europe"
+}
+
+resource "azurerm_synapse_workspace" "azurerm_synapse_workspace_pass" {
+  name                                 = "MyAzureSynapseWorkspace"
+  resource_group_name                  = azurerm_resource_group.example.name
+  location                             = azurerm_resource_group.example.location
+  storage_data_lake_gen2_filesystem_id = azurerm_storage_data_lake_gen2_filesystem.example.id
+  sql_administrator_login              = "sqladminuser"
+  sql_administrator_login_password     = "H@Sh1CoR3!"
+  customer_managed_key {
+    key_versionless_id = azurerm_key_vault_key.example.versionless_id
+    key_name           = "enckey"
+  }
+
+  identity {
+    type = "SystemAssigned"
+  }
+}
+
+resource "azurerm_synapse_workspace" "azurerm_synapse_workspace_fail" {
+  name                                 = "MyAzureSynapseWorkspace"
+  resource_group_name                  = azurerm_resource_group.example.name
+  location                             = azurerm_resource_group.example.location
+  storage_data_lake_gen2_filesystem_id = azurerm_storage_data_lake_gen2_filesystem.example.id
+  sql_administrator_login              = "sqladminuser"
+  sql_administrator_login_password     = "H@Sh1CoR3!"
+
+  identity {
+    type = "SystemAssigned"
+  }
+}

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

@@ -0,0 +1,39 @@
+import os
+import unittest
+
+from checkov.runner_filter import RunnerFilter
+from checkov.terraform.checks.resource.azure.SynapseWorkspaceAdministratorLoginPasswordHidden import check
+from checkov.terraform.runner import Runner
+
+class TestSynapseWorkspaceAdministratorLoginPasswordHidden(unittest.TestCase):
+    def test(self):
+        runner = Runner()
+        current_dir = os.path.dirname(os.path.realpath(__file__))
+
+        test_files_dir = current_dir + "/example_SynapseWorkspaceAdministratorLoginPasswordHidden"
+        report = runner.run(
+            root_folder=test_files_dir, runner_filter=RunnerFilter(checks=[check.id])
+        )
+        summary = report.get_summary()
+
+        passing_resources = {
+            "azurerm_synapse_workspace.azurerm_synapse_workspace_pass",
+        }
+        failing_resources = {
+            "azurerm_synapse_workspace.azurerm_synapse_workspace_fail",
+        }
+
+        passed_check_resources = set([c.resource for c in report.passed_checks])
+        failed_check_resources = set([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/azure/test_SynapseWorkspaceCMKEncryption.py

@@ -0,0 +1,40 @@
+import os
+import unittest
+
+from checkov.runner_filter import RunnerFilter
+from checkov.terraform.checks.resource.azure.SynapseWorkspaceCMKEncryption import check
+from checkov.terraform.runner import Runner
+
+
+class TestSynapseWorkspaceCMKEncryption(unittest.TestCase):
+    def test(self):
+        runner = Runner()
+        current_dir = os.path.dirname(os.path.realpath(__file__))
+
+        test_files_dir = current_dir + "/example_SynapseWorkspaceCMKEncryption"
+        report = runner.run(
+            root_folder=test_files_dir, runner_filter=RunnerFilter(checks=[check.id])
+        )
+        summary = report.get_summary()
+
+        passing_resources = {
+            "azurerm_synapse_workspace.azurerm_synapse_workspace_pass",
+        }
+        failing_resources = {
+            "azurerm_synapse_workspace.azurerm_synapse_workspace_fail",
+        }
+
+        passed_check_resources = set([c.resource for c in report.passed_checks])
+        failed_check_resources = set([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/graph/checks/resources/SynapseLogMonitoringEnabledForSQLPool/expected.yaml

@@ -0,0 +1,6 @@
+pass:
+  - "azurerm_synapse_sql_pool.azurerm_synapse_sql_pool_pass_A"
+  - "azurerm_synapse_sql_pool.azurerm_synapse_sql_pool_pass_B"
+fail:
+  - "azurerm_synapse_sql_pool.azurerm_synapse_sql_pool_fail_A"
+  - "azurerm_synapse_sql_pool.azurerm_synapse_sql_pool_fail_B"