Merge branch 'fix/issue-6686-unrecognised-sg-in-docdbelastic' of gith…

…ub.com:bhean/checkov into fix/issue-6686-unrecognised-sg-in-docdbelastic
bridgecrewio · Aug 27, 2024 · a0a1b20 · a0a1b20
2 parents 4faa1b8 + 9de662c
commit a0a1b20
Show file tree

Hide file tree

Showing 19 changed files with 106 additions and 13 deletions.
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -1,6 +1,10 @@
 # CHANGELOG
 
-## [Unreleased](https://github.com/bridgecrewio/checkov/compare/3.2.235...HEAD)
+## [Unreleased](https://github.com/bridgecrewio/checkov/compare/3.2.236...HEAD)
+
+## [3.2.236](https://github.com/bridgecrewio/checkov/compare/3.2.235...3.2.236) - 2024-08-26
+
+- no noteworthy changes
 
 ## [3.2.235](https://github.com/bridgecrewio/checkov/compare/3.2.234...3.2.235) - 2024-08-21
 

diff --git a/checkov/secrets/runner.py b/checkov/secrets/runner.py
@@ -152,7 +152,7 @@ def run(
                                           for suppression in suppressions if suppression['suppressionType'] == 'SecretsPolicy']
             if policies_list:
                 runnable_plugins: dict[str, str] = get_runnable_plugins(policies_list)
-                logging.info(f"Found {len(runnable_plugins)} runnable plugins")
+                logging.debug(f"Found {len(runnable_plugins)} runnable plugins")
                 if len(runnable_plugins) > 0:
                     plugins_index += 1
                 for name, runnable_plugin in runnable_plugins.items():
@@ -164,7 +164,7 @@ def run(
                         'path': f'file://{work_path}/runnable_plugin_{plugins_index}.py'
                     })
                     plugins_index += 1
-                    logging.info(f"Loaded runnable plugin {name}")
+                    logging.debug(f"Loaded runnable plugin {name}")
         # load internal regex detectors
         detector_path = f"{current_dir}/plugins/custom_regex_detector.py"
         logging.info(f"Custom detector found at {detector_path}. Loading...")
@@ -390,7 +390,7 @@ def _safe_scan(file_path: str, base_path: str) -> tuple[str, list[PotentialSecre
         try:
             start_time = datetime.datetime.now()
             file_results = [*scan.scan_file(full_file_path)]
-            logging.info(f'file {full_file_path} results len {len(file_results)}')
+            logging.debug(f'file {full_file_path} results len {len(file_results)}')
             end_time = datetime.datetime.now()
             run_time = end_time - start_time
             if run_time > datetime.timedelta(seconds=10):

diff --git a/checkov/terraform/checks/data/aws/IAMPrivilegeEscalation.py b/checkov/terraform/checks/data/aws/IAMPrivilegeEscalation.py
@@ -15,7 +15,15 @@ def __init__(self) -> None:
         super().__init__(name=name, id=id)
 
     def cloudsplaining_analysis(self, policy: PolicyDocument) -> Union[List[str], List[Dict[str, Any]]]:
-        return policy.allows_privilege_escalation
+        escalations = policy.allows_privilege_escalation
+        flattened_escalations: list[str] = []
+        if escalations:
+            for escalation in escalations:
+                if isinstance(escalation, dict):
+                    flattened_escalations.extend(escalation.get('actions'))
+                else:
+                    flattened_escalations.append(escalation)
+        return flattened_escalations
 
 
 check = CloudSplainingPrivilegeEscalation()
diff --git a/checkov/terraform/checks/data/base_cloudsplaining_data_iam_check.py b/checkov/terraform/checks/data/base_cloudsplaining_data_iam_check.py
@@ -1,5 +1,7 @@
+import fnmatch
+import logging
 from abc import ABC
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Union
 
 from cloudsplaining.scan.policy_document import PolicyDocument
 
@@ -28,3 +30,24 @@ def should_scan_conf(self, conf: Dict[str, List[Any]]) -> bool:
     def convert_to_iam_policy(self, conf: Dict[str, List[Any]]) -> PolicyDocument:
         converted_conf = convert_terraform_conf_to_iam_policy(conf)
         return PolicyDocument(converted_conf)
+
+    def cloudsplaining_enrich_evaluated_keys(self, policy: PolicyDocument,
+                                             violating_actions: Union[List[str], List[Dict[str, Any]]]) -> None:
+        try:
+            # in case we have violating actions for this policy we start looking for it through the statements
+            for stmt_idx, statement in enumerate(policy.statements):
+                actions = statement.statement.get('Action')  # get the actions for this statement
+                if actions:
+                    if isinstance(actions, str):
+                        for violating_action in violating_actions:
+                            if fnmatch.fnmatch(violating_action.lower(), actions.lower()):  # found the violating action in our list of actions
+                                self.evaluated_keys.append(f"statement/[{stmt_idx}]/actions")
+                                return
+                    if isinstance(actions, list):
+                        for action in actions:      # go through the actions of this statement and try to match one violation
+                            for violating_action in violating_actions:
+                                if isinstance(action, str) and fnmatch.fnmatch(violating_action.lower(), action.lower()):      # found the violating action in our list of actions
+                                    self.evaluated_keys.append(f"statement/[{stmt_idx}]/actions")
+                                    return
+        except Exception as e:
+            logging.warning(f'Failed enriching cloudsplaining evaluated keys due to: {e}')
diff --git a/checkov/terraform/checks/resource/aws/IAMPrivilegeEscalation.py b/checkov/terraform/checks/resource/aws/IAMPrivilegeEscalation.py
@@ -16,7 +16,15 @@ def __init__(self) -> None:
         super().__init__(name=name, id=id)
 
     def cloudsplaining_analysis(self, policy: PolicyDocument) -> Union[List[str], List[Dict[str, Any]]]:
-        return policy.allows_privilege_escalation
+        escalations = policy.allows_privilege_escalation
+        flattened_escalations: list[str] = []
+        if escalations:
+            for escalation in escalations:
+                if isinstance(escalation, dict):
+                    flattened_escalations.extend(escalation.get('actions'))
+                else:
+                    flattened_escalations.append(escalation)
+        return flattened_escalations
 
 
 check = ResourceCloudSplainingPrivilegeEscalation()
diff --git a/checkov/terraform/checks/resource/base_cloudsplaining_resource_iam_check.py b/checkov/terraform/checks/resource/base_cloudsplaining_resource_iam_check.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import fnmatch
+import logging
 from abc import ABC
-from typing import Dict, List, Any
+from typing import Dict, List, Any, Union
 
 from cloudsplaining.scan.policy_document import PolicyDocument
 
@@ -41,3 +43,24 @@ def convert_to_iam_policy(self, conf: Dict[str, Any]) -> PolicyDocument:
             policy = conf['policy'][0]
 
         return PolicyDocument(policy)
+
+    def cloudsplaining_enrich_evaluated_keys(self, policy: PolicyDocument,
+                                             violating_actions: Union[List[str], List[Dict[str, Any]]]) -> None:
+        try:
+            # in case we have violating actions for this policy we start looking for it through the statements
+            for stmt_idx, statement in enumerate(policy.statements):
+                actions = statement.statement.get('Action')  # get the actions for this statement
+                if actions:
+                    if isinstance(actions, str):
+                        for violating_action in violating_actions:
+                            if fnmatch.fnmatch(violating_action.lower(), actions.lower()):  # found the violating action in our list of actions
+                                self.evaluated_keys.append(f"policy/Statement/[{stmt_idx}]/Action")
+                                return
+                    if isinstance(actions, list):
+                        for action in actions:  # go through the actions of this statement and try to match one violation
+                            for violating_action in violating_actions:
+                                if isinstance(action, str) and fnmatch.fnmatch(violating_action.lower(), action.lower()):  # found the violating action in our list of actions
+                                    self.evaluated_keys.append(f"policy/Statement/[{stmt_idx}]/Action")
+                                    return
+        except Exception as e:
+            logging.warning(f'Failed enriching cloudsplaining evaluated keys due to: {e}')
diff --git a/checkov/terraform/checks/utils/base_cloudsplaining_iam_scanner.py b/checkov/terraform/checks/utils/base_cloudsplaining_iam_scanner.py
@@ -23,9 +23,11 @@ def scan_conf(self, conf: Dict[str, List[Any]]) -> CheckResult:
                 if self.cache_key not in BaseTerraformCloudsplainingIAMScanner.policy_document_cache.keys():
                     policy = self.convert_to_iam_policy(conf)
                     BaseTerraformCloudsplainingIAMScanner.policy_document_cache[self.cache_key] = policy
-                violations = self.cloudsplaining_analysis(
-                    BaseTerraformCloudsplainingIAMScanner.policy_document_cache[self.cache_key]
-                )
+
+                policy_document: PolicyDocument = BaseTerraformCloudsplainingIAMScanner.policy_document_cache[self.cache_key]
+                violations = self.cloudsplaining_analysis(policy_document)
+                if violations and hasattr(self, 'evaluated_keys'):
+                    self.cloudsplaining_enrich_evaluated_keys(policy_document, violations)
             except Exception:
                 # this might occur with templated iam policies where ARN is not in place or similar
                 logging.debug(f"could not run cloudsplaining analysis on policy {conf}")
@@ -51,3 +53,8 @@ def convert_to_iam_policy(self, conf: Dict[str, List[Any]]) -> PolicyDocument:
     @abstractmethod
     def cloudsplaining_analysis(self, policy: PolicyDocument) -> Union[List[str], List[Dict[str, Any]]]:
         raise NotImplementedError()
+
+    @abstractmethod
+    def cloudsplaining_enrich_evaluated_keys(self, policy: PolicyDocument,
+                                             violating_actions: Union[List[str], List[Dict[str, Any]]]) -> None:
+        raise NotImplementedError()
diff --git a/checkov/version.py b/checkov/version.py
@@ -1 +1 @@
-version = '3.2.236'
+version = '3.2.238'
diff --git a/kubernetes/requirements.txt b/kubernetes/requirements.txt
@@ -1 +1 @@
-checkov==3.2.236
+checkov==3.2.238
diff --git a/tests/terraform/checks/data/aws/test_CloudSplainingCredentialsExposure.py b/tests/terraform/checks/data/aws/test_CloudSplainingCredentialsExposure.py
@@ -35,6 +35,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(['statement/[0]/actions'], report.failed_checks[0].check_result.get('evaluated_keys'))
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/data/aws/test_CloudSplainingDataExfiltration.py b/tests/terraform/checks/data/aws/test_CloudSplainingDataExfiltration.py
@@ -33,6 +33,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(['statement/[0]/actions'], report.failed_checks[0].check_result.get('evaluated_keys'))
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/data/aws/test_CloudSplainingPrivilegeEscalation.py b/tests/terraform/checks/data/aws/test_CloudSplainingPrivilegeEscalation.py
@@ -33,6 +33,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(['statement/[0]/actions'], report.failed_checks[0].check_result.get('evaluated_keys'))
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/data/aws/test_CloudsplainingIAMWrite.py b/tests/terraform/checks/data/aws/test_CloudsplainingIAMWrite.py
@@ -34,6 +34,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(['statement/[0]/actions'], report.failed_checks[0].check_result.get('evaluated_keys'))
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/data/aws/test_CloudsplainingPermissionsManagement.py b/tests/terraform/checks/data/aws/test_CloudsplainingPermissionsManagement.py
@@ -33,6 +33,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(['statement/[0]/actions'], report.failed_checks[0].check_result.get('evaluated_keys'))
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/resource/aws/test_IAMCredentialsExposure.py b/tests/terraform/checks/resource/aws/test_IAMCredentialsExposure.py
@@ -35,6 +35,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(report.failed_checks[0].check_result.get('evaluated_keys'), ['policy/Statement/[0]/Action'])
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/resource/aws/test_IAMDataExfiltration.py b/tests/terraform/checks/resource/aws/test_IAMDataExfiltration.py
@@ -33,6 +33,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(report.failed_checks[0].check_result.get('evaluated_keys'), ['policy/Statement/[0]/Action'])
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/resource/aws/test_IAMPermissionsManagement.py b/tests/terraform/checks/resource/aws/test_IAMPermissionsManagement.py
@@ -33,6 +33,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(report.failed_checks[0].check_result.get('evaluated_keys'), ['policy/Statement/[0]/Action'])
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/resource/aws/test_IAMPrivilegeEscalation.py b/tests/terraform/checks/resource/aws/test_IAMPrivilegeEscalation.py
@@ -32,6 +32,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(report.failed_checks[0].check_result.get('evaluated_keys'), ['policy/Statement/[0]/Action'])
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)
 

diff --git a/tests/terraform/checks/resource/aws/test_IAMWriteAccess.py b/tests/terraform/checks/resource/aws/test_IAMWriteAccess.py
@@ -34,6 +34,8 @@ def test(self):
         self.assertEqual(summary["skipped"], 0)
         self.assertEqual(summary["parsing_errors"], 0)
 
+        self.assertEqual(report.failed_checks[0].check_result.get('evaluated_keys'), ['policy/Statement/[0]/Action'])
+
         self.assertEqual(passing_resources, passed_check_resources)
         self.assertEqual(failing_resources, failed_check_resources)