From 339390820892e8de61b34da74036f43f6b3eff01 Mon Sep 17 00:00:00 2001 From: Andoni Alonso <14891798+andoniaf@users.noreply.github.com> Date: Wed, 28 May 2025 15:54:33 +0200 Subject: [PATCH] feat(IaC): handling output for IaC Provider POC (#7861) Co-authored-by: MrCloudSec --- .pre-commit-config.yaml | 2 +- prowler/lib/check/models.py | 25 +-- prowler/lib/outputs/compliance/cis/models.py | 12 +- prowler/lib/outputs/finding.py | 12 +- prowler/lib/outputs/html/html.py | 49 +++++- prowler/lib/outputs/ocsf/ocsf.py | 4 +- prowler/providers/azure/models.py | 4 +- prowler/providers/common/provider.py | 1 - prowler/providers/gcp/models.py | 2 +- prowler/providers/iac/iac_provider.py | 152 +++++++++++------- .../kubernetes/services/core/core_service.py | 2 +- 11 files changed, 170 insertions(+), 95 deletions(-) diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 0382dd901d..5f13fda438 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -115,7 +115,7 @@ repos: - id: safety name: safety description: "Safety is a tool that checks your installed dependencies for known security vulnerabilities" - entry: bash -c 'safety check --ignore 70612,66963,74429' + entry: bash -c 'safety check --ignore 70612,66963,74429,76352,76353' language: system - id: vulture diff --git a/prowler/lib/check/models.py b/prowler/lib/check/models.py index d0dab6ca59..e35dc0eb64 100644 --- a/prowler/lib/check/models.py +++ b/prowler/lib/check/models.py @@ -612,16 +612,9 @@ class CheckReportM365(Check_Report): class CheckReportIAC(Check_Report): """Contains the IAC Check's finding information using Checkov.""" - metadata: dict - check_id: str - check_name: str - check_result: dict - check_result_status: str - file_path: str - file_line_range: list - guideline: str - resource: str - severity: str + resource_name: str + resource_path: str + resource_line_range: str def __init__(self, metadata: dict = {}, finding: dict = {}) -> None: """ @@ -633,15 +626,9 @@ class CheckReportIAC(Check_Report): """ super().__init__(metadata, finding) - self.check_id = finding.get("check_id", "") - self.check_name = finding.get("check_name", "") - self.check_result = finding.get("check_result", {}) - self.check_result_status = self.check_result.get("result", "UNKNOWN") - self.file_path = finding.get("file_path", "") - self.file_line_range = finding.get("file_line_range", []) - self.guideline = finding.get("guideline", "") - self.resource = finding - self.severity = finding.get("severity", "UNKNOWN") + self.resource_name = getattr(finding, "resource", "") + self.resource_path = getattr(finding, "file_path", "") + self.resource_line_range = getattr(finding, "file_line_range", "") @dataclass diff --git a/prowler/lib/outputs/compliance/cis/models.py b/prowler/lib/outputs/compliance/cis/models.py index 1a4764c294..433142e0d9 100644 --- a/prowler/lib/outputs/compliance/cis/models.py +++ b/prowler/lib/outputs/compliance/cis/models.py @@ -16,7 +16,7 @@ class AWSCISModel(BaseModel): Requirements_Id: str Requirements_Description: str Requirements_Attributes_Section: str - Requirements_Attributes_SubSection: Optional[str] + Requirements_Attributes_SubSection: Optional[str] = None Requirements_Attributes_Profile: str Requirements_Attributes_AssessmentStatus: str Requirements_Attributes_Description: str @@ -50,7 +50,7 @@ class AzureCISModel(BaseModel): Requirements_Id: str Requirements_Description: str Requirements_Attributes_Section: str - Requirements_Attributes_SubSection: Optional[str] + Requirements_Attributes_SubSection: Optional[str] = None Requirements_Attributes_Profile: str Requirements_Attributes_AssessmentStatus: str Requirements_Attributes_Description: str @@ -82,7 +82,7 @@ class M365CISModel(BaseModel): Requirements_Id: str Requirements_Description: str Requirements_Attributes_Section: str - Requirements_Attributes_SubSection: Optional[str] + Requirements_Attributes_SubSection: Optional[str] = None Requirements_Attributes_Profile: str Requirements_Attributes_AssessmentStatus: str Requirements_Attributes_Description: str @@ -114,7 +114,7 @@ class GCPCISModel(BaseModel): Requirements_Id: str Requirements_Description: str Requirements_Attributes_Section: str - Requirements_Attributes_SubSection: Optional[str] + Requirements_Attributes_SubSection: Optional[str] = None Requirements_Attributes_Profile: str Requirements_Attributes_AssessmentStatus: str Requirements_Attributes_Description: str @@ -145,8 +145,8 @@ class KubernetesCISModel(BaseModel): Requirements_Id: str Requirements_Description: str Requirements_Attributes_Section: str - Requirements_Attributes_SubSection: Optional[str] - Requirements_Attributes_Profile: str + Requirements_Attributes_SubSection: Optional[str] = None + Requirements_Attributes_Profile: Optional[str] = None Requirements_Attributes_AssessmentStatus: str Requirements_Attributes_Description: str Requirements_Attributes_RationaleStatement: str diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index 2df79a954a..347a31ecd0 100644 --- a/prowler/lib/outputs/finding.py +++ b/prowler/lib/outputs/finding.py @@ -283,12 +283,16 @@ class Finding(BaseModel): output_data["region"] = check_output.location elif provider.type == "iac": - output_data["auth_method"] = "iac" + output_data["auth_method"] = "local" # Until we support remote repos output_data["account_uid"] = "iac" output_data["account_name"] = "iac" - output_data["resource_name"] = check_output.check_name - output_data["resource_uid"] = check_output.check_id - output_data["region"] = check_output.file_path + output_data["resource_name"] = check_output.resource["resource"] + output_data["resource_uid"] = check_output.resource["resource"] + output_data["region"] = check_output.resource_path + output_data["resource_line_range"] = check_output.resource_line_range + output_data["framework"] = ( + check_output.check_metadata.ServiceName + ) # TODO: can we get the framework from the check_output? # check_output Unique ID # TODO: move this to a function diff --git a/prowler/lib/outputs/html/html.py b/prowler/lib/outputs/html/html.py index f775a00d3d..6c54501640 100644 --- a/prowler/lib/outputs/html/html.py +++ b/prowler/lib/outputs/html/html.py @@ -41,7 +41,7 @@ class HTML(Output): {finding_status} {finding.metadata.Severity.value} {finding.metadata.ServiceName} - {finding.region.lower()} + {":".join([finding.resource_metadata['file_path'], "-".join(map(str, finding.resource_metadata['file_line_range']))]) if finding.metadata.Provider == "iac" else finding.region.lower()} {finding.metadata.CheckID.replace("_", "_")} {finding.metadata.CheckTitle} {finding.resource_uid.replace("<", "<").replace(">", ">").replace("_", "_")} @@ -204,7 +204,7 @@ class HTML(Output): Status Severity Service Name - Region + {"File" if provider.type == "iac" else "Region"} Check ID Check Title Resource ID @@ -689,6 +689,51 @@ class HTML(Output): ) return "" + @staticmethod + def get_iac_assessment_summary(provider: Provider) -> str: + """ + get_iac_assessment_summary gets the HTML assessment summary for the provider + + Args: + provider (Provider): the provider object + + Returns: + str: the HTML assessment summary + """ + try: + return f""" +
+
+
+ IAC Assessment Summary +
+
    +
  • + IAC path: {provider.scan_path} +
    • +
+
+
+
+
+
+ IAC Credentials +
+
    +
  • + IAC authentication method: local +
    • +
+
+
""" + except Exception as error: + logger.error( + f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}" + ) + return "" + @staticmethod def get_assessment_summary(provider: Provider) -> str: """ diff --git a/prowler/lib/outputs/ocsf/ocsf.py b/prowler/lib/outputs/ocsf/ocsf.py index c84a6acf7c..4c47036ef4 100644 --- a/prowler/lib/outputs/ocsf/ocsf.py +++ b/prowler/lib/outputs/ocsf/ocsf.py @@ -3,9 +3,9 @@ from datetime import datetime from typing import List from py_ocsf_models.events.base_event import SeverityID, StatusID -from py_ocsf_models.events.findings.detection_finding import DetectionFinding from py_ocsf_models.events.findings.detection_finding import ( - TypeID as DetectionFindingTypeID, + DetectionFinding, + DetectionFindingTypeID, ) from py_ocsf_models.events.findings.finding import ActivityID, FindingInformation from py_ocsf_models.objects.account import Account, TypeID diff --git a/prowler/providers/azure/models.py b/prowler/providers/azure/models.py index cf0cd4be9b..1745e67214 100644 --- a/prowler/providers/azure/models.py +++ b/prowler/providers/azure/models.py @@ -1,3 +1,5 @@ +from typing import Optional + from pydantic import BaseModel from prowler.config.config import output_file_timestamp @@ -15,7 +17,7 @@ class AzureIdentityInfo(BaseModel): class AzureRegionConfig(BaseModel): name: str = "" - authority: str = None + authority: Optional[str] = None base_url: str = "" credential_scopes: list = [] diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 44e7b067c3..e7f4a74c2f 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -247,7 +247,6 @@ class Provider(ABC): provider_class( scan_path=arguments.scan_path, config_path=arguments.config_file, - mutelist_path=arguments.mutelist_file, fixer_config=fixer_config, ) diff --git a/prowler/providers/gcp/models.py b/prowler/providers/gcp/models.py index 4abdd6b0ff..a6d925fd2f 100644 --- a/prowler/providers/gcp/models.py +++ b/prowler/providers/gcp/models.py @@ -14,7 +14,7 @@ class GCPOrganization(BaseModel): id: str name: str # TODO: the name needs to be retrieved from another API - display_name: Optional[str] + display_name: Optional[str] = None class GCPProject(BaseModel): diff --git a/prowler/providers/iac/iac_provider.py b/prowler/providers/iac/iac_provider.py index a2e3e502ec..c5d8e1e93c 100644 --- a/prowler/providers/iac/iac_provider.py +++ b/prowler/providers/iac/iac_provider.py @@ -5,6 +5,10 @@ from typing import List from colorama import Fore, Style +from prowler.config.config import ( + default_config_file_path, + load_and_validate_config_file, +) from prowler.lib.check.models import CheckReportIAC from prowler.lib.logger import logger from prowler.lib.utils.utils import print_boxes @@ -22,8 +26,6 @@ class IacProvider(Provider): config_path: str = None, config_content: dict = None, fixer_config: dict = {}, - mutelist_path: str = None, - mutelist_content: dict = None, ): logger.info("Instantiating IAC Provider...") @@ -33,8 +35,18 @@ class IacProvider(Provider): self._session = None self._identity = "prowler" - self._audit_config = config_content if config_content else {} + # Audit Config + if config_content: + self._audit_config = config_content + else: + if not config_path: + config_path = default_config_file_path + self._audit_config = load_and_validate_config_file(self._type, config_path) + + # Fixer Config self._fixer_config = fixer_config + + # Mutelist (not needed for IAC since Checkov has its own mutelist logic) self._mutelist = None self.audit_metadata = Audit_Metadata( @@ -74,6 +86,67 @@ 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 + """ + + metadata_dict = { + "Provider": "iac", + "CheckID": check.get("check_id", ""), + "CheckTitle": check.get("check_name", ""), + "CheckType": ["Infrastructure as Code"], + "ServiceName": finding["check_type"], + "SubServiceName": "", + "ResourceIdTemplate": "", + "Severity": ( + check.get("severity", "low").lower() if check.get("severity") else "low" + ), + "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": "", + } + + # Convert metadata dict to JSON string + metadata = json.dumps(metadata_dict) + + report = CheckReportIAC(metadata=metadata, finding=check) + report.status = status + report.resource_tags = check.get("entity_tags", {}) + report.status_extended = check.get("check_name", "") + if status == "MUTED": + report.muted = True + return report + def run(self) -> List[CheckReportIAC]: return self.run_scan(self.scan_path) @@ -99,65 +172,30 @@ class IacProvider(Provider): reports = [] - # Process failed checks + # If only one framework has findings, the output is a dict, otherwise it's a list of dicts + if isinstance(output, dict): + output = [output] + + # Process all frameworks 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", ""), - } + # 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) - # Convert metadata dict to JSON string - metadata = json.dumps(metadata_dict) - - report = CheckReportIAC(metadata=metadata, finding=failed_check) - report.status = "FAIL" + # Process skipped checks (muted) + skipped_checks = results.get("skipped_checks", []) + for skipped_check in skipped_checks: + report = self._process_check(finding, skipped_check, "MUTED") reports.append(report) return reports diff --git a/prowler/providers/kubernetes/services/core/core_service.py b/prowler/providers/kubernetes/services/core/core_service.py index ec6fb48ca8..98138d0d79 100644 --- a/prowler/providers/kubernetes/services/core/core_service.py +++ b/prowler/providers/kubernetes/services/core/core_service.py @@ -171,7 +171,7 @@ class Pod(BaseModel): host_ip: Optional[str] host_pid: Optional[str] host_ipc: Optional[str] - host_network: Optional[str] + host_network: Optional[bool] security_context: Optional[dict] containers: Optional[dict]