feat(IaC): handling output for IaC Provider POC (#7861)

Co-authored-by: MrCloudSec <hello@mistercloudsec.com>

.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

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

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

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

prowler/lib/outputs/html/html.py

@@ -41,7 +41,7 @@ class HTML(Output):
                             <td>{finding_status}</td>
                             <td>{finding.metadata.Severity.value}</td>
                             <td>{finding.metadata.ServiceName}</td>
-                            <td>{finding.region.lower()}</td>
+                            <td>{":".join([finding.resource_metadata['file_path'], "-".join(map(str, finding.resource_metadata['file_line_range']))]) if finding.metadata.Provider == "iac" else finding.region.lower()}</td>
                             <td>{finding.metadata.CheckID.replace("_", "<wbr />_")}</td>
                             <td>{finding.metadata.CheckTitle}</td>
                             <td>{finding.resource_uid.replace("<", "&lt;").replace(">", "&gt;").replace("_", "<wbr />_")}</td>
@@ -204,7 +204,7 @@ class HTML(Output):
                     <th scope="col">Status</th>
                     <th scope="col">Severity</th>
                     <th scope="col">Service Name</th>
-                    <th scope="col">Region</th>
+                    <th scope="col">{"File" if provider.type == "iac" else "Region"}</th>
                     <th style="width:20%" scope="col">Check ID</th>
                     <th style="width:20%" scope="col">Check Title</th>
                     <th scope="col">Resource ID</th>
@@ -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"""
+                <div class="col-md-2">
+                    <div class="card">
+                        <div class="card-header">
+                            IAC Assessment Summary
+                        </div>
+                        <ul class="list-group
+                        list-group-flush">
+                            <li class="list-group-item">
+                                <b>IAC path:</b> {provider.scan_path}
+                            </li>
+                        </ul>
+                    </div>
+                </div>
+                <div class="col-md-4">
+                    <div class="card">
+                        <div class="card-header">
+                            IAC Credentials
+                        </div>
+                        <ul class="list-group
+                        list-group-flush">
+                            <li class="list-group-item">
+                                <b>IAC authentication method:</b> local
+                            </li>
+                        </ul>
+                    </div>
+                </div>"""
+        except Exception as error:
+            logger.error(
+                f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}] -- {error}"
+            )
+            return ""
+
     @staticmethod
     def get_assessment_summary(provider: Provider) -> str:
         """

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

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 = []
 

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,
                     )
 

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):

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

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]