feat(gcp): add custom exceptions clas (#4908)

prowler/providers/gcp/exceptions/exceptions.py

@@ -0,0 +1,87 @@
+from prowler.exceptions.exceptions import ProwlerException
+
+
+class GCPBaseException(ProwlerException):
+    """Base class for GCP Errors."""
+
+    GCP_ERROR_CODES = {
+        (1925, "GCPCloudResourceManagerAPINotUsedError"): {
+            "message": "Cloud Resource Manager API not used",
+            "remediation": "Enable the Cloud Resource Manager API for the project.",
+        },
+        (1926, "GCPHTTPError"): {
+            "message": "HTTP error",
+            "remediation": "Check the HTTP error and ensure the request is properly formatted.",
+        },
+        (1927, "GCPNoAccesibleProjectsError"): {
+            "message": "No Project IDs can be accessed via Google Credentials",
+            "remediation": "Ensure the project is accessible and properly set up.",
+        },
+        (1928, "GCPSetUpSessionError"): {
+            "message": "Error setting up session",
+            "remediation": "Check the session setup and ensure it is properly set up.",
+        },
+        (1929, "GCPGetProjectError"): {
+            "message": "Error getting project",
+            "remediation": "Check the project and ensure it is properly set up.",
+        },
+        (1930, "GCPTestConnectionError"): {
+            "message": "Error testing connection to GCP",
+            "remediation": "Check the connection and ensure it is properly set up.",
+        },
+    }
+
+    def __init__(self, code, file=None, original_exception=None, message=None):
+        provider = "GCP"
+        error_info = self.GCP_ERROR_CODES.get((code, self.__class__.__name__))
+        if message:
+            error_info["message"] = message
+        super().__init__(
+            code=code,
+            provider=provider,
+            file=file,
+            original_exception=original_exception,
+            error_info=error_info,
+        )
+
+
+class GCPCloudResourceManagerAPINotUsedError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1925, file=file, original_exception=original_exception, message=message
+        )
+
+
+class GCPHTTPError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1926, file=file, original_exception=original_exception, message=message
+        )
+
+
+class GCPNoAccesibleProjectsError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1927, file=file, original_exception=original_exception, message=message
+        )
+
+
+class GCPSetUpSessionError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1928, file=file, original_exception=original_exception, message=message
+        )
+
+
+class GCPGetProjectError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1929, file=file, original_exception=original_exception, message=message
+        )
+
+
+class GCPTestConnectionError(GCPBaseException):
+    def __init__(self, file=None, original_exception=None, message=None):
+        super().__init__(
+            1930, file=file, original_exception=original_exception, message=message
+        )

prowler/providers/gcp/gcp_provider.py

@@ -17,6 +17,14 @@ from prowler.lib.logger import logger
 from prowler.lib.utils.utils import print_boxes
 from prowler.providers.common.models import Audit_Metadata, Connection
 from prowler.providers.common.provider import Provider
+from prowler.providers.gcp.exceptions.exceptions import (
+    GCPCloudResourceManagerAPINotUsedError,
+    GCPGetProjectError,
+    GCPHTTPError,
+    GCPNoAccesibleProjectsError,
+    GCPSetUpSessionError,
+    GCPTestConnectionError,
+)
 from prowler.providers.gcp.lib.mutelist.mutelist import GCPMutelist
 from prowler.providers.gcp.models import (
     GCPIdentityInfo,
@@ -73,8 +81,10 @@ class GcpProvider(Provider):
         accessible_projects = self.get_projects()
         if not accessible_projects:
             logger.critical("No Project IDs can be accessed via Google Credentials.")
-            # TODO: add custom exception once we have the GCP exceptions
-            raise SystemExit
+            raise GCPNoAccesibleProjectsError(
+                file=__file__,
+                message="No Project IDs can be accessed via Google Credentials.",
+            )
 
         if project_ids:
             for input_project in project_ids:
@@ -106,8 +116,10 @@ class GcpProvider(Provider):
             logger.critical(
                 "No Input Project IDs can be accessed via Google Credentials."
             )
-            # TODO: add custom exception once we have the GCP exceptions
-            raise SystemExit
+            raise GCPNoAccesibleProjectsError(
+                file=__file__,
+                message="No Input Project IDs can be accessed via Google Credentials.",
+            )
 
         if list_project_ids:
             print(
@@ -256,7 +268,7 @@ class GcpProvider(Provider):
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
-            raise error
+            raise GCPSetUpSessionError(file=__file__, original_exception=error)
 
     @staticmethod
     def test_connection(
@@ -281,13 +293,22 @@ class GcpProvider(Provider):
             request = service.projects().list()
             request.execute()
             return Connection(is_connected=True)
+
+        # Errors from setup_session
+        except GCPSetUpSessionError as setup_session_error:
+            logger.critical(str(setup_session_error))
+            if raise_on_exception:
+                raise setup_session_error
+            return Connection(error=setup_session_error)
         except HttpError as http_error:
             if "Cloud Resource Manager API has not been used" in str(http_error):
                 logger.critical(
                     "Cloud Resource Manager API has not been used before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/api/cloudresourcemanager.googleapis.com/ then retry."
                 )
                 if raise_on_exception:
-                    raise http_error
+                    raise GCPCloudResourceManagerAPINotUsedError(
+                        file=__file__, original_exception=http_error
+                    )
             else:
                 logger.critical(
                     f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error}"
@@ -300,7 +321,7 @@ class GcpProvider(Provider):
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
             if raise_on_exception:
-                raise error
+                raise GCPTestConnectionError(file=__file__, original_exception=error)
             return Connection(error=error)
 
     def print_credentials(self):
@@ -374,16 +395,19 @@ class GcpProvider(Provider):
                 logger.critical(
                     "Cloud Resource Manager API has not been used before or it is disabled. Enable it by visiting https://console.developers.google.com/apis/api/cloudresourcemanager.googleapis.com/ then retry."
                 )
-                raise http_error
+                raise GCPCloudResourceManagerAPINotUsedError(
+                    file=__file__, original_exception=http_error
+                )
             else:
                 logger.error(
                     f"{http_error.__class__.__name__}[{http_error.__traceback__.tb_lineno}]: {http_error}"
                 )
+                raise GCPHTTPError(file=__file__, original_exception=http_error)
         except Exception as error:
             logger.critical(
                 f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
             )
-            raise error
+            raise GCPGetProjectError(file=__file__, original_exception=error)
         finally:
             return projects