feat(sns): add Amazon SNS integration for email alerts

Add complete Amazon SNS integration to Prowler that allows sending security
findings as email alerts via SNS topics. The integration supports comprehensive
filtering by severity, provider, region, resource name, and resource tags.

Features:
- SNS topic-based email alerting system
- AWS credential authentication (access keys, roles, session tokens)
- Support for filtering findings before dispatch
- Async task processing with Celery
- Full CRUD operations for SNS integrations
- Connection testing and validation

SNS Client (prowler/lib):
- SNS class for publishing finding alerts to topics
- Email-formatted messages with comprehensive finding details
- Support for remediation recommendations and code examples
- Exception handling with custom error classes
- Connection testing with topic validation

Backend API (api/src):
- IntegrationSNSFindingsFilter with severity, region, provider, and resource filtering
- IntegrationSNSDispatchSerializer for dispatch validation
- IntegrationSNSViewSet with RBAC permissions
- SNS integration task (sns_integration_task)
- Job logic (send_findings_to_sns) for batch processing

Models & Serializers:
- Added SNS to Integration.IntegrationChoices
- SNSConfigSerializer with topic_arn validation
- Uses AWSCredentialSerializer for AWS authentication
- Connection testing integrated into utils

Email Alert Format:
- Subject: [Prowler Alert] SEVERITY - CHECK_ID - RESOURCE_NAME
- Body: Comprehensive text format with:
  - Finding details (severity, status, check info)
  - Resource information (name, type, UID, region, account, provider)
  - Risk description
  - Remediation recommendations with URLs
  - Remediation code (CLI, Terraform, Other)
  - Resource tags
  - Compliance framework mappings
  - Link back to Prowler UI

API Endpoints:
- POST /api/v1/integrations (create SNS integration)
- POST /api/v1/integrations/{id}/connection (test SNS topic)
- POST /api/v1/integrations/{id}/sns/dispatches (send filtered findings)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <noreply@anthropic.com>

api/src/backend/api/filters.py

@@ -977,6 +977,48 @@ class IntegrationJiraFindingsFilter(FilterSet):
         return super().filter_queryset(queryset)
 
 
+class IntegrationSNSFindingsFilter(FilterSet):
+    """Filter for SNS integration with support for severity, region, provider, resource name, and tag filtering."""
+
+    finding_id = UUIDFilter(field_name="id", lookup_expr="exact")
+    finding_id__in = UUIDInFilter(field_name="id", lookup_expr="in")
+
+    # Severity filtering
+    severity = ChoiceFilter(choices=SeverityChoices)
+    severity__in = ChoiceInFilter(choices=SeverityChoices, field_name="severity")
+
+    # Provider filtering
+    provider = UUIDFilter(field_name="scan__provider__id", lookup_expr="exact")
+    provider__in = UUIDInFilter(field_name="scan__provider__id", lookup_expr="in")
+    provider_type = ChoiceFilter(
+        choices=Provider.ProviderChoices.choices, field_name="scan__provider__provider"
+    )
+
+    # Region filtering
+    region = CharFilter(field_name="region", lookup_expr="exact")
+    region__in = CharInFilter(field_name="region", lookup_expr="in")
+    region__icontains = CharFilter(field_name="region", lookup_expr="icontains")
+
+    # Resource filtering
+    resource_name = CharFilter(field_name="resources__name", lookup_expr="icontains")
+    resource_uid = CharFilter(field_name="resources__uid", lookup_expr="exact")
+    resource_tags = CharFilter(field_name="resources__tags", lookup_expr="icontains")
+
+    class Meta:
+        model = Finding
+        fields = {}
+
+    def filter_queryset(self, queryset):
+        # Validate that there is at least one filter provided
+        if not self.data:
+            raise ValidationError(
+                {
+                    "findings": "No finding filters provided. At least one filter is required."
+                }
+            )
+        return super().filter_queryset(queryset)
+
+
 class TenantApiKeyFilter(FilterSet):
     inserted_at = DateFilter(field_name="created", lookup_expr="date")
     inserted_at__gte = DateFilter(field_name="created", lookup_expr="gte")

api/src/backend/api/models.py

@@ -1586,8 +1586,10 @@ class Integration(RowLevelSecurityProtectedModel):
     class IntegrationChoices(models.TextChoices):
         AMAZON_S3 = "amazon_s3", _("Amazon S3")
         AWS_SECURITY_HUB = "aws_security_hub", _("AWS Security Hub")
+        GITHUB = "github", _("GitHub")
         JIRA = "jira", _("JIRA")
         SLACK = "slack", _("Slack")
+        SNS = "sns", _("Amazon SNS")
 
     id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
     inserted_at = models.DateTimeField(auto_now_add=True, editable=False)

api/src/backend/api/utils.py

@@ -15,6 +15,7 @@ from prowler.providers.alibabacloud.alibabacloud_provider import AlibabacloudPro
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.s3.s3 import S3
 from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
+from prowler.providers.aws.lib.sns.sns import SNS
 from prowler.providers.azure.azure_provider import AzureProvider
 from prowler.providers.common.models import Connection
 from prowler.providers.gcp.gcp_provider import GcpProvider
@@ -297,6 +298,12 @@ def prowler_integration_connection_test(integration: Integration) -> Connection:
             integration.configuration["projects"] = project_keys
             integration.save()
         return jira_connection
+    elif integration.integration_type == Integration.IntegrationChoices.SNS:
+        return SNS.test_connection(
+            **integration.credentials,
+            topic_arn=integration.configuration["topic_arn"],
+            raise_on_exception=False,
+        )
     elif integration.integration_type == Integration.IntegrationChoices.SLACK:
         pass
     else:
@@ -406,7 +413,7 @@ def get_findings_metadata_no_aggregations(tenant_id: str, filtered_queryset):
     return serializer.data
 
 
-def initialize_prowler_integration(integration: Integration) -> Jira:
+def initialize_prowler_integration(integration: Integration):
     # TODO Refactor other integrations to use this function
     if integration.integration_type == Integration.IntegrationChoices.JIRA:
         try:
@@ -418,3 +425,15 @@ def initialize_prowler_integration(integration: Integration) -> Jira:
                 integration.connection_last_checked_at = datetime.now(tz=timezone.utc)
                 integration.save()
             raise jira_auth_error
+    elif integration.integration_type == Integration.IntegrationChoices.SNS:
+        try:
+            return SNS(
+                topic_arn=integration.configuration["topic_arn"],
+                **integration.credentials,
+            )
+        except Exception as sns_error:
+            with rls_transaction(str(integration.tenant_id)):
+                integration.connected = False
+                integration.connection_last_checked_at = datetime.now(tz=timezone.utc)
+                integration.save()
+            raise sns_error

api/src/backend/api/v1/serializer_utils/integrations.py

@@ -67,6 +67,31 @@ class SecurityHubConfigSerializer(BaseValidateSerializer):
         resource_name = "integrations"
 
 
+class SNSConfigSerializer(BaseValidateSerializer):
+    topic_arn = serializers.CharField(required=True)
+
+    def validate_topic_arn(self, value):
+        """
+        Validate the topic_arn field to ensure it's a properly formatted SNS topic ARN.
+        """
+        if not value:
+            raise serializers.ValidationError("SNS topic ARN is required")
+
+        # Check if it matches the SNS ARN pattern: arn:partition:sns:region:account-id:topic-name
+        arn_pattern = (
+            r"^arn:(aws|aws-cn|aws-us-gov):sns:[a-z0-9-]+:\d{12}:[a-zA-Z0-9_-]+$"
+        )
+        if not re.match(arn_pattern, value):
+            raise serializers.ValidationError(
+                "Invalid SNS topic ARN format. Expected: arn:partition:sns:region:account-id:topic-name"
+            )
+
+        return value
+
+    class Meta:
+        resource_name = "integrations"
+
+
 class JiraConfigSerializer(BaseValidateSerializer):
     domain = serializers.CharField(read_only=True)
     issue_types = serializers.ListField(
@@ -229,6 +254,19 @@ class IntegrationCredentialField(serializers.JSONField):
                 "properties": {},
                 "additionalProperties": False,
             },
+            {
+                "type": "object",
+                "title": "Amazon SNS",
+                "properties": {
+                    "topic_arn": {
+                        "type": "string",
+                        "description": "The Amazon Resource Name (ARN) of the SNS topic to send alerts to. Format: "
+                        "arn:partition:sns:region:account-id:topic-name",
+                        "pattern": "^arn:(aws|aws-cn|aws-us-gov):sns:[a-z0-9-]+:\\d{12}:[a-zA-Z0-9_-]+$",
+                    },
+                },
+                "required": ["topic_arn"],
+            },
         ]
     }
 )

api/src/backend/api/v1/serializers.py

@@ -60,6 +60,7 @@ from api.v1.serializer_utils.integrations import (
     JiraCredentialSerializer,
     S3ConfigSerializer,
     SecurityHubConfigSerializer,
+    SNSConfigSerializer,
 )
 from api.v1.serializer_utils.lighthouse import (
     BedrockCredentialsSerializer,
@@ -2432,6 +2433,15 @@ class BaseWriteIntegrationSerializer(BaseWriteSerializer):
                     )
             config_serializer = SecurityHubConfigSerializer
             credentials_serializers = [AWSCredentialSerializer]
+        elif integration_type == Integration.IntegrationChoices.SNS:
+            if providers:
+                raise serializers.ValidationError(
+                    {
+                        "providers": "Relationship field is not accepted. This integration applies to all providers."
+                    }
+                )
+            config_serializer = SNSConfigSerializer
+            credentials_serializers = [AWSCredentialSerializer]
         elif integration_type == Integration.IntegrationChoices.JIRA:
             if providers:
                 raise serializers.ValidationError(
@@ -2704,6 +2714,40 @@ class IntegrationJiraDispatchSerializer(BaseSerializerV1):
         return validated_attrs
 
 
+class IntegrationSNSDispatchSerializer(BaseSerializerV1):
+    """
+    Serializer for dispatching findings to SNS integration as email alerts.
+    Supports filtering by severity, region, provider, resource name, and tags.
+    """
+
+    class JSONAPIMeta:
+        resource_name = "integrations-sns-dispatches"
+
+    def validate(self, attrs):
+        validated_attrs = super().validate(attrs)
+        integration_instance = Integration.objects.get(
+            id=self.context.get("integration_id")
+        )
+        if integration_instance.integration_type != Integration.IntegrationChoices.SNS:
+            raise ValidationError(
+                {"integration_type": "The given integration is not an SNS integration"}
+            )
+
+        if not integration_instance.enabled:
+            raise ValidationError(
+                {"integration": "The given integration is not enabled"}
+            )
+
+        if not integration_instance.connected:
+            raise ValidationError(
+                {
+                    "integration": "The SNS integration is not connected. Please test the connection first."
+                }
+            )
+
+        return validated_attrs
+
+
 # Processors
 
 

api/src/backend/api/v1/urls.py

@@ -13,6 +13,7 @@ from api.v1.views import (
     GithubSocialLoginView,
     GoogleSocialLoginView,
     IntegrationJiraViewSet,
+    IntegrationSNSViewSet,
     IntegrationViewSet,
     InvitationAcceptViewSet,
     InvitationViewSet,
@@ -97,6 +98,7 @@ integrations_router = routers.NestedSimpleRouter(
 integrations_router.register(
     r"jira", IntegrationJiraViewSet, basename="integration-jira"
 )
+integrations_router.register(r"sns", IntegrationSNSViewSet, basename="integration-sns")
 
 urlpatterns = [
     path("tokens", CustomTokenObtainView.as_view(), name="token-obtain"),

api/src/backend/api/v1/views.py

@@ -87,6 +87,7 @@ from tasks.tasks import (
     mute_historical_findings_task,
     perform_scan_task,
     refresh_lighthouse_provider_models_task,
+    sns_integration_task,
 )
 
 from api.base_views import BaseRLSViewSet, BaseTenantViewset, BaseUserViewset
@@ -106,6 +107,7 @@ from api.filters import (
     FindingFilter,
     IntegrationFilter,
     IntegrationJiraFindingsFilter,
+    IntegrationSNSFindingsFilter,
     InvitationFilter,
     LatestFindingFilter,
     LatestResourceFilter,
@@ -192,6 +194,7 @@ from api.v1.serializers import (
     IntegrationCreateSerializer,
     IntegrationJiraDispatchSerializer,
     IntegrationSerializer,
+    IntegrationSNSDispatchSerializer,
     IntegrationUpdateSerializer,
     InvitationAcceptSerializer,
     InvitationCreateSerializer,
@@ -5211,6 +5214,72 @@ class IntegrationJiraViewSet(BaseRLSViewSet):
         )
 
 
+class IntegrationSNSViewSet(BaseRLSViewSet):
+    queryset = Finding.all_objects.all()
+    serializer_class = IntegrationSNSDispatchSerializer
+    http_method_names = ["post"]
+    filter_backends = [CustomDjangoFilterBackend]
+    filterset_class = IntegrationSNSFindingsFilter
+    # RBAC required permissions
+    required_permissions = [Permissions.MANAGE_INTEGRATIONS]
+
+    @extend_schema(exclude=True)
+    def create(self, request, *args, **kwargs):
+        raise MethodNotAllowed(method="POST")
+
+    def get_queryset(self):
+        tenant_id = self.request.tenant_id
+        user_roles = get_role(self.request.user)
+        if user_roles.unlimited_visibility:
+            # User has unlimited visibility, return all findings
+            queryset = Finding.all_objects.filter(tenant_id=tenant_id)
+        else:
+            # User lacks permission, filter findings based on provider groups associated with the role
+            queryset = Finding.all_objects.filter(
+                scan__provider__in=get_providers(user_roles)
+            )
+
+        return queryset
+
+    @action(detail=False, methods=["post"], url_name="dispatches")
+    def dispatches(self, request, integration_pk=None):
+        get_object_or_404(Integration, pk=integration_pk)
+        serializer = self.get_serializer(
+            data=request.data, context={"integration_id": integration_pk}
+        )
+        serializer.is_valid(raise_exception=True)
+
+        if self.filter_queryset(self.get_queryset()).count() == 0:
+            raise ValidationError(
+                {"findings": "No findings match the provided filters"}
+            )
+
+        finding_ids = [
+            str(finding_id)
+            for finding_id in self.filter_queryset(self.get_queryset()).values_list(
+                "id", flat=True
+            )
+        ]
+
+        with transaction.atomic():
+            task = sns_integration_task.delay(
+                tenant_id=self.request.tenant_id,
+                integration_id=integration_pk,
+                finding_ids=finding_ids,
+            )
+        prowler_task = Task.objects.get(id=task.id)
+        serializer = TaskSerializer(prowler_task)
+        return Response(
+            data=serializer.data,
+            status=status.HTTP_202_ACCEPTED,
+            headers={
+                "Content-Location": reverse(
+                    "task-detail", kwargs={"pk": prowler_task.id}
+                )
+            },
+        )
+
+
 @extend_schema_view(
     list=extend_schema(
         tags=["Lighthouse AI"],

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

@@ -17,11 +17,11 @@ from prowler.lib.outputs.html.html import HTML
 from prowler.lib.outputs.ocsf.ocsf import OCSF
 from prowler.providers.aws.aws_provider import AwsProvider
 from prowler.providers.aws.lib.s3.s3 import S3
-from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
-from prowler.providers.common.models import Connection
 from prowler.providers.aws.lib.security_hub.exceptions.exceptions import (
     SecurityHubNoEnabledRegionsError,
 )
+from prowler.providers.aws.lib.security_hub.security_hub import SecurityHub
+from prowler.providers.common.models import Connection
 
 logger = get_task_logger(__name__)
 
@@ -509,3 +509,85 @@ def send_findings_to_jira(
         "created_count": num_tickets_created,
         "failed_count": len(finding_ids) - num_tickets_created,
     }
+
+
+def send_findings_to_sns(
+    tenant_id: str,
+    integration_id: str,
+    finding_ids: list[str],
+):
+    with rls_transaction(tenant_id):
+        integration = Integration.objects.get(id=integration_id)
+        sns_integration = initialize_prowler_integration(integration)
+
+    num_alerts_sent = 0
+    for finding_id in finding_ids:
+        with rls_transaction(tenant_id):
+            finding_instance = (
+                Finding.all_objects.select_related("scan__provider")
+                .prefetch_related("resources")
+                .get(id=finding_id)
+            )
+
+            # Extract resource information
+            resource = (
+                finding_instance.resources.first()
+                if finding_instance.resources.exists()
+                else None
+            )
+            resource_uid = resource.uid if resource else ""
+            resource_name = resource.name if resource else ""
+            resource_type = resource.type if resource else ""
+            resource_tags = {}
+            if resource and hasattr(resource, "tags"):
+                resource_tags = resource.get_tags(tenant_id)
+
+            # Get region
+            region = resource.region if resource and resource.region else ""
+
+            # Extract remediation information from check_metadata
+            check_metadata = finding_instance.check_metadata
+            remediation = check_metadata.get("remediation", {})
+            recommendation = remediation.get("recommendation", {})
+            remediation_code = remediation.get("code", {})
+
+            # Build finding data for SNS
+            finding_data = {
+                "severity": finding_instance.severity,
+                "status": finding_instance.status,
+                "check_id": finding_instance.check_id,
+                "check_title": check_metadata.get("checktitle", ""),
+                "resource_name": resource_name,
+                "resource_type": resource_type,
+                "resource_uid": resource_uid,
+                "region": region,
+                "account_id": finding_instance.scan.provider.uid,
+                "service": check_metadata.get("service", ""),
+                "provider": finding_instance.scan.provider.provider,
+                "risk": check_metadata.get("risk", ""),
+                "remediation_recommendation_text": recommendation.get("text", ""),
+                "remediation_recommendation_url": recommendation.get("url", ""),
+                "remediation_code_cli": remediation_code.get("cli", ""),
+                "remediation_code_terraform": remediation_code.get("terraform", ""),
+                "remediation_code_other": remediation_code.get("other", ""),
+                "resource_tags": resource_tags,
+                "compliance": finding_instance.compliance or {},
+                "prowler_url": f"https://prowler.com/findings/{finding_id}",  # Adjust URL as needed
+            }
+
+            # Send the individual finding to SNS
+            result = sns_integration.send_finding(finding_data)
+            if result.get("success"):
+                num_alerts_sent += 1
+                logger.info(
+                    f"Successfully sent finding {finding_id} to SNS. Message ID: {result.get('message_id')}"
+                )
+            else:
+                logger.error(
+                    f"Failed to send finding {finding_id} to SNS: {result.get('error')}"
+                )
+
+    return {
+        "created_count": num_alerts_sent,
+        "failed_count": len(finding_ids) - num_alerts_sent,
+    }

api/src/backend/tasks/tasks.py

@@ -29,6 +29,7 @@ from tasks.jobs.export import (
 )
 from tasks.jobs.integrations import (
     send_findings_to_jira,
+    send_findings_to_sns,
     upload_s3_integration,
     upload_security_hub_integration,
 )
@@ -808,6 +809,19 @@ def jira_integration_task(
     )
 
 
+@shared_task(
+    base=RLSTask,
+    name="integration-sns",
+    queue="integrations",
+)
+def sns_integration_task(
+    tenant_id: str,
+    integration_id: str,
+    finding_ids: list[str],
+):
+    return send_findings_to_sns(tenant_id, integration_id, finding_ids)
+
+
 @shared_task(
     base=RLSTask,
     name="scan-compliance-reports",