fix(iac): raise typed exceptions instead of sys.exit on provider failures (#12227)

Co-authored-by: Juhef <117518034+juheff@users.noreply.github.com>
Co-authored-by: Daniel Barranquero <danielbo2001@gmail.com>

prowler/__main__.py

@@ -148,6 +148,7 @@ from prowler.providers.gcp.models import GCPOutputOptions
 from prowler.providers.github.models import GithubOutputOptions
 from prowler.providers.googleworkspace.models import GoogleWorkspaceOutputOptions
 from prowler.providers.huaweicloud.models import HuaweiCloudOutputOptions
+from prowler.providers.iac.exceptions.exceptions import IacBaseException
 from prowler.providers.iac.models import IACOutputOptions
 from prowler.providers.image.exceptions.exceptions import ImageBaseException
 from prowler.providers.image.models import ImageOutputOptions
@@ -565,12 +566,16 @@ def prowler():
                 except ImageBaseException as error:
                     logger.critical(f"{error}")
                     sys.exit(1)
+            elif provider == "iac":
+                try:
+                    findings = global_provider.run()
+                except IacBaseException as error:
+                    logger.critical(f"{error}")
+                    sys.exit(1)
             else:
-                # IAC and external tool-wrapper providers registered via entry
-                # points. Unexpected failures propagate to the outer except
-                # Exception backstop further down in this file — keeping the
-                # branch free of an Image-specific catch that would otherwise
-                # mislead plug-in authors reading this code.
+                # External tool-wrapper providers registered via entry points.
+                # Unexpected failures propagate to the outer except Exception
+                # backstop further down in this file.
                 findings = global_provider.run()
             # Note: External tool providers don't support granular progress tracking since
             # they run external tools as a black box and return all findings at once.

prowler/changelog.d/iac-clone-failure-exit.fixed.md

@@ -0,0 +1 @@
+IaC provider now raises typed `IacBaseException` errors (repository clone, Trivy missing, scan and output processing failures) instead of calling `sys.exit(1)`; the CLI still stops with the logged message, and API scans fail as regular task errors instead of a `SystemExit` escaping the worker

prowler/providers/iac/exceptions/exceptions.py

@@ -0,0 +1,80 @@
+from prowler.exceptions.exceptions import ProwlerException
+
+
+# Exceptions codes from 21000 to 21999 are reserved for IaC exceptions
+class IacBaseException(ProwlerException):
+    """Base class for IaC provider errors."""
+
+    IAC_ERROR_CODES = {
+        (21000, "IacRepositoryCloneError"): {
+            "message": "Unable to clone the repository to scan",
+            "remediation": "Check that the repository URL is correct, that it is reachable, and that the provided credentials can read it.",
+        },
+        (21001, "IacTrivyNotFoundError"): {
+            "message": "Trivy binary not found",
+            "remediation": "Install Trivy from https://trivy.dev/latest/getting-started/installation/ or use your system package manager (e.g., 'brew install trivy' on macOS, 'apt-get install trivy' on Ubuntu).",
+        },
+        (21002, "IacScanError"): {
+            "message": "Error running the IaC scan",
+            "remediation": "Check the Trivy output and the scanned path, then try again.",
+        },
+        (21003, "IacOutputProcessingError"): {
+            "message": "Error processing the IaC scan output",
+            "remediation": "Check the Trivy output format and try again.",
+        },
+    }
+
+    def __init__(self, code, file=None, original_exception=None, message=None):
+        provider = "IaC"
+        error_info = self.IAC_ERROR_CODES.get((code, self.__class__.__name__))
+        if error_info is None:
+            error_info = {
+                "message": message or "Unknown IaC error",
+                "remediation": "Check the error message and try again.",
+            }
+        elif message:
+            error_info = error_info.copy()
+            error_info["message"] = message
+        super().__init__(
+            code=code,
+            source=provider,
+            file=file,
+            original_exception=original_exception,
+            error_info=error_info,
+        )
+
+
+class IacRepositoryCloneError(IacBaseException):
+    """Exception raised when the repository to scan cannot be cloned."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            21000, file=file, original_exception=original_exception, message=message
+        )
+
+
+class IacTrivyNotFoundError(IacBaseException):
+    """Exception raised when the Trivy binary is not available."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            21001, file=file, original_exception=original_exception, message=message
+        )
+
+
+class IacScanError(IacBaseException):
+    """Exception raised when the Trivy scan fails."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            21002, file=file, original_exception=original_exception, message=message
+        )
+
+
+class IacOutputProcessingError(IacBaseException):
+    """Exception raised when a Trivy finding cannot be processed."""
+
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            21003, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/iac/iac_provider.py

@@ -24,6 +24,13 @@ from prowler.lib.utils.vulnerability_references import (
 )
 from prowler.providers.common.models import Audit_Metadata, Connection
 from prowler.providers.common.provider import Provider
+from prowler.providers.iac.exceptions.exceptions import (
+    IacBaseException,
+    IacOutputProcessingError,
+    IacRepositoryCloneError,
+    IacScanError,
+    IacTrivyNotFoundError,
+)
 
 
 class IacProvider(Provider):
@@ -270,7 +277,9 @@ class IacProvider(Provider):
             logger.critical(
                 f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
             )
-            sys.exit(1)
+            raise IacOutputProcessingError(
+                file=__file__, original_exception=error
+            ) from error
 
     def _detect_branch_name(self, repo_path: str) -> str:
         """
@@ -376,6 +385,9 @@ class IacProvider(Provider):
             logger.critical(
                 f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
             )
+            raise IacRepositoryCloneError(
+                file=__file__, original_exception=error
+            ) from error
 
     def run(self) -> List[CheckReportIAC]:
         """
@@ -499,7 +511,7 @@ class IacProvider(Provider):
                 logger.critical(
                     f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
                 )
-                sys.exit(1)
+                raise IacScanError(file=__file__, original_exception=error) from error
 
             batch = []
             batch_size = 100
@@ -551,16 +563,20 @@ class IacProvider(Provider):
             if batch:
                 yield batch
 
+        except IacBaseException:
+            raise
         except Exception as error:
             if "No such file or directory: 'trivy'" in str(error):
                 logger.critical(
                     "Trivy binary not found. Please install Trivy from https://trivy.dev/latest/getting-started/installation/ or use your system package manager (e.g., 'brew install trivy' on macOS, 'apt-get install trivy' on Ubuntu)"
                 )
-                sys.exit(1)
+                raise IacTrivyNotFoundError(
+                    file=__file__, original_exception=error
+                ) from error
             logger.critical(
                 f"{error.__class__.__name__}:{error.__traceback__.tb_lineno} -- {error}"
             )
-            sys.exit(1)
+            raise IacScanError(file=__file__, original_exception=error) from error
 
     def print_credentials(self):
         if self.scan_repository_url:

tests/providers/iac/iac_provider_test.py

@@ -7,6 +7,10 @@ from unittest.mock import MagicMock, patch
 import pytest
 
 from prowler.lib.check.models import CheckReportIAC
+from prowler.providers.iac.exceptions.exceptions import (
+    IacRepositoryCloneError,
+    IacScanError,
+)
 from prowler.providers.iac.iac_provider import IacProvider
 from tests.providers.iac.iac_fixtures import (
     DEFAULT_SCAN_PATH,
@@ -323,13 +327,11 @@ class TestIacProvider:
             stdout=get_invalid_trivy_output(), stderr=""
         )
 
-        with pytest.raises(SystemExit) as excinfo:
+        with pytest.raises(IacScanError):
             # Consume the generator
             for _ in provider.run_scan("/test/directory", ["all"], []):
                 pass
 
-        assert excinfo.value.code == 1
-
     @patch("subprocess.run")
     def test_iac_provider_run_scan_null_output(self, mock_subprocess):
         """Test IAC scan with null Trivy output"""
@@ -337,13 +339,12 @@ class TestIacProvider:
 
         mock_subprocess.return_value = MagicMock(stdout="null", stderr="")
 
-        with pytest.raises(SystemExit) as exc_info:
+        with pytest.raises(IacScanError):
             # Consume the generator
             for _ in provider.run_scan(
                 "/test/directory", ["vuln", "misconfig", "secret"], []
             ):
                 pass
-        assert exc_info.value.code == 1
 
     def test_iac_provider_process_finding_dockerfile(self):
         """Test processing a Dockerfile finding"""
@@ -514,15 +515,13 @@ class TestIacProvider:
         # Make subprocess.run raise an exception
         mock_subprocess.side_effect = Exception("Test exception")
 
-        with pytest.raises(SystemExit) as exc_info:
+        with pytest.raises(IacScanError):
             # Consume the generator
             for _ in provider.run_scan(
                 "/test/directory", ["vuln", "misconfig", "secret"], []
             ):
                 pass
 
-        assert exc_info.value.code == 1
-
     @patch("subprocess.run")
     def test_run_scan_with_different_frameworks(self, mock_subprocess):
         """Test run_scan with different scanner configurations"""
@@ -817,6 +816,26 @@ class TestIacProvider:
         assert temp_dir == "/tmp/fake-dir"
         assert branch_name == "master"
 
+    @mock.patch("prowler.providers.iac.iac_provider.porcelain.clone")
+    @mock.patch("tempfile.mkdtemp", return_value="/tmp/fake-dir")
+    def test_clone_repository_failure_raises(self, _mock_mkdtemp, mock_clone):
+        """A failed clone must raise a typed error instead of returning `None`.
+
+        `_clone_repository` is annotated `-> tuple[str, str]` and `__init__`
+        unpacks the result directly, so falling through the error handler used
+        to raise `TypeError: cannot unpack non-sequence NoneType`. Raising
+        `IacRepositoryCloneError` lets the CLI exit with the logged message and
+        lets the API report the failure as a normal task error instead of a
+        `SystemExit` escaping the Celery worker.
+        """
+        mock_clone.side_effect = Exception("repository not found")
+
+        with pytest.raises(IacRepositoryCloneError) as exc_info:
+            IacProvider(scan_repository_url="https://github.com/user/repo.git")
+
+        assert "repository not found" in str(exc_info.value)
+        assert exc_info.value.code == 21000
+
     def test_detect_branch_name_main(self):
         """Test detecting 'main' branch from .git/HEAD"""
         provider = IacProvider()