feat(terraform): add CKV_AZURE_248 - Azure batch account network access restriction

checkov/arm/checks/resource/AzureBatchAccountEndpointAccessDefaultAction.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.arm.base_resource_check import BaseResourceCheck
+from checkov.common.models.enums import CheckCategories, CheckResult
+
+
+class AzureBatchAccountEndpointAccessDefaultAction(BaseResourceCheck):
+
+    DISABLED_PUBLIC_NETWORK_ACCESS = "disabled"
+    FORBIDDEN_NETWORK_ACCESS_DEFAULT_ACTION = "allow"
+
+    def __init__(self) -> None:
+        name = "Ensure that if Azure Batch account public network access in case 'enabled' then its account access must be 'deny'"
+        id = "CKV_AZURE_248"
+        supported_resources = ("Microsoft.Batch/batchAccounts",)
+        categories = [CheckCategories.NETWORKING]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources,)
+
+    @staticmethod
+    def _exists_and_lower_equal(actual_value: Any, expected_lowercase_value: str) -> bool:
+        return actual_value and str(actual_value).lower() == expected_lowercase_value
+
+    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
+        properties = conf.get('properties')
+        if not properties or not isinstance(properties, dict):
+            return CheckResult.FAILED
+
+        public_network_access = properties.get('publicNetworkAccess')
+        # public network access is disabled, no need to check for account access default action
+        if self._exists_and_lower_equal(public_network_access, self.DISABLED_PUBLIC_NETWORK_ACCESS):
+            return CheckResult.PASSED
+
+        network_profile = properties.get('networkProfile')
+        if not network_profile:
+            return CheckResult.PASSED
+        account_access = network_profile.get('accountAccess')
+        if not account_access:
+            return CheckResult.PASSED
+        default_action = account_access.get('defaultAction')
+        if not self._exists_and_lower_equal(default_action, self.FORBIDDEN_NETWORK_ACCESS_DEFAULT_ACTION):
+            return CheckResult.PASSED
+
+        self.evaluated_keys = ["properties/networkProfile/accountAccess/defaultAction"]
+        return CheckResult.FAILED
+
+
+check = AzureBatchAccountEndpointAccessDefaultAction()

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

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Any
+
+from checkov.terraform.checks.resource.base_resource_check import BaseResourceCheck
+from checkov.common.models.enums import CheckCategories, CheckResult
+
+
+class AzureBatchAccountEndpointAccessDefaultAction(BaseResourceCheck):
+
+    def __init__(self) -> None:
+        name = "Ensure that if Azure Batch account public network access in case 'enabled' then its account access must be 'deny'"
+        id = "CKV_AZURE_248"
+        supported_resources = ("azurerm_batch_account",)
+        categories = [CheckCategories.NETWORKING]
+        super().__init__(name=name, id=id, categories=categories, supported_resources=supported_resources,)
+
+    def scan_resource_conf(self, conf: dict[str, Any]) -> CheckResult:
+
+        public_network_access = conf.get('public_network_access_enabled', [None])[0]
+        # public network access is disabled, no need to check for account access default action
+        if public_network_access is False:
+            return CheckResult.PASSED
+
+        network_profile: dict[str, Any] | None = conf.get('network_profile', [None])[0]
+        if not network_profile:
+            return CheckResult.PASSED
+        account_access: dict[str, Any] | None = network_profile.get('account_access', [None])[0]
+        if not account_access:
+            return CheckResult.PASSED
+        default_action: str | None = account_access.get('default_action', [None])[0]
+        if not default_action or str(default_action).lower() != "allow":
+            return CheckResult.PASSED
+
+        self.evaluated_keys = ["network_profile/[0]/account_access/[0]/default_action"]
+        return CheckResult.FAILED
+
+
+check = AzureBatchAccountEndpointAccessDefaultAction()

tests/arm/checks/resource/example_AzureBatchAccountEndpointAccessDefaultAction.py/fail.json

@@ -0,0 +1,31 @@
+{
+  "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
+  "contentVersion": "1.0.0.0",
+  "resources": [
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "fail_explicit_publicNetworkAccess",
+      "properties": {
+        "publicNetworkAccess": "Enabled",
+        "networkProfile": {
+          "accountAccess": {
+            "defaultAction": "Allow"
+          }
+        }
+      }
+    },
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "fail_default_publicNetworkAccess",
+      "properties": {
+        "networkProfile": {
+          "accountAccess": {
+            "defaultAction": "Allow"
+          }
+        }
+      }
+    }
+  ]
+}

tests/arm/checks/resource/example_AzureBatchAccountEndpointAccessDefaultAction.py/pass.json

@@ -0,0 +1,53 @@
+{
+  "$schema": "https://schema.management.azure.com/schemas/2019-04-01/deploymentTemplate.json#",
+  "contentVersion": "1.0.0.0",
+  "resources": [
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "pass_empty",
+      "properties": {
+      }
+    },
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "pass_publicNetworkAccess_disabled",
+      "properties": {
+        "publicNetworkAccess": "Disabled"
+      }
+    },
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "pass_publicNetworkAccess_enabled_no_network_profile",
+      "properties": {
+        "publicNetworkAccess": "Enabled"
+      }
+    },
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "pass_publicNetworkAccess_enabled_no_account_access",
+      "properties": {
+        "publicNetworkAccess": "Enabled",
+        "networkProfile": {
+
+        }
+      }
+    },
+    {
+      "type": "Microsoft.Batch/batchAccounts",
+      "apiVersion": "2024-02-01",
+      "name": "pass_publicNetworkAccess_enabled_default_action_deny",
+      "properties": {
+        "publicNetworkAccess": "Enabled",
+        "networkProfile": {
+          "accountAccess": {
+            "defaultAction": "Deny"
+          }
+        }
+      }
+    }
+  ]
+}

tests/arm/checks/resource/test_AzureBatchAccountEndpointAccessDefaultAction.py

@@ -0,0 +1,36 @@
+import unittest
+from pathlib import Path
+from checkov.arm.checks.resource.AzureBatchAccountEndpointAccessDefaultAction import check
+from checkov.arm.runner import Runner
+from checkov.runner_filter import RunnerFilter
+from tests.common.check_assertion_utils import checks_report_assertions
+
+
+class TestAzureBatchAccountEndpointAccessDefaultAction(unittest.TestCase):
+    def test_summary(self):
+        passing_resources = {
+            "Microsoft.Batch/batchAccounts.pass_empty",
+            "Microsoft.Batch/batchAccounts.pass_publicNetworkAccess_disabled",
+            "Microsoft.Batch/batchAccounts.pass_publicNetworkAccess_enabled_no_network_profile",
+            "Microsoft.Batch/batchAccounts.pass_publicNetworkAccess_enabled_no_account_access",
+            "Microsoft.Batch/batchAccounts.pass_publicNetworkAccess_enabled_default_action_deny",
+        }
+        failing_resources = {
+            "Microsoft.Batch/batchAccounts.fail_explicit_publicNetworkAccess":
+                ["properties/networkProfile/accountAccess/defaultAction"],
+            "Microsoft.Batch/batchAccounts.fail_default_publicNetworkAccess":
+                ["properties/networkProfile/accountAccess/defaultAction"],
+        }
+
+        # given
+        test_files_dir = Path(__file__).parent / "example_AzureBatchAccountEndpointAccessDefaultAction.py"
+
+        # when
+        report = Runner().run(root_folder=str(test_files_dir), runner_filter=RunnerFilter(checks=[check.id]))
+
+        # then
+        checks_report_assertions(self, report, passing_resources, failing_resources)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/common/check_assertion_utils.py

@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+import unittest
+from typing import Union
+
+from checkov.common.output.report import Report
+
+
+def checks_report_assertions(test_case: unittest.TestCase, report: Report,
+                             expected_passing_resources: set[str],
+                             expected_failing_resources: Union[set[str], dict[str, list[str]]],
+                             expected_skipped_resources: set[str] = None) -> None:
+    """
+    validates:
+    1. summary field includes correct count of passing / failing / skipped resources
+    2. the resource themselves match the expected resources
+    3. for failing resources, there's an option to send expected as dict, in which case both the resource and the evaluated keys of that check will be validated
+    """
+    if expected_skipped_resources is None:
+        expected_skipped_resources = set()
+
+    summary = report.get_summary()
+
+    passed_check_resources = {c.resource for c in report.passed_checks}
+    skipped_check_resources = {c.resource for c in report.skipped_checks}
+
+    if isinstance(expected_failing_resources, dict):
+        failed_check_resources = {c.resource: c.check_result.get("evaluated_keys") for c in report.failed_checks}
+    else:
+        failed_check_resources = {c.resource for c in report.failed_checks}
+
+    test_case.assertEqual(summary["passed"], len(expected_passing_resources))
+    test_case.assertEqual(summary["failed"], len(expected_failing_resources))
+    test_case.assertEqual(summary["skipped"], len(expected_skipped_resources))
+    test_case.assertEqual(summary["parsing_errors"], 0)
+
+    test_case.assertEqual(expected_passing_resources, passed_check_resources)
+    test_case.assertEqual(expected_failing_resources, failed_check_resources)
+    test_case.assertEqual(expected_skipped_resources, skipped_check_resources)

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

@@ -0,0 +1,72 @@
+#pass
+resource "azurerm_batch_account" "pass_no_publicNetworkAccess" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+}
+
+resource "azurerm_batch_account" "pass_publicNetworkAccess_disabled" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+  public_network_access_enabled       = false
+}
+
+resource "azurerm_batch_account" "pass_publicNetworkAccess_enabled_no_network_profile" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+  public_network_access_enabled       = true
+}
+
+resource "azurerm_batch_account" "pass_publicNetworkAccess_enabled_no_account_access" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+  public_network_access_enabled       = true
+  network_profile {
+
+  }
+}
+
+resource "azurerm_batch_account" "pass_publicNetworkAccess_enabled_default_action_deny" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+  public_network_access_enabled       = true
+  network_profile {
+    account_access {
+      default_action = "deny"
+    }
+  }
+}
+
+resource "azurerm_batch_account" "fail_publicNetworkAccess_enabled_default_action_allow" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+  public_network_access_enabled       = true
+  network_profile {
+    account_access {
+      default_action = "allow"
+    }
+  }
+}
+
+resource "azurerm_batch_account" "fail_bad_default_action_no_public_network" {
+  name                                = "testbatchaccount"
+  resource_group_name                 = "group"
+  location                            = "azurerm_resource_group.example.location"
+  pool_allocation_mode                = "BatchService"
+  network_profile {
+    account_access {
+      default_action = "allow"
+    }
+  }
+}

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

@@ -0,0 +1,38 @@
+import os
+import unittest
+from pathlib import Path
+
+from checkov.runner_filter import RunnerFilter
+from checkov.terraform.runner import Runner
+from checkov.terraform.checks.resource.azure.AzureBatchAccountEndpointAccessDefaultAction import check
+from tests.common.check_assertion_utils import checks_report_assertions
+
+
+class TestAzureBatchAccountEndpointAccessDefaultAction(unittest.TestCase):
+
+    def test(self):
+        runner = Runner()
+
+        test_files_dir = Path(__file__).parent / "example_AzureBatchAccountEndpointAccessDefaultAction"
+        report = runner.run(root_folder=str(test_files_dir),
+                            runner_filter=RunnerFilter(checks=[check.id]))
+
+        passing_resources = {
+            'azurerm_batch_account.pass_no_publicNetworkAccess',
+            'azurerm_batch_account.pass_publicNetworkAccess_disabled',
+            'azurerm_batch_account.pass_publicNetworkAccess_enabled_no_network_profile',
+            'azurerm_batch_account.pass_publicNetworkAccess_enabled_no_account_access',
+            'azurerm_batch_account.pass_publicNetworkAccess_enabled_default_action_deny',
+
+        }
+        failing_resources = {
+            'azurerm_batch_account.fail_publicNetworkAccess_enabled_default_action_allow',
+            'azurerm_batch_account.fail_bad_default_action_no_public_network',
+        }
+
+        # then
+        checks_report_assertions(self, report, passing_resources, failing_resources)
+
+
+if __name__ == '__main__':
+    unittest.main()