feat(exceptions): add custom error for provider connection during scans (#8234)

api/CHANGELOG.md

@@ -18,9 +18,11 @@ All notable changes to the **Prowler API** are documented in this file.
 
 ## [v1.9.1] (Prowler v5.8.1)
 
+### Added
+- Custom exception for provider connection errors during scans [(#8234)](https://github.com/prowler-cloud/prowler/pull/8234)
+
 ### Changed
 - Summary and overview tasks now use a dedicated queue and no longer propagate errors to compliance tasks [(#8214)](https://github.com/prowler-cloud/prowler/pull/8214)
-- Optimized include parameters for resources view ([#8229])(https://github.com/prowler-cloud/prowler/pull/8229)
 
 ### Fixed
 - Scan with no resources will not trigger legacy code for findings metadata [(#8183)](https://github.com/prowler-cloud/prowler/pull/8183)

api/src/backend/api/exceptions.py

@@ -57,6 +57,11 @@ class TaskInProgressException(TaskManagementError):
         super().__init__()
 
 
+# Provider connection errors
+class ProviderConnectionError(Exception):
+    """Base exception for provider connection errors."""
+
+
 def custom_exception_handler(exc, context):
     if isinstance(exc, django_validation_error):
         if hasattr(exc, "error_dict"):

api/src/backend/config/settings/sentry.py

@@ -4,6 +4,7 @@ from config.env import env
 IGNORED_EXCEPTIONS = [
     # Provider is not connected due to credentials errors
     "is not connected",
+    "ProviderConnectionError",
     # Authentication Errors from AWS
     "InvalidToken",
     "AccessDeniedException",
@@ -16,7 +17,7 @@ IGNORED_EXCEPTIONS = [
     "InternalServerErrorException",
     "AccessDenied",
     "No Shodan API Key",  # Shodan Check
-    "RequestLimitExceeded",  # For now we don't want to log the RequestLimitExceeded errors
+    "RequestLimitExceeded",  # For now, we don't want to log the RequestLimitExceeded errors
     "ThrottlingException",
     "Rate exceeded",
     "SubscriptionRequiredException",
@@ -42,7 +43,9 @@ IGNORED_EXCEPTIONS = [
     "AWSAccessKeyIDInvalidError",
     "AWSSessionTokenExpiredError",
     "EndpointConnectionError",  # AWS Service is not available in a region
-    "Pool is closed",  # The following comes from urllib3: eu-west-1 -- HTTPClientError[126]: An HTTP Client raised an unhandled exception: AWSHTTPSConnectionPool(host='hostname.s3.eu-west-1.amazonaws.com', port=443): Pool is closed.
+    # The following comes from urllib3: eu-west-1 -- HTTPClientError[126]: An HTTP Client raised an
+    # unhandled exception: AWSHTTPSConnectionPool(host='hostname.s3.eu-west-1.amazonaws.com', port=443): Pool is closed.
+    "Pool is closed",
     # Authentication Errors from GCP
     "ClientAuthenticationError",
     "AuthorizationFailed",
@@ -71,7 +74,7 @@ IGNORED_EXCEPTIONS = [
 
 def before_send(event, hint):
     """
-    before_send handles the Sentry events in order to sent them or not
+    before_send handles the Sentry events in order to send them or not
     """
     # Ignore logs with the ignored_exceptions
     # https://docs.python.org/3/library/logging.html#logrecord-objects

api/src/backend/tasks/jobs/scan.py

@@ -14,6 +14,7 @@ from api.compliance import (
     generate_scan_compliance,
 )
 from api.db_utils import create_objects_in_batches, rls_transaction
+from api.exceptions import ProviderConnectionError
 from api.models import (
     ComplianceRequirementOverview,
     Finding,
@@ -154,7 +155,7 @@ def perform_prowler_scan(
                 provider_instance.connected = True
             except Exception as e:
                 provider_instance.connected = False
-                exc = ValueError(
+                exc = ProviderConnectionError(
                     f"Provider {provider_instance.provider} is not connected: {e}"
                 )
             finally:

api/src/backend/tasks/tests/test_scan.py

@@ -12,6 +12,7 @@ from tasks.jobs.scan import (
 )
 from tasks.utils import CustomEncoder
 
+from api.exceptions import ProviderConnectionError
 from api.models import (
     ComplianceRequirementOverview,
     Finding,
@@ -204,7 +205,7 @@ class TestPerformScan:
         provider_id = str(provider.id)
         checks_to_execute = ["check1", "check2"]
 
-        with pytest.raises(ValueError):
+        with pytest.raises(ProviderConnectionError):
             perform_prowler_scan(tenant_id, scan_id, provider_id, checks_to_execute)
 
         scan.refresh_from_db()