feat(IaC): POC for IaC Provider

iac/main.tf

@@ -0,0 +1,4 @@
+resource "aws_s3_bucket" "bad_example" {
+  bucket = "my-unsafe-bucket"
+  acl    = "public-read"
+}

prowler/__main__.py

@@ -98,6 +98,7 @@ from prowler.providers.common.provider import Provider
 from prowler.providers.common.quick_inventory import run_provider_quick_inventory
 from prowler.providers.gcp.models import GCPOutputOptions
 from prowler.providers.github.models import GithubOutputOptions
+from prowler.providers.iac.models import IACOutputOptions
 from prowler.providers.kubernetes.models import KubernetesOutputOptions
 from prowler.providers.m365.models import M365OutputOptions
 from prowler.providers.nhn.models import NHNOutputOptions
@@ -175,11 +176,13 @@ def prowler():
     # Load compliance frameworks
     logger.debug("Loading compliance frameworks from .json files")
 
-    bulk_compliance_frameworks = Compliance.get_bulk(provider)
-    # Complete checks metadata with the compliance framework specification
-    bulk_checks_metadata = update_checks_metadata_with_compliance(
-        bulk_compliance_frameworks, bulk_checks_metadata
-    )
+    # Skip compliance frameworks for IAC provider
+    if provider != "iac":
+        bulk_compliance_frameworks = Compliance.get_bulk(provider)
+        # Complete checks metadata with the compliance framework specification
+        bulk_checks_metadata = update_checks_metadata_with_compliance(
+            bulk_compliance_frameworks, bulk_checks_metadata
+        )
 
     # Update checks metadata if the --custom-checks-metadata-file is present
     custom_checks_metadata = None
@@ -231,39 +234,45 @@ def prowler():
     if not args.only_logs:
         global_provider.print_credentials()
 
-    # Import custom checks from folder
-    if checks_folder:
-        custom_checks = parse_checks_from_folder(global_provider, checks_folder)
-        # Workaround to be able to execute custom checks alongside all checks if nothing is explicitly set
-        if (
-            not checks_file
-            and not checks
-            and not services
-            and not severities
-            and not compliance_framework
-            and not categories
-        ):
-            checks_to_execute.update(custom_checks)
+    # Skip service and check loading for IAC provider
+    if provider != "iac":
+        # Import custom checks from folder
+        if checks_folder:
+            custom_checks = parse_checks_from_folder(global_provider, checks_folder)
+            # Workaround to be able to execute custom checks alongside all checks if nothing is explicitly set
+            if (
+                not checks_file
+                and not checks
+                and not services
+                and not severities
+                and not compliance_framework
+                and not categories
+            ):
+                checks_to_execute.update(custom_checks)
 
-    # Exclude checks if -e/--excluded-checks
-    if excluded_checks:
-        checks_to_execute = exclude_checks_to_run(checks_to_execute, excluded_checks)
+        # Exclude checks if -e/--excluded-checks
+        if excluded_checks:
+            checks_to_execute = exclude_checks_to_run(
+                checks_to_execute, excluded_checks
+            )
 
-    # Exclude services if --excluded-services
-    if excluded_services:
-        checks_to_execute = exclude_services_to_run(
-            checks_to_execute, excluded_services, provider
+        # Exclude services if --excluded-services
+        if excluded_services:
+            checks_to_execute = exclude_services_to_run(
+                checks_to_execute, excluded_services, provider
+            )
+
+        # Once the provider is set and we have the eventual checks based on the resource identifier,
+        # it is time to check what Prowler's checks are going to be executed
+        checks_from_resources = (
+            global_provider.get_checks_to_execute_by_audit_resources()
         )
+        # Intersect checks from resources with checks to execute so we only run the checks that apply to the resources with the specified ARNs or tags
+        if getattr(args, "resource_arn", None) or getattr(args, "resource_tag", None):
+            checks_to_execute = checks_to_execute.intersection(checks_from_resources)
 
-    # Once the provider is set and we have the eventual checks based on the resource identifier,
-    # it is time to check what Prowler's checks are going to be executed
-    checks_from_resources = global_provider.get_checks_to_execute_by_audit_resources()
-    # Intersect checks from resources with checks to execute so we only run the checks that apply to the resources with the specified ARNs or tags
-    if getattr(args, "resource_arn", None) or getattr(args, "resource_tag", None):
-        checks_to_execute = checks_to_execute.intersection(checks_from_resources)
-
-    # Sort final check list
-    checks_to_execute = sorted(checks_to_execute)
+        # Sort final check list
+        checks_to_execute = sorted(checks_to_execute)
 
     # Setup Output Options
     if provider == "aws":
@@ -294,6 +303,8 @@ def prowler():
         output_options = NHNOutputOptions(
             args, bulk_checks_metadata, global_provider.identity
         )
+    elif provider == "iac":
+        output_options = IACOutputOptions(args, bulk_checks_metadata)
 
     # Run the quick inventory for the provider if available
     if hasattr(args, "quick_inventory") and args.quick_inventory:
@@ -303,7 +314,10 @@ def prowler():
     # Execute checks
     findings = []
 
-    if len(checks_to_execute):
+    if provider == "iac":
+        # For IAC provider, run the scan directly
+        findings = global_provider.run()
+    elif len(checks_to_execute):
         findings = execute_checks(
             checks_to_execute,
             global_provider,

prowler/config/config.py

@@ -30,6 +30,7 @@ class Provider(str, Enum):
     KUBERNETES = "kubernetes"
     M365 = "m365"
     GITHUB = "github"
+    IAC = "iac"
     NHN = "nhn"
 
 

prowler/lib/check/checks_loader.py

@@ -20,6 +20,10 @@ def load_checks_to_execute(
 ) -> set:
     """Generate the list of checks to execute based on the cloud provider and the input arguments given"""
     try:
+        # Bypass check loading for IAC provider since it uses Checkov directly
+        if provider == "iac":
+            return set()
+
         # Local subsets
         checks_to_execute = set()
         check_aliases = {}

prowler/lib/check/compliance_models.py

@@ -80,6 +80,7 @@ class CIS_Requirement_Attribute_AssessmentStatus(str):
 
 # CIS Requirement Attribute
 class CIS_Requirement_Attribute(BaseModel):
+    model_config = {"arbitrary_types_allowed": True}
     """CIS Requirement Attribute"""
 
     Section: str

prowler/lib/check/models.py

@@ -608,6 +608,42 @@ class CheckReportM365(Check_Report):
         self.location = resource_location
 
 
+@dataclass
+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
+
+    def __init__(self, metadata: dict = {}, finding: dict = {}) -> None:
+        """
+        Initialize the IAC Check's finding information from a Checkov failed_check dict.
+
+        Args:
+            metadata (Dict): Optional check metadata (can be None).
+            failed_check (dict): A single failed_check result from Checkov's JSON output.
+        """
+        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.get("resource", "")
+        self.severity = finding.get("severity", "UNKNOWN")
+
+
 @dataclass
 class CheckReportNHN(Check_Report):
     """Contains the NHN Check's finding information."""

prowler/lib/check/utils.py

@@ -14,6 +14,10 @@ def recover_checks_from_provider(
     Returns a list of tuples with the following format (check_name, check_path)
     """
     try:
+        # Bypass check loading for IAC provider since it uses Checkov directly
+        if provider == "iac":
+            return []
+
         checks = []
         modules = list_modules(provider, service)
         for module_name in modules:
@@ -59,6 +63,10 @@ def recover_checks_from_service(service_list: list, provider: str) -> set:
     Returns a set of checks from the given services
     """
     try:
+        # Bypass check loading for IAC provider since it uses Checkov directly
+        if provider == "iac":
+            return set()
+
         checks = set()
         service_list = [
             "awslambda" if service == "lambda" else service for service in service_list

prowler/lib/cli/parser.py

@@ -26,16 +26,17 @@ class ProwlerArgumentParser:
         self.parser = argparse.ArgumentParser(
             prog="prowler",
             formatter_class=RawTextHelpFormatter,
-            usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,nhn,dashboard} ...",
+            usage="prowler [-h] [--version] {aws,azure,gcp,kubernetes,m365,nhn,dashboard,iac} ...",
             epilog="""
 Available Cloud Providers:
-  {aws,azure,gcp,kubernetes,m365,nhn}
+  {aws,azure,gcp,kubernetes,m365,nhn,iac}
     aws                 AWS Provider
     azure               Azure Provider
     gcp                 GCP Provider
     kubernetes          Kubernetes Provider
     github              GitHub Provider
     m365                Microsoft 365 Provider
+    iac                 IaC Provider
     nhn                 NHN Provider (Unofficial)
 
 Available components:

prowler/lib/outputs/finding.py

@@ -282,6 +282,14 @@ class Finding(BaseModel):
                 output_data["resource_uid"] = check_output.resource_id
                 output_data["region"] = check_output.location
 
+            elif provider.type == "iac":
+                output_data["auth_method"] = "iac"
+                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
+
             # check_output Unique ID
             # TODO: move this to a function
             # TODO: in Azure, GCP and K8s there are findings without resource_name

prowler/lib/outputs/summary_table.py

@@ -54,6 +54,9 @@ def display_summary_table(
         elif provider.type == "nhn":
             entity_type = "Tenant Domain"
             audited_entities = provider.identity.tenant_domain
+        elif provider.type == "iac":
+            entity_type = "Directory"
+            audited_entities = provider.scan_path
 
         # Check if there are findings and that they are not all MANUAL
         if findings and not all(finding.status == "MANUAL" for finding in findings):

prowler/providers/common/models.py

@@ -9,6 +9,10 @@ from prowler.providers.common.provider import Provider
 
 # TODO: include this for all the providers
 class Audit_Metadata(BaseModel):
+    provider: str
+    account_id: str
+    account_name: str
+    region: str
     services_scanned: int
     # We can't use a set in the expected
     # checks because the set is unordered

prowler/providers/common/provider.py

@@ -149,7 +149,11 @@ class Provider(ABC):
             provider_class_path = (
                 f"{providers_path}.{arguments.provider}.{arguments.provider}_provider"
             )
-            provider_class_name = f"{arguments.provider.capitalize()}Provider"
+            # Special case for IAC provider
+            if arguments.provider == "iac":
+                provider_class_name = "IACProvider"
+            else:
+                provider_class_name = f"{arguments.provider.capitalize()}Provider"
             provider_class = getattr(
                 import_module(provider_class_path), provider_class_name
             )
@@ -243,6 +247,14 @@ class Provider(ABC):
                         mutelist_path=arguments.mutelist_file,
                         config_path=arguments.config_file,
                     )
+                elif "iac" in provider_class_name.lower():
+                    provider = provider_class(
+                        scan_path=arguments.scan_path,
+                        config_path=arguments.config_file,
+                        mutelist_path=arguments.mutelist_file,
+                        fixer_config=fixer_config,
+                    )
+                    Provider.set_global_provider(provider)
 
         except TypeError as error:
             logger.critical(

prowler/providers/iac/iac.py

@@ -0,0 +1,3 @@
+from prowler.providers.iac.iac_provider import IACProvider
+
+provider = IACProvider(scan_path=globals().get("args", None).scan_path)

prowler/providers/iac/iac_provider.py

@@ -0,0 +1,166 @@
+import json
+import subprocess
+import sys
+from typing import List
+
+from colorama import Fore, Style
+
+from prowler.lib.check.models import CheckReportIAC
+from prowler.lib.logger import logger
+from prowler.lib.utils.utils import print_boxes
+from prowler.providers.common.models import Audit_Metadata
+from prowler.providers.common.provider import Provider
+
+
+class IACProvider(Provider):
+    _type: str = "iac"
+    audit_metadata: Audit_Metadata
+
+    def __init__(
+        self,
+        scan_path: str = ".",
+        config_path: str = None,
+        config_content: dict = None,
+        fixer_config: dict = {},
+        mutelist_path: str = None,
+        mutelist_content: dict = None,
+    ):
+        logger.info("Instantiating IAC Provider...")
+
+        self.scan_path = scan_path
+        self.region = "global"
+        self.audited_account = "local-iac"
+        self._session = None
+        self._identity = "prowler"
+
+        self._audit_config = config_content if config_content else {}
+        self._fixer_config = fixer_config
+        self._mutelist = None
+
+        self.audit_metadata = Audit_Metadata(
+            provider=self._type,
+            account_id=self.audited_account,
+            account_name="iac",
+            region=self.region,
+            services_scanned=0,  # IAC doesn't use services
+            expected_checks=[],  # IAC doesn't use checks
+            completed_checks=0,  # IAC doesn't use checks
+            audit_progress=0,  # IAC doesn't use progress tracking
+        )
+
+        Provider.set_global_provider(self)
+
+    @property
+    def type(self):
+        return self._type
+
+    @property
+    def identity(self):
+        return self._identity
+
+    @property
+    def session(self):
+        return self._session
+
+    @property
+    def audit_config(self):
+        return self._audit_config
+
+    @property
+    def fixer_config(self):
+        return self._fixer_config
+
+    def setup_session(self):
+        """IAC provider doesn't need a session since it uses Checkov directly"""
+        return None
+
+    def run(self) -> List[CheckReportIAC]:
+        return self.run_scan(self.scan_path)
+
+    def run_scan(self, directory: str) -> List[CheckReportIAC]:
+        try:
+            logger.info(f"Running IaC scan on {directory}...")
+
+            # Run Checkov with JSON output
+            process = subprocess.run(
+                ["checkov", "-d", directory, "-o", "json"],
+                capture_output=True,
+                text=True,
+            )
+
+            try:
+                output = json.loads(process.stdout)
+            except json.JSONDecodeError:
+                logger.error("Failed to parse IaC scan output as JSON")
+                return []
+
+            reports = []
+
+            # Process failed checks
+            for finding in output.get("results", {}).get("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()
+
+                # Create a basic metadata structure for the check
+                metadata_dict = {
+                    "Provider": "iac",
+                    "CheckID": finding.get("check_id", ""),
+                    "CheckTitle": finding.get("check_name", ""),
+                    "CheckType": ["Infrastructure as Code"],
+                    "ServiceName": "iac",
+                    "SubServiceName": "",
+                    "ResourceIdTemplate": "",
+                    "Severity": severity,
+                    "ResourceType": "iac",
+                    "Description": finding.get("check_name", ""),
+                    "Risk": "",
+                    "RelatedUrl": finding.get("guideline", ""),
+                    "Remediation": {
+                        "Code": {
+                            "NativeIaC": "",
+                            "Terraform": "",
+                            "CLI": "",
+                            "Other": "",
+                        },
+                        "Recommendation": {
+                            "Text": "",
+                            "Url": finding.get("guideline", ""),
+                        },
+                    },
+                    "Categories": ["security"],
+                    "DependsOn": [],
+                    "RelatedTo": [],
+                    "Notes": "",
+                }
+
+                # Convert metadata dict to JSON string
+                metadata = json.dumps(metadata_dict)
+
+                report = CheckReportIAC(metadata=metadata, finding=finding)
+                reports.append(report)
+
+            # Log summary
+            summary = output.get("summary", {})
+            logger.info(
+                f"IaC scan completed with {summary.get('passed', 0)} passed, "
+                f"{summary.get('failed', 0)} failed, and "
+                f"{summary.get('skipped', 0)} skipped checks."
+            )
+
+            return reports
+
+        except Exception as error:
+            logger.critical(
+                f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
+            )
+            sys.exit(1)
+
+    def print_credentials(self):
+        report_lines = [
+            f"Directory: {Fore.YELLOW}{self.scan_path}{Style.RESET_ALL}",
+        ]
+        report_title = f"{Style.BRIGHT}Scanning local IaC directory:{Style.RESET_ALL}"
+        print_boxes(report_lines, report_title)

prowler/providers/iac/lib/arguments/arguments.py

@@ -0,0 +1,12 @@
+def init_parser(self):
+    """Init the IAC Provider CLI parser"""
+    iac_parser = self.subparsers.add_parser(
+        "iac", parents=[self.common_providers_parser], help="IaC Provider"
+    )
+    iac_parser.add_argument(
+        "--scan-path",
+        "-P",
+        dest="scan_path",
+        default=".",
+        help="Path to the folder containing your infrastructure-as-code files. Default: current directory",
+    )

prowler/providers/iac/models.py

@@ -0,0 +1,27 @@
+from prowler.config.config import output_file_timestamp
+from prowler.providers.common.models import ProviderOutputOptions
+
+
+class IACOutputOptions(ProviderOutputOptions):
+    """
+    IACOutputOptions overrides ProviderOutputOptions for IAC-specific output logic.
+    For example, generating a filename that includes the IAC tenant_id.
+
+    Attributes inherited from ProviderOutputOptions:
+        - output_filename (str): The base filename used for generated reports.
+        - output_directory (str): The directory to store the output files.
+        - ... see ProviderOutputOptions for more details.
+
+    Methods:
+        - __init__: Customizes the output filename logic for IAC.
+    """
+
+    def __init__(self, arguments, bulk_checks_metadata):
+        super().__init__(arguments, bulk_checks_metadata)
+
+        # If --output-filename is not specified, build a default name.
+        if not getattr(arguments, "output_filename", None):
+            self.output_filename = f"prowler-output-iac-{output_file_timestamp}"
+        # If --output-filename was explicitly given, respect that
+        else:
+            self.output_filename = arguments.output_filename

pyproject.toml

@@ -33,7 +33,7 @@ dependencies = [
   "azure-mgmt-subscription==3.1.1",
   "azure-mgmt-web==8.0.0",
   "azure-storage-blob==12.24.1",
-  "boto3==1.35.99",
+  "boto3 (==1.35.49)",
   "botocore==1.35.99",
   "colorama==0.4.6",
   "cryptography==44.0.1",
@@ -48,16 +48,17 @@ dependencies = [
   "msgraph-sdk==1.23.0",
   "numpy==2.0.2",
   "pandas==2.2.3",
-  "py-ocsf-models==0.3.1",
-  "pydantic==1.10.21",
+  "py-ocsf-models==0.5.0",
+  "pydantic (>=2.0,<3.0)",
   "pygithub==2.5.0",
   "python-dateutil (>=2.9.0.post0,<3.0.0)",
   "pytz==2025.1",
-  "schema==0.7.7",
+  "schema (==0.7.5)",
   "shodan==1.31.0",
   "slack-sdk==3.34.0",
   "tabulate==0.9.0",
-  "tzlocal==5.3.1"
+  "tzlocal==5.3.1",
+  "checkov (>=3.2.434,<4.0.0)"
 ]
 description = "Prowler is an Open Source security tool to perform AWS, GCP and Azure security best practices assessments, audits, incident response, continuous monitoring, hardening and forensics readiness. It contains hundreds of controls covering CIS, NIST 800, NIST CSF, CISA, RBI, FedRAMP, PCI-DSS, GDPR, HIPAA, FFIEC, SOC2, GXP, AWS Well-Architected Framework Security Pillar, AWS Foundational Technical Review (FTR), ENS (Spanish National Security Scheme) and your custom security frameworks."
 license = "Apache-2.0"