Revert "platform(general): Get resources from platform and filter tag…

…gable resources for policies (#3621)"

This reverts commit 1cd0ed3.

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 NXGraphCheckParser
-from checkov.common.checks_infra.registry import get_graph_checks_registry
+from checkov.common.checks_infra.registry import Registry, get_graph_checks_registry
 
 if TYPE_CHECKING:
     from checkov.common.bridgecrew.platform_integration import BcPlatformIntegration
@@ -24,8 +24,9 @@
 
 class CustomPoliciesIntegration(BaseIntegrationFeature):
     def __init__(self, bc_integration: BcPlatformIntegration) -> None:
-        super().__init__(bc_integration=bc_integration, order=2)  # must be after policy metadata and before suppression integration
+        super().__init__(bc_integration=bc_integration, order=1)  # must be after policy metadata and before suppression integration
         self.platform_policy_parser = NXGraphCheckParser()
+        self.policies_url = f"{self.bc_integration.api_url}/api/v1/policies/table/data"
         self.bc_cloned_checks: dict[str, list[dict[str, Any]]] = defaultdict(list)
 
     def is_valid(self) -> bool:
@@ -52,7 +53,8 @@ def pre_scan(self) -> None:
                         policy['severity'] = Severities[policy['severity']]
                         self.bc_cloned_checks[source_incident_id].append(policy)
                         continue
-                    check = self.platform_policy_parser.parse_raw_check(converted_check)
+                    resource_types = Registry._get_resource_types(converted_check['metadata'])
+                    check = self.platform_policy_parser.parse_raw_check(converted_check, resources_types=resource_types)
                     check.severity = Severities[policy['severity']]
                     check.bc_id = check.id
                     if check.frameworks:
@@ -82,13 +84,6 @@ def _convert_raw_check(policy: dict[str, Any]) -> dict[str, Any]:
             'category': policy['category'],
             'frameworks': policy.get('frameworks', [])
         }
-
-        provider = policy.get('provider')
-        if provider:
-            metadata['scope'] = {
-                'provider': provider.lower()
-            }
-
         check = {
             'metadata': metadata,
             'definition': json.loads(policy['code'])

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

@@ -16,7 +16,7 @@
 
 class RepoConfigIntegration(BaseIntegrationFeature):
     def __init__(self, bc_integration: BcPlatformIntegration) -> None:
-        super().__init__(bc_integration=bc_integration, order=1)
+        super().__init__(bc_integration=bc_integration, order=0)
         self.skip_paths: set[str] = set()
         self.enforcement_rule: dict[str, Any] = {}
         self.code_category_configs: dict[str, CodeCategoryConfiguration] = {}

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

@@ -22,7 +22,7 @@
 
 class SuppressionsIntegration(BaseIntegrationFeature):
     def __init__(self, bc_integration: BcPlatformIntegration) -> None:
-        super().__init__(bc_integration=bc_integration, order=4)  # must be after the custom policies integration
+        super().__init__(bc_integration=bc_integration, order=2)  # must be after the custom policies integration
         self.suppressions: dict[str, list[dict[str, Any]]] = {}
         self.suppressions_url = f"{self.bc_integration.api_url}/api/v1/suppressions"
 

checkov/common/bridgecrew/platform_integration.py

@@ -524,7 +524,7 @@ def get_platform_run_config(self) -> None:
             self.get_public_run_config()
 
     def get_run_config_url(self) -> str:
-        return f'{self.platform_run_config_url}?module={"bc" if self.is_bc_token(self.bc_api_key) else "pc"}&includeResources=true'
+        return f'{self.platform_run_config_url}?module={"bc" if self.is_bc_token(self.bc_api_key) else "pc"}'
 
     def get_customer_run_config(self) -> None:
         if self.skip_download is True:

checkov/common/checks_infra/checks_parser.py

@@ -46,7 +46,6 @@
     EqualsIgnoreCaseAttributeSolver,
     NotEqualsIgnoreCaseAttributeSolver
 )
-from checkov.common.bridgecrew.integration_features.features.attribute_resource_types_integration import integration as attribute_resource_type_integration
 from checkov.common.checks_infra.solvers.connections_solvers.connection_one_exists_solver import \
     ConnectionOneExistsSolver
 from checkov.common.graph.checks_infra.base_check import BaseGraphCheck
@@ -134,33 +133,18 @@
 class NXGraphCheckParser(BaseGraphCheckParser):
     def parse_raw_check(self, raw_check: Dict[str, Dict[str, Any]], **kwargs: Any) -> BaseGraphCheck:
         policy_definition = raw_check.get("definition", {})
-
-        metadata = raw_check.get("metadata", {})
-
-        # the first approach comes from the custom policy integration
-        provider = metadata.get("scope", {}).get("provider")
-
-        # but the platform injects check metadata in a different way
-        if not provider and "scope" in raw_check:
-            raw_provider = raw_check["scope"].get("provider")  # will be a None, an empty list, or a list with the provider
-            if raw_provider:
-                provider = raw_provider[0].lower()
-
-        check = self._parse_raw_check(policy_definition, provider)
-
-        check.id = metadata.get("id", "")
-        check.name = metadata.get("name", "")
-        check.category = metadata.get("category", "")
-        check.frameworks = metadata.get("frameworks", [])
-        check.guideline = metadata.get("guideline")
-        check.provider = provider
-
+        check = self._parse_raw_check(policy_definition, kwargs.get("resources_types"))
+        check.id = raw_check.get("metadata", {}).get("id", "")
+        check.name = raw_check.get("metadata", {}).get("name", "")
+        check.category = raw_check.get("metadata", {}).get("category", "")
+        check.frameworks = raw_check.get("metadata", {}).get("frameworks", [])
+        check.guideline = raw_check.get("metadata", {}).get("guideline")
         solver = self.get_check_solver(check)
         check.set_solver(solver)
 
         return check
 
-    def _parse_raw_check(self, raw_check: Dict[str, Any], provider: Optional[str]) -> BaseGraphCheck:
+    def _parse_raw_check(self, raw_check: Dict[str, Any], resources_types: Optional[List[str]]) -> BaseGraphCheck:
         check = BaseGraphCheck()
         complex_operator = get_complex_operator(raw_check)
         if complex_operator:
@@ -174,11 +158,7 @@ def _parse_raw_check(self, raw_check: Dict[str, Any], provider: Optional[str]) -
                 sub_solvers = [sub_solvers]
 
             for sub_solver in sub_solvers:
-                check.sub_checks.append(self._parse_raw_check(sub_solver, provider))
-
-            # conditions with enumerated resource types will have them as a list. conditions where `all` is replaced with the
-            # actual list of resource for the attribute (e.g. tags) will have them as a set, because that logic works best with sets
-            # here, they will end up as a list in the policy resource types
+                check.sub_checks.append(self._parse_raw_check(sub_solver, resources_types))
             resources_types_of_sub_solvers = [
                 force_list(q.resource_types) for q in check.sub_checks if q is not None and q.resource_types is not None
             ]
@@ -193,17 +173,13 @@ def _parse_raw_check(self, raw_check: Dict[str, Any], provider: Optional[str]) -
                     or (isinstance(resource_type, str) and resource_type.lower() == "all")
                     or (isinstance(resource_type, list) and resource_type[0].lower() == "all")
             ):
-                resource_types_for_attribute = attribute_resource_type_integration.get_attribute_resource_types(raw_check, provider)
-                check.resource_types = resource_types_for_attribute or []
+                check.resource_types = resources_types or []
             else:
                 check.resource_types = resource_type
 
             connected_resources_type = raw_check.get("connected_resource_types", [])
-
-            # TODO this code has a capital 'All', so I am pretty sure this rarely gets used. need to validate the use case
-            # and make it work with the resource types from the platform if needed
             if connected_resources_type == ["All"] or connected_resources_type == "all":
-                check.connected_resources_types = []
+                check.connected_resources_types = resources_types or []
             else:
                 check.connected_resources_types = connected_resources_type
 

checkov/common/checks_infra/registry.py

@@ -4,14 +4,15 @@
 import logging
 import os
 from pathlib import Path
-from typing import TYPE_CHECKING
+from typing import Any, TYPE_CHECKING
 
 import yaml
 
 from checkov.common.checks_infra.checks_parser import NXGraphCheckParser
 from checkov.common.graph.checks_infra.base_parser import BaseGraphCheckParser
 from checkov.common.graph.checks_infra.registry import BaseRegistry
 from checkov.runner_filter import RunnerFilter
+from checkov.common.checks_infra.resources_types import resources_types
 
 if TYPE_CHECKING:
     from checkov.common.graph.checks_infra.base_check import BaseGraphCheck
@@ -47,7 +48,9 @@ def _load_checks_from_dir(self, directory: str, external_check: bool) -> None:
                         if not isinstance(check_json, dict):
                             self.logger.error(f"Loaded data from JSON is not Dict. Skipping. Data: {check_json}.")
                             continue
-                        check = self.parser.parse_raw_check(check_json)
+                        check = self.parser.parse_raw_check(
+                            check_json, resources_types=self._get_resource_types(check_json)
+                        )
                         if not any(c for c in self.checks if check.id == c.id):
                             if external_check:
                                 # Note the external check; used in the should_run_check logic
@@ -57,6 +60,11 @@ def _load_checks_from_dir(self, directory: str, external_check: bool) -> None:
     def load_external_checks(self, dir: str) -> None:
         self._load_checks_from_dir(dir, True)
 
+    @staticmethod
+    def _get_resource_types(check_json: dict[str, dict[str, Any]]) -> list[str] | None:
+        provider = check_json.get("scope", {}).get("provider", "").lower()
+        return resources_types.get(provider)
+
 
 _registry_instances: dict[str, Registry] = {}