ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 12:31:54 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(iac): process passed_checks

Browse Source
This commit is contained in:
Andoni A.
2025-05-28 09:15:00 +02:00
parent cd0960babd
commit b94d032dbb
1 changed files with 72 additions and 55 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/providers/iac/iac_provider.py
+72 -55
View File
@@ -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