From b94d032dbbd5ce04984aebdc50ab9b04643c4ad7 Mon Sep 17 00:00:00 2001 From: "Andoni A." <14891798+andoniaf@users.noreply.github.com> Date: Wed, 28 May 2025 09:15:00 +0200 Subject: [PATCH] feat(iac): process passed_checks --- prowler/providers/iac/iac_provider.py | 127 +++++++++++++++----------- 1 file changed, 72 insertions(+), 55 deletions(-) diff --git a/prowler/providers/iac/iac_provider.py b/prowler/providers/iac/iac_provider.py index a2e3e502ec..a8f3d74c80 100644 --- a/prowler/providers/iac/iac_provider.py +++ b/prowler/providers/iac/iac_provider.py @@ -74,6 +74,68 @@ class IacProvider(Provider): """IAC provider doesn't need a session since it uses Checkov directly""" return None + def _process_check(self, finding: dict, check: dict, status: str) -> CheckReportIAC: + """ + Process a single check (failed or passed) and create a CheckReportIAC object. + + Args: + finding: The finding object from Checkov output + check: The individual check data (failed_check or passed_check) + status: The status of the check ("FAIL" or "PASS") + + Returns: + CheckReportIAC: The processed check report + """ + # Get severity and ensure it's a valid string + severity = finding.get("severity", "low") + if severity is None: + severity = "low" + severity = str(severity).lower() + + # Create a basic metadata structure for the check + metadata_dict = { + "Provider": "iac", + "CheckID": check.get("check_id", ""), + "CheckTitle": check.get("check_name", ""), + "CheckType": ["Infrastructure as Code"], + "ServiceName": "iac", + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": severity, + "ResourceType": "iac", + "Description": check.get("check_name", ""), + "Risk": "", + "RelatedUrl": ( + check.get("guideline", "") if check.get("guideline") else "" + ), + "Remediation": { + "Code": { + "NativeIaC": "", + "Terraform": "", + "CLI": "", + "Other": "", + }, + "Recommendation": { + "Text": "", + "Url": ( + check.get("guideline", "") if check.get("guideline") else "" + ), + }, + }, + "Categories": [], + "DependsOn": [], + "RelatedTo": [], + "Notes": "", + "GCPProject": check.get("project_id", ""), + } + + # Convert metadata dict to JSON string + metadata = json.dumps(metadata_dict) + + report = CheckReportIAC(metadata=metadata, finding=check) + report.status = status + return report + def run(self) -> List[CheckReportIAC]: return self.run_scan(self.scan_path) @@ -99,65 +161,20 @@ class IacProvider(Provider): reports = [] - # Process failed checks + # Process all findings for finding in output: - failed_checks = finding.get("results", {}).get("failed_checks", []) - if not failed_checks: - continue + results = finding.get("results", {}) + # Process failed checks + failed_checks = results.get("failed_checks", []) for failed_check in failed_checks: - # Get severity and ensure it's a valid string - severity = finding.get("severity", "low") - if severity is None: - severity = "low" - severity = str(severity).lower() + report = self._process_check(finding, failed_check, "FAIL") + reports.append(report) - # Create a basic metadata structure for the check - metadata_dict = { - "Provider": "iac", - "CheckID": failed_check.get("check_id", ""), - "CheckTitle": failed_check.get("check_name", ""), - "CheckType": ["Infrastructure as Code"], - "ServiceName": "iac", - "SubServiceName": "", - "ResourceIdTemplate": "", - "Severity": severity, - "ResourceType": "iac", - "Description": failed_check.get("check_name", ""), - "Risk": "", - "RelatedUrl": ( - failed_check.get("guideline", "") - if failed_check.get("guideline") - else "" - ), - "Remediation": { - "Code": { - "NativeIaC": "", - "Terraform": "", - "CLI": "", - "Other": "", - }, - "Recommendation": { - "Text": "", - "Url": ( - failed_check.get("guideline", "") - if failed_check.get("guideline") - else "" - ), - }, - }, - "Categories": [], - "DependsOn": [], - "RelatedTo": [], - "Notes": "", - "GCPProject": failed_check.get("project_id", ""), - } - - # Convert metadata dict to JSON string - metadata = json.dumps(metadata_dict) - - report = CheckReportIAC(metadata=metadata, finding=failed_check) - report.status = "FAIL" + # Process passed checks + passed_checks = results.get("passed_checks", []) + for passed_check in passed_checks: + report = self._process_check(finding, passed_check, "PASS") reports.append(report) return reports