support muliple frameworks in one custom policy

checkov/common/bridgecrew/integration_features/features/custom_policies_integration.py

@@ -11,7 +11,7 @@
 from checkov.common.bridgecrew.platform_integration import bc_integration
 from checkov.common.bridgecrew.severities import Severities
 from checkov.common.checks_infra.checks_parser import GraphCheckParser
-from checkov.common.checks_infra.registry import Registry, get_graph_checks_registry
+from checkov.common.checks_infra.registry import Registry, get_graph_checks_registry, get_all_graph_checks_registries
 from checkov.common.util.data_structures_utils import pickle_deepcopy
 
 if TYPE_CHECKING:
@@ -86,10 +86,9 @@ def pre_scan(self) -> None:
                                 get_graph_checks_registry("kubernetes").checks.append(check)
                             elif f.lower() == "bicep":
                                 get_graph_checks_registry("bicep").checks.append(check)
-                    elif re.match(CFN_RESOURCE_TYPE_IDENTIFIER, check.resource_types[0]):
-                        get_graph_checks_registry("cloudformation").checks.append(check)
                     else:
-                        get_graph_checks_registry("terraform").checks.append(check)
+                        for registry in get_all_graph_checks_registries():
+                            registry.checks.append(check)
                 except Exception:
                     logging.debug(f"Failed to load policy id: {policy.get('id')}", exc_info=True)
             logging.debug(f'Found {len(policies)} custom policies from the platform.')

checkov/common/checks_infra/registry.py

@@ -5,6 +5,7 @@
 import os
 from pathlib import Path
 from typing import Any, TYPE_CHECKING
+from enum import Enum
 
 import yaml
 
@@ -21,6 +22,34 @@
 CHECKS_POSSIBLE_ENDING = {".json", ".yaml", ".yml"}
 
 
+class IACFrameworkTypes(str, Enum):
+    TERRAFORM = "terraform"
+    CLOUDFORMATION = "cloudformation"
+    SERVERLESS = "serverless"
+    KUBERNETES = "kubernetes"
+    SECRETS = "secrets"
+    TERRAFORM_PLAN = "terraform_plan"
+    ARM = "arm"
+    DOCKERFILE = "dockerfile"
+    DOCKER_IMAGE = "docker_image"
+    GITHUB_CONFIGURATION = "github_configuration"
+    GITLAB_CONFIGURATION = "gitlab_configuration"
+    KUSTOMIZE = "kustomize"
+    BICEP = "bicep"
+    OPENAPI = "openapi"
+    GITHUB_ACTION = "github_actions"
+    HELM = "helm"
+    GITLAB_CI = "gitlab_ci"
+    AZURE_PIPELINES = "azure_pipelines"
+    CIRCLECI_PIPELINES = "circleci_pipelines"
+    ANSIBLE = "ansible"
+
+
+IACFrameworksWithRunner = [IACFrameworkTypes.TERRAFORM, IACFrameworkTypes.CLOUDFORMATION, IACFrameworkTypes.KUBERNETES,
+                           IACFrameworkTypes.TERRAFORM_PLAN, IACFrameworkTypes.KUSTOMIZE, IACFrameworkTypes.BICEP,
+                           IACFrameworkTypes.GITHUB_ACTION, IACFrameworkTypes.HELM, IACFrameworkTypes.ANSIBLE]
+
+
 class Registry(BaseRegistry):
     def __init__(self, checks_dir: str, parser: BaseGraphCheckParser | None = None) -> None:
         parser = parser or BaseGraphCheckParser()
@@ -65,7 +94,8 @@ def _load_checks_from_dir(self, directory: str, external_check: bool) -> None:
                             continue
 
                         check = self.parser.parse_raw_check(
-                            check_json, resources_types=self._get_resource_types(check_json), check_path=f'{root}/{file}'
+                            check_json, resources_types=self._get_resource_types(check_json),
+                            check_path=f'{root}/{file}'
                         )
                         if not any(c for c in self.checks if check.id == c.id):
                             if external_check:
@@ -85,10 +115,21 @@ def _get_resource_types(check_json: dict[str, dict[str, Any]]) -> list[str] | No
 _registry_instances: dict[str, Registry] = {}
 
 
+def _initialize_registry(check_type: str) -> None:
+    _registry_instances[check_type] = Registry(
+        parser=GraphCheckParser(),
+        checks_dir=f"{Path(__file__).parent.parent.parent}/{check_type}/checks/graph_checks",
+    )
+
+
 def get_graph_checks_registry(check_type: str) -> Registry:
     if not _registry_instances.get(check_type):
-        _registry_instances[check_type] = Registry(
-            parser=GraphCheckParser(),
-            checks_dir=f"{Path(__file__).parent.parent.parent}/{check_type}/checks/graph_checks",
-        )
+        _initialize_registry(check_type)
     return _registry_instances[check_type]
+
+
+def get_all_graph_checks_registries() -> list[Registry]:
+    for framework in IACFrameworksWithRunner:
+        if not _registry_instances.get(framework):
+            _initialize_registry(framework)
+    return list(_registry_instances.values())

tests/common/integration_features/test_custom_policies_integration.py

@@ -7,7 +7,7 @@
     CustomPoliciesIntegration
 from checkov.common.bridgecrew.platform_integration import BcPlatformIntegration
 from checkov.common.checks_infra.checks_parser import GraphCheckParser
-from checkov.common.checks_infra.registry import Registry, get_graph_checks_registry
+from checkov.common.checks_infra.registry import Registry, get_all_graph_checks_registries, get_graph_checks_registry
 from checkov.common.models.enums import CheckResult
 from checkov.common.output.record import Record
 from checkov.common.output.report import Report
@@ -228,6 +228,25 @@ def test_pre_scan_with_cloned_checks(self):
         self.assertEqual('kpande_bicep_1650378013212', bicep_registry[0].id)
         self.assertEqual('kpande_bicep_1650378013212', bicep_registry[0].bc_id)
 
+    def test_pre_scan_with_multiple_frameworks_graph_check(self):
+        instance = BcPlatformIntegration()
+        instance.skip_download = False
+        instance.platform_integration_configured = True
+        custom_policies_integration = CustomPoliciesIntegration(instance)
+
+        instance.customer_run_config_response = mock_multiple_frameworks_custom_policy_response()
+
+        custom_policies_integration.pre_scan()
+        bicep_registry = get_graph_checks_registry("bicep").checks
+        all_graph_checks = get_all_graph_checks_registries()
+        for registry in all_graph_checks:
+            multiple_frameworks_custom_policy_exist = False
+            for check in registry.checks:
+                if check.bc_id == 'multiple_frameworks_policy_1625063607541':
+                    multiple_frameworks_custom_policy_exist = True
+            self.assertEqual(True, multiple_frameworks_custom_policy_exist)
+        self.assertEqual(2, len(bicep_registry))
+
     def test_post_runner_with_cloned_checks(self):
         instance = BcPlatformIntegration()
         instance.skip_download = False
@@ -474,8 +493,8 @@ def test_policy_load_with_resources_types_as_str(self):
             Path(__file__).parent.parent.parent.parent / "checkov" / "terraform" / "checks" / "graph_checks"))
         checks = [parser.parse_raw_check(CustomPoliciesIntegration._convert_raw_check(p)) for p in policies]
         registry.checks = checks  # simulate that the policy downloader will do
-        
-        
+
+
 def mock_custom_policies_response():
     return {
         "customPolicies": [
@@ -556,5 +575,45 @@ def mock_custom_policies_response():
     }
 
 
+def mock_multiple_frameworks_custom_policy_response():
+    return {
+        "customPolicies": [
+            {
+                "id": "kpande_bicep_1650378013212",
+                "code": "{\"operator\":\"exists\",\"attribute\":\"spec.runAsUser.rule\",\"cond_type\":\"attribute\","
+                        "\"resource_types\":[\"PodSecurityPolicy\"]}",
+                "title": "bicep policy",
+                "guideline": "meaningful guideline for bicep policy",
+                "severity": "HIGH",
+                "pcSeverity": None,
+                "category": "bicep",
+                "pcPolicyId": None,
+                "additionalPcPolicyIds": None,
+                "sourceIncidentId": None,
+                "benchmarks": {},
+                "frameworks": [
+                    "bicep"
+                ]
+            },
+            {
+                "id": "multiple_frameworks_policy_1625063607541",
+                "title": "multiple frameworks policy",
+                "code": "{\"and\":[{\"operator\":\"exists\",\"cond_type\":\"connection\",\"resource_types\":["
+                        "\"azurerm_subnet_network_security_group_association\"],\"connected_resource_types\":["
+                        "\"azurerm_subnet\",\"azurerm_network_security_group\"]},{\"value\":[\"azurerm_subnet\"],"
+                        "\"operator\":\"within\",\"attribute\":\"resource_type\",\"cond_type\":\"filter\"}]}",
+                "severity": "CRITICAL",
+                "category": "General",
+                "frameworks": [],
+                "resourceTypes": ["aws_s3_bucket", "PodSecurityPolicy"],
+                "guideline": "multiple_frameworks_policy_1625063607541",
+                "benchmarks": {},
+                "createdBy": "[email protected]",
+                "sourceIncidentId": None
+            }
+        ]
+    }
+
+
 if __name__ == '__main__':
     unittest.main()