feat(attack-surfaces): add check ID list to response

Author: vicferpoy

api/src/backend/api/apps.py

@@ -40,6 +40,7 @@ def ready(self):
             self._ensure_crypto_keys()
 
         load_prowler_compliance()
+        self._initialize_attack_surface_mapping()
 
     def _ensure_crypto_keys(self):
         """
@@ -167,3 +168,13 @@ def _generate_jwt_keys(self):
                 f"Error generating JWT keys: {e}. Please set '{SIGNING_KEY_ENV}' and '{VERIFYING_KEY_ENV}' manually."
             )
             raise e
+
+    def _initialize_attack_surface_mapping(self):
+        from tasks.jobs.scan import (  # noqa: F401
+            _get_attack_surface_mapping_from_provider,
+        )
+
+        from api.models import Provider  # noqa: F401
+
+        for provider_type, _label in Provider.ProviderChoices.choices:
+            _get_attack_surface_mapping_from_provider(provider_type)

api/src/backend/api/specs/v1.yaml

@@ -4502,7 +4502,7 @@ paths:
       operationId: overviews_attack_surfaces_retrieve
       description: Retrieve aggregated attack surface metrics from latest completed
         scans per provider.
-      summary: Get Attack surface overview
+      summary: Get attack surface overview
       parameters:
       - in: query
         name: fields[attack-surface-overviews]
@@ -4515,30 +4515,31 @@ paths:
             - total_findings
             - failed_findings
             - muted_failed_findings
+            - check_ids
         description: endpoint return only specific fields in the response on a per-type
           basis by including a fields[TYPE] query parameter.
         explode: false
+      - in: query
+        name: filter[provider_id.in]
+        schema:
+          type: string
+        description: Filter by multiple provider IDs (comma-separated UUIDs)
       - in: query
         name: filter[provider_id]
         schema:
           type: string
           format: uuid
         description: Filter by specific provider ID
       - in: query
-        name: filter[provider_id__in]
+        name: filter[provider_type.in]
         schema:
           type: string
-        description: Filter by multiple provider IDs (comma-separated UUIDs)
+        description: Filter by multiple provider types (comma-separated)
       - in: query
         name: filter[provider_type]
         schema:
           type: string
         description: Filter by provider type (aws, azure, gcp, etc.)
-      - in: query
-        name: filter[provider_type__in]
-        schema:
-          type: string
-        description: Filter by multiple provider types (comma-separated)
       tags:
       - Overview
       security:
@@ -10697,6 +10698,11 @@ components:
               type: integer
             muted_failed_findings:
               type: integer
+            check_ids:
+              type: array
+              items:
+                type: string
+              readOnly: true
           required:
           - id
           - total_findings

api/src/backend/api/tests/test_views.py

@@ -6866,6 +6866,7 @@ def test_overview_attack_surface_no_data(self, authenticated_client):
             assert item["attributes"]["total_findings"] == 0
             assert item["attributes"]["failed_findings"] == 0
             assert item["attributes"]["muted_failed_findings"] == 0
+            assert item["attributes"]["check_ids"] == []
 
     def test_overview_attack_surface_with_data(
         self,
@@ -6877,6 +6878,13 @@ def test_overview_attack_surface_with_data(
         tenant = tenants_fixture[0]
         provider = providers_fixture[0]
 
+        mapping = {
+            "internet-exposed": {"aws-check-1", "aws-check-2"},
+            "secrets": {"aws-secret-check"},
+            "privilege-escalation": {"aws-priv-check"},
+            "ec2-imdsv1": {"aws-imdsv1-check"},
+        }
+
         scan = Scan.objects.create(
             name="attack-surface-scan",
             provider=provider,
@@ -6902,18 +6910,31 @@ def test_overview_attack_surface_with_data(
             muted_failed=2,
         )
 
-        response = authenticated_client.get(reverse("overview-attack-surface"))
+        with patch(
+            "api.v1.views._get_attack_surface_mapping_from_provider",
+            return_value=mapping,
+        ):
+            response = authenticated_client.get(reverse("overview-attack-surface"))
         assert response.status_code == status.HTTP_200_OK
         data = response.json()["data"]
         assert len(data) == 4
 
         results_by_type = {item["id"]: item["attributes"] for item in data}
         assert results_by_type["internet-exposed"]["total_findings"] == 20
         assert results_by_type["internet-exposed"]["failed_findings"] == 10
+        assert set(results_by_type["internet-exposed"]["check_ids"]) == {
+            "aws-check-1",
+            "aws-check-2",
+        }
         assert results_by_type["secrets"]["total_findings"] == 15
         assert results_by_type["secrets"]["failed_findings"] == 8
+        assert set(results_by_type["secrets"]["check_ids"]) == {"aws-secret-check"}
         assert results_by_type["privilege-escalation"]["total_findings"] == 0
+        assert set(results_by_type["privilege-escalation"]["check_ids"]) == {
+            "aws-priv-check"
+        }
         assert results_by_type["ec2-imdsv1"]["total_findings"] == 0
+        assert set(results_by_type["ec2-imdsv1"]["check_ids"]) == {"aws-imdsv1-check"}
 
     def test_overview_attack_surface_provider_filter(
         self,
@@ -6940,6 +6961,13 @@ def test_overview_attack_surface_provider_filter(
             tenant=tenant,
         )
 
+        mapping = {
+            "internet-exposed": {"shared-check", "shared-check"},
+            "secrets": set(),
+            "privilege-escalation": {"priv-check"},
+            "ec2-imdsv1": {"imdsv1-check"},
+        }
+
         create_attack_surface_overview(
             tenant,
             scan1,
@@ -6957,15 +6985,20 @@ def test_overview_attack_surface_provider_filter(
             muted_failed=3,
         )
 
-        response = authenticated_client.get(
-            reverse("overview-attack-surface"),
-            {"filter[provider_id]": str(provider1.id)},
-        )
+        with patch(
+            "api.v1.views._get_attack_surface_mapping_from_provider",
+            return_value=mapping,
+        ):
+            response = authenticated_client.get(
+                reverse("overview-attack-surface"),
+                {"filter[provider_id]": str(provider1.id)},
+            )
         assert response.status_code == status.HTTP_200_OK
         data = response.json()["data"]
         results_by_type = {item["id"]: item["attributes"] for item in data}
         assert results_by_type["internet-exposed"]["total_findings"] == 10
         assert results_by_type["internet-exposed"]["failed_findings"] == 5
+        assert results_by_type["internet-exposed"]["check_ids"] == ["shared-check"]
 
     def test_overview_services_region_filter(
         self, authenticated_client, scan_summaries_fixture

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

@@ -2226,6 +2226,9 @@ class AttackSurfaceOverviewSerializer(BaseSerializerV1):
     total_findings = serializers.IntegerField()
     failed_findings = serializers.IntegerField()
     muted_failed_findings = serializers.IntegerField()
+    check_ids = serializers.ListField(
+        child=serializers.CharField(), allow_empty=True, default=list, read_only=True
+    )
 
     class JSONAPIMeta:
         resource_name = "attack-surface-overviews"

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

@@ -74,6 +74,7 @@
 from rest_framework_simplejwt.exceptions import InvalidToken, TokenError
 from tasks.beat import schedule_provider_scan
 from tasks.jobs.export import get_s3_client
+from tasks.jobs.scan import _get_attack_surface_mapping_from_provider
 from tasks.tasks import (
     backfill_scan_resource_summaries_task,
     check_integration_connection_task,
@@ -3893,8 +3894,8 @@ def attributes(self, request):
         filters=True,
     ),
     attack_surface=extend_schema(
-        summary="Retrieve attack surface overview",
-        description="Returns aggregated attack surface metrics from latest completed scans per provider.",
+        summary="Get attack surface overview",
+        description="Retrieve aggregated attack surface metrics from latest completed scans per provider.",
         tags=["Overview"],
         parameters=[
             OpenApiParameter(
@@ -4056,6 +4057,19 @@ def _latest_scan_ids_for_allowed_providers(self, tenant_id):
             .values_list("id", flat=True)
         )
 
+    def _attack_surface_check_ids_by_provider_types(self, provider_types):
+        check_ids_by_type = {
+            attack_surface_type: set()
+            for attack_surface_type in AttackSurfaceOverview.AttackSurfaceTypeChoices.values
+        }
+        for provider_type in provider_types:
+            attack_surface_mapping = _get_attack_surface_mapping_from_provider(
+                provider_type=provider_type
+            )
+            for attack_surface_type, check_ids in attack_surface_mapping.items():
+                check_ids_by_type[attack_surface_type].update(check_ids)
+        return check_ids_by_type
+
     @action(detail=False, methods=["get"], url_name="providers")
     def providers(self, request):
         tenant_id = self.request.tenant_id
@@ -4566,7 +4580,14 @@ def attack_surface(self, request):
         filtered_queryset = self._apply_filterset(
             base_queryset, AttackSurfaceOverviewFilter
         )
-
+        provider_types = list(
+            filtered_queryset.values_list(
+                "scan__provider__provider", flat=True
+            ).distinct()
+        )
+        attack_surface_check_ids = self._attack_surface_check_ids_by_provider_types(
+            provider_types
+        )
         # Aggregate attack surface data
         aggregation = filtered_queryset.values("attack_surface_type").annotate(
             total_findings=Coalesce(Sum("total_findings"), 0),
@@ -4590,7 +4611,12 @@ def attack_surface(self, request):
             }
 
         response_data = [
-            {"attack_surface_type": key, **value} for key, value in results.items()
+            {
+                "attack_surface_type": key,
+                **value,
+                "check_ids": attack_surface_check_ids.get(key, []),
+            }
+            for key, value in results.items()
         ]
 
         return Response(