feat(api): new endpoint for categories overviews (#9529)

api/src/backend/api/apps.py

@@ -40,7 +40,6 @@ class ApiConfig(AppConfig):
             self._ensure_crypto_keys()
 
         load_prowler_compliance()
-        self._initialize_attack_surface_mapping()
 
     def _ensure_crypto_keys(self):
         """
@@ -168,13 +167,3 @@ class ApiConfig(AppConfig):
                 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/filters.py

@@ -43,6 +43,7 @@ from api.models import (
     ResourceTag,
     Role,
     Scan,
+    ScanCategorySummary,
     ScanSummary,
     SeverityChoices,
     StateChoices,
@@ -157,6 +158,9 @@ class CommonFindingFilters(FilterSet):
         field_name="resources__type", lookup_expr="icontains"
     )
 
+    category = CharFilter(method="filter_category")
+    category__in = CharInFilter(field_name="categories", lookup_expr="overlap")
+
     # Temporarily disabled until we implement tag filtering in the UI
     # resource_tag_key = CharFilter(field_name="resources__tags__key")
     # resource_tag_key__in = CharInFilter(
@@ -188,6 +192,9 @@ class CommonFindingFilters(FilterSet):
     def filter_resource_type(self, queryset, name, value):
         return queryset.filter(resource_types__contains=[value])
 
+    def filter_category(self, queryset, name, value):
+        return queryset.filter(categories__contains=[value])
+
     def filter_resource_tag(self, queryset, name, value):
         overall_query = Q()
         for key_value_pair in value:
@@ -1096,3 +1103,22 @@ class AttackSurfaceOverviewFilter(FilterSet):
     class Meta:
         model = AttackSurfaceOverview
         fields = {}
+
+
+class CategoryOverviewFilter(FilterSet):
+    provider_id = UUIDFilter(field_name="scan__provider__id", lookup_expr="exact")
+    provider_id__in = UUIDInFilter(field_name="scan__provider__id", lookup_expr="in")
+    provider_type = ChoiceFilter(
+        field_name="scan__provider__provider", choices=Provider.ProviderChoices.choices
+    )
+    provider_type__in = ChoiceInFilter(
+        field_name="scan__provider__provider",
+        choices=Provider.ProviderChoices.choices,
+        lookup_expr="in",
+    )
+    category = CharFilter(field_name="category", lookup_expr="exact")
+    category__in = CharInFilter(field_name="category", lookup_expr="in")
+
+    class Meta:
+        model = ScanCategorySummary
+        fields = {}

api/src/backend/api/migrations/0063_scan_category_summary.py

@@ -0,0 +1,108 @@
+import uuid
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+import api.db_utils
+import api.rls
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0062_backfill_daily_severity_summaries"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="ScanCategorySummary",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                (
+                    "tenant_id",
+                    models.UUIDField(db_index=True, editable=False),
+                ),
+                (
+                    "inserted_at",
+                    models.DateTimeField(auto_now_add=True),
+                ),
+                (
+                    "scan",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="category_summaries",
+                        related_query_name="category_summary",
+                        to="api.scan",
+                    ),
+                ),
+                (
+                    "category",
+                    models.CharField(max_length=100),
+                ),
+                (
+                    "severity",
+                    api.db_utils.SeverityEnumField(
+                        choices=[
+                            ("critical", "Critical"),
+                            ("high", "High"),
+                            ("medium", "Medium"),
+                            ("low", "Low"),
+                            ("informational", "Informational"),
+                        ],
+                        max_length=50,
+                    ),
+                ),
+                (
+                    "total_findings",
+                    models.IntegerField(
+                        default=0, help_text="Non-muted findings (PASS + FAIL)"
+                    ),
+                ),
+                (
+                    "failed_findings",
+                    models.IntegerField(
+                        default=0,
+                        help_text="Non-muted FAIL findings (subset of total_findings)",
+                    ),
+                ),
+                (
+                    "new_failed_findings",
+                    models.IntegerField(
+                        default=0,
+                        help_text="Non-muted FAIL with delta='new' (subset of failed_findings)",
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "scan_category_summaries",
+            },
+        ),
+        migrations.AddIndex(
+            model_name="scancategorysummary",
+            index=models.Index(
+                fields=["tenant_id", "scan"], name="scs_tenant_scan_idx"
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="scancategorysummary",
+            constraint=models.UniqueConstraint(
+                fields=("tenant_id", "scan_id", "category", "severity"),
+                name="unique_category_severity_per_scan",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="scancategorysummary",
+            constraint=api.rls.RowLevelSecurityConstraint(
+                field="tenant_id",
+                name="rls_on_scancategorysummary",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ),
+    ]

api/src/backend/api/migrations/0064_finding_categories.py

@@ -0,0 +1,21 @@
+import django.contrib.postgres.fields
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0063_scan_category_summary"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="finding",
+            name="categories",
+            field=django.contrib.postgres.fields.ArrayField(
+                base_field=models.CharField(max_length=100),
+                blank=True,
+                null=True,
+                size=None,
+            ),
+        ),
+    ]

api/src/backend/api/models.py

@@ -868,6 +868,14 @@ class Finding(PostgresPartitionedModel, RowLevelSecurityProtectedModel):
         null=True,
     )
 
+    # Check metadata denormalization
+    categories = ArrayField(
+        models.CharField(max_length=100),
+        blank=True,
+        null=True,
+        help_text="Categories from check metadata for efficient filtering",
+    )
+
     # Relationships
     scan = models.ForeignKey(to=Scan, related_name="findings", on_delete=models.CASCADE)
 
@@ -1951,6 +1959,64 @@ class ResourceScanSummary(RowLevelSecurityProtectedModel):
         ]
 
 
+class ScanCategorySummary(RowLevelSecurityProtectedModel):
+    """
+    Pre-aggregated category metrics per scan by severity.
+
+    Stores one row per (category, severity) combination per scan for efficient
+    overview queries. Categories come from check_metadata.categories.
+
+    Count relationships (each is a subset of the previous):
+        - total_findings >= failed_findings >= new_failed_findings
+    """
+
+    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
+    inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
+
+    scan = models.ForeignKey(
+        Scan,
+        on_delete=models.CASCADE,
+        related_name="category_summaries",
+        related_query_name="category_summary",
+    )
+
+    category = models.CharField(max_length=100)
+    severity = SeverityEnumField(choices=SeverityChoices)
+
+    total_findings = models.IntegerField(
+        default=0, help_text="Non-muted findings (PASS + FAIL)"
+    )
+    failed_findings = models.IntegerField(
+        default=0, help_text="Non-muted FAIL findings (subset of total_findings)"
+    )
+    new_failed_findings = models.IntegerField(
+        default=0,
+        help_text="Non-muted FAIL with delta='new' (subset of failed_findings)",
+    )
+
+    class Meta(RowLevelSecurityProtectedModel.Meta):
+        db_table = "scan_category_summaries"
+
+        indexes = [
+            models.Index(fields=["tenant_id", "scan"], name="scs_tenant_scan_idx"),
+        ]
+
+        constraints = [
+            models.UniqueConstraint(
+                fields=("tenant_id", "scan_id", "category", "severity"),
+                name="unique_category_severity_per_scan",
+            ),
+            RowLevelSecurityConstraint(
+                field="tenant_id",
+                name="rls_on_%(class)s",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ]
+
+    class JSONAPIMeta:
+        resource_name = "scan-category-summaries"
+
+
 class LighthouseConfiguration(RowLevelSecurityProtectedModel):
     """
     Stores configuration and API keys for LLM services.

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

@@ -711,6 +711,7 @@ paths:
             - severity
             - check_id
             - check_metadata
+            - categories
             - raw_result
             - inserted_at
             - updated_at
@@ -723,6 +724,19 @@ paths:
         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[category]
+        schema:
+          type: string
+      - in: query
+        name: filter[category__in]
+        schema:
+          type: array
+          items:
+            type: string
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
       - in: query
         name: filter[check_id]
         schema:
@@ -1205,6 +1219,7 @@ paths:
             - severity
             - check_id
             - check_metadata
+            - categories
             - raw_result
             - inserted_at
             - updated_at
@@ -1265,6 +1280,19 @@ paths:
         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[category]
+        schema:
+          type: string
+      - in: query
+        name: filter[category__in]
+        schema:
+          type: array
+          items:
+            type: string
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
       - in: query
         name: filter[check_id]
         schema:
@@ -1722,6 +1750,7 @@ paths:
             - severity
             - check_id
             - check_metadata
+            - categories
             - raw_result
             - inserted_at
             - updated_at
@@ -1734,6 +1763,19 @@ paths:
         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[category]
+        schema:
+          type: string
+      - in: query
+        name: filter[category__in]
+        schema:
+          type: array
+          items:
+            type: string
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
       - in: query
         name: filter[check_id]
         schema:
@@ -2151,9 +2193,23 @@ paths:
             - services
             - regions
             - resource_types
+            - categories
         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[category]
+        schema:
+          type: string
+      - in: query
+        name: filter[category__in]
+        schema:
+          type: array
+          items:
+            type: string
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
       - in: query
         name: filter[check_id]
         schema:
@@ -2609,9 +2665,23 @@ paths:
             - services
             - regions
             - resource_types
+            - categories
         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[category]
+        schema:
+          type: string
+      - in: query
+        name: filter[category__in]
+        schema:
+          type: array
+          items:
+            type: string
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
       - in: query
         name: filter[check_id]
         schema:
@@ -4499,7 +4569,7 @@ paths:
           description: No response body
   /api/v1/overviews/attack-surfaces:
     get:
-      operationId: overviews_attack_surfaces_retrieve
+      operationId: overviews_attack_surfaces_list
       description: Retrieve aggregated attack surface metrics from latest completed
         scans per provider.
       summary: Get attack surface overview
@@ -4515,31 +4585,116 @@ 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_type.in]
+        name: filter[provider_id__in]
         schema:
-          type: string
-        description: Filter by multiple provider types (comma-separated)
+          type: array
+          items:
+            type: string
+            format: uuid
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
       - in: query
         name: filter[provider_type]
         schema:
           type: string
-        description: Filter by provider type (aws, azure, gcp, etc.)
+          x-spec-enum-id: eca8c51e6bd28935
+          enum:
+          - aws
+          - azure
+          - gcp
+          - github
+          - iac
+          - kubernetes
+          - m365
+          - mongodbatlas
+          - oraclecloud
+        description: |-
+          * `aws` - AWS
+          * `azure` - Azure
+          * `gcp` - GCP
+          * `kubernetes` - Kubernetes
+          * `m365` - M365
+          * `github` - GitHub
+          * `mongodbatlas` - MongoDB Atlas
+          * `iac` - IaC
+          * `oraclecloud` - Oracle Cloud Infrastructure
+      - in: query
+        name: filter[provider_type__in]
+        schema:
+          type: array
+          items:
+            type: string
+            x-spec-enum-id: eca8c51e6bd28935
+            enum:
+            - aws
+            - azure
+            - gcp
+            - github
+            - iac
+            - kubernetes
+            - m365
+            - mongodbatlas
+            - oraclecloud
+        description: |-
+          Multiple values may be separated by commas.
+
+          * `aws` - AWS
+          * `azure` - Azure
+          * `gcp` - GCP
+          * `kubernetes` - Kubernetes
+          * `m365` - M365
+          * `github` - GitHub
+          * `mongodbatlas` - MongoDB Atlas
+          * `iac` - IaC
+          * `oraclecloud` - Oracle Cloud Infrastructure
+        explode: false
+        style: form
+      - name: filter[search]
+        required: false
+        in: query
+        description: A search term.
+        schema:
+          type: string
+      - name: page[number]
+        required: false
+        in: query
+        description: A page number within the paginated result set.
+        schema:
+          type: integer
+      - name: page[size]
+        required: false
+        in: query
+        description: Number of results to return per page.
+        schema:
+          type: integer
+      - name: sort
+        required: false
+        in: query
+        description: '[list of fields to sort by](https://jsonapi.org/format/#fetching-sorting)'
+        schema:
+          type: array
+          items:
+            type: string
+            enum:
+            - id
+            - -id
+            - total_findings
+            - -total_findings
+            - failed_findings
+            - -failed_findings
+            - muted_failed_findings
+            - -muted_failed_findings
+        explode: false
       tags:
       - Overview
       security:
@@ -4549,7 +4704,164 @@ paths:
           content:
             application/vnd.api+json:
               schema:
-                $ref: '#/components/schemas/AttackSurfaceOverviewResponse'
+                $ref: '#/components/schemas/PaginatedAttackSurfaceOverviewList'
+          description: ''
+  /api/v1/overviews/categories:
+    get:
+      operationId: overviews_categories_list
+      description: 'Retrieve aggregated category metrics from latest completed scans
+        per provider. Returns one row per category with total, failed, and new failed
+        findings counts, plus a severity breakdown showing failed findings per severity
+        level. '
+      summary: Get category overview
+      parameters:
+      - in: query
+        name: fields[category-overviews]
+        schema:
+          type: array
+          items:
+            type: string
+            enum:
+            - id
+            - total_findings
+            - failed_findings
+            - new_failed_findings
+            - severity
+        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[category]
+        schema:
+          type: string
+      - in: query
+        name: filter[category__in]
+        schema:
+          type: array
+          items:
+            type: string
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
+      - in: query
+        name: filter[provider_id]
+        schema:
+          type: string
+          format: uuid
+      - in: query
+        name: filter[provider_id__in]
+        schema:
+          type: array
+          items:
+            type: string
+            format: uuid
+        description: Multiple values may be separated by commas.
+        explode: false
+        style: form
+      - in: query
+        name: filter[provider_type]
+        schema:
+          type: string
+          x-spec-enum-id: eca8c51e6bd28935
+          enum:
+          - aws
+          - azure
+          - gcp
+          - github
+          - iac
+          - kubernetes
+          - m365
+          - mongodbatlas
+          - oraclecloud
+        description: |-
+          * `aws` - AWS
+          * `azure` - Azure
+          * `gcp` - GCP
+          * `kubernetes` - Kubernetes
+          * `m365` - M365
+          * `github` - GitHub
+          * `mongodbatlas` - MongoDB Atlas
+          * `iac` - IaC
+          * `oraclecloud` - Oracle Cloud Infrastructure
+      - in: query
+        name: filter[provider_type__in]
+        schema:
+          type: array
+          items:
+            type: string
+            x-spec-enum-id: eca8c51e6bd28935
+            enum:
+            - aws
+            - azure
+            - gcp
+            - github
+            - iac
+            - kubernetes
+            - m365
+            - mongodbatlas
+            - oraclecloud
+        description: |-
+          Multiple values may be separated by commas.
+
+          * `aws` - AWS
+          * `azure` - Azure
+          * `gcp` - GCP
+          * `kubernetes` - Kubernetes
+          * `m365` - M365
+          * `github` - GitHub
+          * `mongodbatlas` - MongoDB Atlas
+          * `iac` - IaC
+          * `oraclecloud` - Oracle Cloud Infrastructure
+        explode: false
+        style: form
+      - name: filter[search]
+        required: false
+        in: query
+        description: A search term.
+        schema:
+          type: string
+      - name: page[number]
+        required: false
+        in: query
+        description: A page number within the paginated result set.
+        schema:
+          type: integer
+      - name: page[size]
+        required: false
+        in: query
+        description: Number of results to return per page.
+        schema:
+          type: integer
+      - name: sort
+        required: false
+        in: query
+        description: '[list of fields to sort by](https://jsonapi.org/format/#fetching-sorting)'
+        schema:
+          type: array
+          items:
+            type: string
+            enum:
+            - id
+            - -id
+            - total_findings
+            - -total_findings
+            - failed_findings
+            - -failed_findings
+            - new_failed_findings
+            - -new_failed_findings
+            - severity
+            - -severity
+        explode: false
+      tags:
+      - Overview
+      security:
+      - JWT or API Key: []
+      responses:
+        '200':
+          content:
+            application/vnd.api+json:
+              schema:
+                $ref: '#/components/schemas/PaginatedCategoryOverviewList'
           description: ''
   /api/v1/overviews/findings:
     get:
@@ -4951,7 +5263,7 @@ paths:
       summary: Get findings severity data over time
       parameters:
       - in: query
-        name: fields[findings-severity-timeseries]
+        name: fields[findings-severity-over-time]
         schema:
           type: array
           items:
@@ -4972,10 +5284,12 @@ paths:
         name: filter[date_from]
         schema:
           type: string
+          format: date
       - in: query
         name: filter[date_to]
         schema:
           type: string
+          format: date
       - in: query
         name: filter[provider_id]
         schema:
@@ -10846,23 +11160,46 @@ components:
               type: integer
             muted_failed_findings:
               type: integer
-            check_ids:
-              type: array
-              items:
-                type: string
-              readOnly: true
           required:
           - id
           - total_findings
           - failed_findings
           - muted_failed_findings
-    AttackSurfaceOverviewResponse:
+    CategoryOverview:
       type: object
-      properties:
-        data:
-          $ref: '#/components/schemas/AttackSurfaceOverview'
       required:
-      - data
+      - type
+      - id
+      additionalProperties: false
+      properties:
+        type:
+          type: string
+          description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
+            member is used to describe resource objects that share common attributes
+            and relationships.
+          enum:
+          - category-overviews
+        id: {}
+        attributes:
+          type: object
+          properties:
+            id:
+              type: string
+            total_findings:
+              type: integer
+            failed_findings:
+              type: integer
+            new_failed_findings:
+              type: integer
+            severity:
+              description: 'Severity breakdown: {informational, low, medium, high,
+                critical}'
+          required:
+          - id
+          - total_findings
+          - failed_findings
+          - new_failed_findings
+          - severity
     ComplianceOverview:
       type: object
       required:
@@ -11079,6 +11416,13 @@ components:
               type: string
               maxLength: 100
             check_metadata: {}
+            categories:
+              type: array
+              items:
+                type: string
+                maxLength: 100
+              nullable: true
+              description: Categories from check metadata for efficient filtering
             raw_result: {}
             inserted_at:
               type: string
@@ -11229,10 +11573,15 @@ components:
               type: array
               items:
                 type: string
+            categories:
+              type: array
+              items:
+                type: string
           required:
           - services
           - regions
           - resource_types
+          - categories
     FindingMetadataResponse:
       type: object
       properties:
@@ -13935,6 +14284,24 @@ components:
           $ref: '#/components/schemas/OverviewSeverity'
       required:
       - data
+    PaginatedAttackSurfaceOverviewList:
+      type: object
+      properties:
+        data:
+          type: array
+          items:
+            $ref: '#/components/schemas/AttackSurfaceOverview'
+      required:
+      - data
+    PaginatedCategoryOverviewList:
+      type: object
+      properties:
+        data:
+          type: array
+          items:
+            $ref: '#/components/schemas/CategoryOverview'
+      required:
+      - data
     PaginatedComplianceOverviewAttributesList:
       type: object
       properties:

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

@@ -4267,6 +4267,74 @@ class TestFindingViewSet:
         assert attributes["regions"] == latest_scan_finding.resource_regions
         assert attributes["resource_types"] == latest_scan_finding.resource_types
 
+    def test_findings_metadata_categories(
+        self, authenticated_client, findings_with_categories
+    ):
+        finding = findings_with_categories
+        response = authenticated_client.get(
+            reverse("finding-metadata"),
+            {"filter[inserted_at]": finding.inserted_at.strftime("%Y-%m-%d")},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        attributes = response.json()["data"]["attributes"]
+        assert set(attributes["categories"]) == {"gen-ai", "security"}
+
+    def test_findings_metadata_latest_categories(
+        self, authenticated_client, latest_scan_finding_with_categories
+    ):
+        response = authenticated_client.get(
+            reverse("finding-metadata_latest"),
+        )
+        assert response.status_code == status.HTTP_200_OK
+        attributes = response.json()["data"]["attributes"]
+        assert set(attributes["categories"]) == {"gen-ai", "iam"}
+
+    def test_findings_filter_by_category(
+        self, authenticated_client, findings_with_categories
+    ):
+        finding = findings_with_categories
+        response = authenticated_client.get(
+            reverse("finding-list"),
+            {
+                "filter[category]": "gen-ai",
+                "filter[inserted_at]": finding.inserted_at.strftime("%Y-%m-%d"),
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == 1
+        assert set(response.json()["data"][0]["attributes"]["categories"]) == {
+            "gen-ai",
+            "security",
+        }
+
+    def test_findings_filter_by_category_in(
+        self, authenticated_client, findings_with_multiple_categories
+    ):
+        finding1, _ = findings_with_multiple_categories
+        response = authenticated_client.get(
+            reverse("finding-list"),
+            {
+                "filter[category__in]": "gen-ai,iam",
+                "filter[inserted_at]": finding1.inserted_at.strftime("%Y-%m-%d"),
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == 2
+
+    def test_findings_filter_by_category_no_match(
+        self, authenticated_client, findings_with_categories
+    ):
+        finding = findings_with_categories
+        response = authenticated_client.get(
+            reverse("finding-list"),
+            {
+                "filter[category]": "nonexistent",
+                "filter[inserted_at]": finding.inserted_at.strftime("%Y-%m-%d"),
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == 0
+
 
 @pytest.mark.django_db
 class TestJWTFields:
@@ -7258,7 +7326,6 @@ class TestOverviewViewSet:
             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,
@@ -7270,13 +7337,6 @@ class TestOverviewViewSet:
         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,
@@ -7302,11 +7362,7 @@ class TestOverviewViewSet:
             muted_failed=2,
         )
 
-        with patch(
-            "api.v1.views._get_attack_surface_mapping_from_provider",
-            return_value=mapping,
-        ):
-            response = authenticated_client.get(reverse("overview-attack-surface"))
+        response = authenticated_client.get(reverse("overview-attack-surface"))
         assert response.status_code == status.HTTP_200_OK
         data = response.json()["data"]
         assert len(data) == 4
@@ -7314,19 +7370,10 @@ class TestOverviewViewSet:
         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,
@@ -7353,13 +7400,6 @@ class TestOverviewViewSet:
             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,
@@ -7377,20 +7417,15 @@ class TestOverviewViewSet:
             muted_failed=3,
         )
 
-        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)},
-            )
+        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
@@ -7633,6 +7668,219 @@ class TestOverviewViewSet:
         assert len(data) == 1
         assert data[0]["attributes"]["overall_score"] == "80.00"
 
+    def test_overview_categories_no_data(self, authenticated_client):
+        response = authenticated_client.get(reverse("overview-categories"))
+        assert response.status_code == status.HTTP_200_OK
+        assert response.json()["data"] == []
+
+    def test_overview_categories_aggregates_by_category_with_severity(
+        self,
+        authenticated_client,
+        tenants_fixture,
+        providers_fixture,
+        create_scan_category_summary,
+    ):
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+
+        scan = Scan.objects.create(
+            name="categories-scan",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+        )
+
+        create_scan_category_summary(
+            tenant,
+            scan,
+            "iam",
+            "high",
+            total_findings=20,
+            failed_findings=10,
+            new_failed_findings=5,
+        )
+        create_scan_category_summary(
+            tenant,
+            scan,
+            "iam",
+            "medium",
+            total_findings=15,
+            failed_findings=8,
+            new_failed_findings=3,
+        )
+        create_scan_category_summary(
+            tenant,
+            scan,
+            "encryption",
+            "critical",
+            total_findings=5,
+            failed_findings=2,
+            new_failed_findings=1,
+        )
+
+        response = authenticated_client.get(reverse("overview-categories"))
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 2
+
+        results_by_category = {item["id"]: item["attributes"] for item in data}
+
+        assert results_by_category["iam"]["total_findings"] == 35
+        assert results_by_category["iam"]["failed_findings"] == 18
+        assert results_by_category["iam"]["new_failed_findings"] == 8
+        assert results_by_category["iam"]["severity"]["high"] == 10
+        assert results_by_category["iam"]["severity"]["medium"] == 8
+        assert results_by_category["iam"]["severity"]["critical"] == 0
+
+        assert results_by_category["encryption"]["total_findings"] == 5
+        assert results_by_category["encryption"]["failed_findings"] == 2
+        assert results_by_category["encryption"]["severity"]["critical"] == 2
+
+    @pytest.mark.parametrize(
+        "filter_key,filter_value_fn,expected_total,expected_failed",
+        [
+            ("filter[provider_id]", lambda p1, _: str(p1.id), 10, 5),
+            ("filter[provider_type]", lambda *_: "aws", 10, 5),
+            ("filter[provider_type__in]", lambda *_: "aws,gcp", 30, 20),
+        ],
+    )
+    def test_overview_categories_filters(
+        self,
+        authenticated_client,
+        tenants_fixture,
+        providers_fixture,
+        create_scan_category_summary,
+        filter_key,
+        filter_value_fn,
+        expected_total,
+        expected_failed,
+    ):
+        tenant = tenants_fixture[0]
+        provider1, _, gcp_provider, *_ = providers_fixture
+
+        scan1 = Scan.objects.create(
+            name="categories-scan-1",
+            provider=provider1,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+        )
+        scan2 = Scan.objects.create(
+            name="categories-scan-2",
+            provider=gcp_provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+        )
+
+        create_scan_category_summary(
+            tenant, scan1, "iam", "high", total_findings=10, failed_findings=5
+        )
+        create_scan_category_summary(
+            tenant, scan2, "iam", "high", total_findings=20, failed_findings=15
+        )
+
+        response = authenticated_client.get(
+            reverse("overview-categories"),
+            {filter_key: filter_value_fn(provider1, gcp_provider)},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+        assert data[0]["attributes"]["total_findings"] == expected_total
+        assert data[0]["attributes"]["failed_findings"] == expected_failed
+
+    def test_overview_categories_category_filter(
+        self,
+        authenticated_client,
+        tenants_fixture,
+        providers_fixture,
+        create_scan_category_summary,
+    ):
+        tenant = tenants_fixture[0]
+        provider = providers_fixture[0]
+
+        scan = Scan.objects.create(
+            name="category-filter-scan",
+            provider=provider,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+        )
+
+        create_scan_category_summary(
+            tenant, scan, "iam", "high", total_findings=10, failed_findings=5
+        )
+        create_scan_category_summary(
+            tenant, scan, "encryption", "medium", total_findings=20, failed_findings=8
+        )
+        create_scan_category_summary(
+            tenant, scan, "logging", "low", total_findings=15, failed_findings=3
+        )
+
+        response = authenticated_client.get(
+            reverse("overview-categories"),
+            {"filter[category__in]": "iam,encryption"},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        category_ids = {item["id"] for item in data}
+        assert category_ids == {"iam", "encryption"}
+
+    def test_overview_categories_aggregates_multiple_providers(
+        self,
+        authenticated_client,
+        tenants_fixture,
+        providers_fixture,
+        create_scan_category_summary,
+    ):
+        tenant = tenants_fixture[0]
+        provider1, provider2, *_ = providers_fixture
+
+        scan1 = Scan.objects.create(
+            name="multi-provider-scan-1",
+            provider=provider1,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+        )
+        scan2 = Scan.objects.create(
+            name="multi-provider-scan-2",
+            provider=provider2,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+        )
+
+        create_scan_category_summary(
+            tenant,
+            scan1,
+            "iam",
+            "high",
+            total_findings=10,
+            failed_findings=5,
+            new_failed_findings=2,
+        )
+        create_scan_category_summary(
+            tenant,
+            scan2,
+            "iam",
+            "high",
+            total_findings=15,
+            failed_findings=8,
+            new_failed_findings=3,
+        )
+
+        response = authenticated_client.get(reverse("overview-categories"))
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+        assert data[0]["id"] == "iam"
+        assert data[0]["attributes"]["total_findings"] == 25
+        assert data[0]["attributes"]["failed_findings"] == 13
+        assert data[0]["attributes"]["new_failed_findings"] == 5
+
 
 @pytest.mark.django_db
 class TestScheduleViewSet:

api/src/backend/api/utils.py

@@ -382,10 +382,18 @@ def get_findings_metadata_no_aggregations(tenant_id: str, filtered_queryset):
     regions = sorted({region for region in aggregation["regions"] or [] if region})
     resource_types = sorted(set(aggregation["resource_types"] or []))
 
+    # Aggregate categories from findings
+    categories_set = set()
+    for categories_list in filtered_queryset.values_list("categories", flat=True):
+        if categories_list:
+            categories_set.update(categories_list)
+    categories = sorted(categories_set)
+
     result = {
         "services": services,
         "regions": regions,
         "resource_types": resource_types,
+        "categories": categories,
     }
 
     serializer = FindingMetadataSerializer(data=result)

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

@@ -1301,6 +1301,7 @@ class FindingSerializer(RLSSerializer):
             "severity",
             "check_id",
             "check_metadata",
+            "categories",
             "raw_result",
             "inserted_at",
             "updated_at",
@@ -1356,6 +1357,7 @@ class FindingMetadataSerializer(BaseSerializerV1):
     resource_types = serializers.ListField(
         child=serializers.CharField(), allow_empty=True
     )
+    categories = serializers.ListField(child=serializers.CharField(), allow_empty=True)
     # Temporarily disabled until we implement tag filtering in the UI
     # tags = serializers.JSONField(help_text="Tags are described as key-value pairs.")
 
@@ -2242,14 +2244,26 @@ 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"
 
 
+class CategoryOverviewSerializer(BaseSerializerV1):
+    """Serializer for category overview aggregations."""
+
+    id = serializers.CharField(source="category")
+    total_findings = serializers.IntegerField()
+    failed_findings = serializers.IntegerField()
+    new_failed_findings = serializers.IntegerField()
+    severity = serializers.JSONField(
+        help_text="Severity breakdown: {informational, low, medium, high, critical}"
+    )
+
+    class JSONAPIMeta:
+        resource_name = "category-overviews"
+
+
 class OverviewRegionSerializer(serializers.Serializer):
     id = serializers.SerializerMethodField()
     provider_type = serializers.CharField()

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

@@ -74,7 +74,6 @@ from rest_framework_json_api.views import RelationshipView, Response
 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_compliance_summaries_task,
     backfill_scan_resource_summaries_task,
@@ -100,6 +99,7 @@ from api.db_utils import rls_transaction
 from api.exceptions import TaskFailedException
 from api.filters import (
     AttackSurfaceOverviewFilter,
+    CategoryOverviewFilter,
     ComplianceOverviewFilter,
     CustomDjangoFilterBackend,
     DailySeveritySummaryFilter,
@@ -157,6 +157,7 @@ from api.models import (
     SAMLDomainIndex,
     SAMLToken,
     Scan,
+    ScanCategorySummary,
     ScanSummary,
     SeverityChoices,
     StateChoices,
@@ -178,6 +179,7 @@ from api.uuid_utils import datetime_to_uuid7, uuid7_start
 from api.v1.mixins import DisablePaginationMixin, PaginateByPkMixin, TaskManagementMixin
 from api.v1.serializers import (
     AttackSurfaceOverviewSerializer,
+    CategoryOverviewSerializer,
     ComplianceOverviewAttributesSerializer,
     ComplianceOverviewDetailSerializer,
     ComplianceOverviewDetailThreatscoreSerializer,
@@ -2760,12 +2762,15 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
 
         queryset = ResourceScanSummary.objects.filter(tenant_id=tenant_id)
         scan_based_filters = {}
+        category_scan_filters = {}  # Filters for ScanCategorySummary
 
         if scans := query_params.get("filter[scan__in]") or query_params.get(
             "filter[scan]"
         ):
-            queryset = queryset.filter(scan_id__in=scans.split(","))
-            scan_based_filters = {"id__in": scans.split(",")}
+            scan_ids_list = scans.split(",")
+            queryset = queryset.filter(scan_id__in=scan_ids_list)
+            scan_based_filters = {"id__in": scan_ids_list}
+            category_scan_filters = {"scan_id__in": scan_ids_list}
         else:
             exact = query_params.get("filter[inserted_at]")
             gte = query_params.get("filter[inserted_at__gte]")
@@ -2809,6 +2814,7 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
                 scan_based_filters = {
                     key.lstrip("scan_"): value for key, value in date_filters.items()
                 }
+                category_scan_filters = date_filters
 
         # ToRemove: Temporary fallback mechanism
         if not queryset.exists():
@@ -2855,10 +2861,31 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
             .order_by("resource_type")
         )
 
+        # Get categories from ScanCategorySummary using same scan filters
+        categories = list(
+            ScanCategorySummary.objects.filter(
+                tenant_id=tenant_id, **category_scan_filters
+            )
+            .values_list("category", flat=True)
+            .distinct()
+            .order_by("category")
+        )
+
+        # Fallback to finding aggregation if no ScanCategorySummary exists
+        if not categories:
+            categories_set = set()
+            for categories_list in filtered_queryset.values_list(
+                "categories", flat=True
+            ):
+                if categories_list:
+                    categories_set.update(categories_list)
+            categories = sorted(categories_set)
+
         result = {
             "services": services,
             "regions": regions,
             "resource_types": resource_types,
+            "categories": categories,
         }
 
         serializer = self.get_serializer(data=result)
@@ -2963,10 +2990,36 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
             .order_by("resource_type")
         )
 
+        # Get categories from ScanCategorySummary for latest scans
+        categories = list(
+            ScanCategorySummary.objects.filter(
+                tenant_id=tenant_id,
+                scan_id__in=latest_scans_queryset.values_list("id", flat=True),
+            )
+            .values_list("category", flat=True)
+            .distinct()
+            .order_by("category")
+        )
+
+        # Fallback to finding aggregation if no ScanCategorySummary exists
+        if not categories:
+            filtered_queryset = self.filter_queryset(self.get_queryset()).filter(
+                tenant_id=tenant_id,
+                scan_id__in=latest_scans_queryset.values_list("id", flat=True),
+            )
+            categories_set = set()
+            for categories_list in filtered_queryset.values_list(
+                "categories", flat=True
+            ):
+                if categories_list:
+                    categories_set.update(categories_list)
+            categories = sorted(categories_set)
+
         result = {
             "services": services,
             "regions": regions,
             "resource_types": resource_types,
+            "categories": categories,
         }
 
         serializer = self.get_serializer(data=result)
@@ -4026,32 +4079,19 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
         summary="Get attack surface overview",
         description="Retrieve aggregated attack surface metrics from latest completed scans per provider.",
         tags=["Overview"],
-        parameters=[
-            OpenApiParameter(
-                name="filter[provider_id]",
-                type=OpenApiTypes.UUID,
-                location=OpenApiParameter.QUERY,
-                description="Filter by specific provider ID",
-            ),
-            OpenApiParameter(
-                name="filter[provider_id.in]",
-                type=OpenApiTypes.STR,
-                location=OpenApiParameter.QUERY,
-                description="Filter by multiple provider IDs (comma-separated UUIDs)",
-            ),
-            OpenApiParameter(
-                name="filter[provider_type]",
-                type=OpenApiTypes.STR,
-                location=OpenApiParameter.QUERY,
-                description="Filter by provider type (aws, azure, gcp, etc.)",
-            ),
-            OpenApiParameter(
-                name="filter[provider_type.in]",
-                type=OpenApiTypes.STR,
-                location=OpenApiParameter.QUERY,
-                description="Filter by multiple provider types (comma-separated)",
-            ),
-        ],
+        filters=True,
+        responses={200: AttackSurfaceOverviewSerializer(many=True)},
+    ),
+    categories=extend_schema(
+        summary="Get category overview",
+        description=(
+            "Retrieve aggregated category metrics from latest completed scans per provider. "
+            "Returns one row per category with total, failed, and new failed findings counts, "
+            "plus a severity breakdown showing failed findings per severity level. "
+        ),
+        tags=["Overview"],
+        filters=True,
+        responses={200: CategoryOverviewSerializer(many=True)},
     ),
 )
 @method_decorator(CACHE_DECORATOR, name="list")
@@ -4100,6 +4140,8 @@ class OverviewViewSet(BaseRLSViewSet):
             return ThreatScoreSnapshotSerializer
         elif self.action == "attack_surface":
             return AttackSurfaceOverviewSerializer
+        elif self.action == "categories":
+            return CategoryOverviewSerializer
         return super().get_serializer_class()
 
     def get_filterset_class(self):
@@ -4111,6 +4153,10 @@ class OverviewViewSet(BaseRLSViewSet):
             return ScanSummarySeverityFilter
         elif self.action == "findings_severity_timeseries":
             return DailySeveritySummaryFilter
+        elif self.action == "categories":
+            return CategoryOverviewFilter
+        elif self.action == "attack_surface":
+            return AttackSurfaceOverviewFilter
         return None
 
     def filter_queryset(self, queryset):
@@ -4196,29 +4242,41 @@ class OverviewViewSet(BaseRLSViewSet):
         filterset = filterset_class(normalized_params, queryset=queryset)
         return filterset.qs
 
-    def _latest_scan_ids_for_allowed_providers(self, tenant_id):
+    def _latest_scan_ids_for_allowed_providers(self, tenant_id, provider_filters=None):
         provider_filter = self._get_provider_filter()
+        queryset = Scan.all_objects.filter(
+            tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
+        )
+        if provider_filters:
+            queryset = queryset.filter(**provider_filters)
         return (
-            Scan.all_objects.filter(
-                tenant_id=tenant_id, state=StateChoices.COMPLETED, **provider_filter
-            )
-            .order_by("provider_id", "-inserted_at")
+            queryset.order_by("provider_id", "-inserted_at")
             .distinct("provider_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
+    def _extract_provider_filters_from_params(self):
+        """Extract provider filters from query params to apply on Scan queryset."""
+        params = self.request.query_params
+        filters = {}
+
+        provider_id = params.get("filter[provider_id]")
+        if provider_id:
+            filters["provider_id"] = provider_id
+
+        provider_id_in = params.get("filter[provider_id__in]")
+        if provider_id_in:
+            filters["provider_id__in"] = provider_id_in.split(",")
+
+        provider_type = params.get("filter[provider_type]")
+        if provider_type:
+            filters["provider__provider"] = provider_type
+
+        provider_type_in = params.get("filter[provider_type__in]")
+        if provider_type_in:
+            filters["provider__provider__in"] = provider_type_in.split(",")
+
+        return filters
 
     @action(detail=False, methods=["get"], url_name="providers")
     def providers(self, request):
@@ -4825,22 +4883,13 @@ class OverviewViewSet(BaseRLSViewSet):
         tenant_id = request.tenant_id
         latest_scan_ids = self._latest_scan_ids_for_allowed_providers(tenant_id)
 
-        # Build base queryset and apply user filters via FilterSet
         base_queryset = AttackSurfaceOverview.objects.filter(
             tenant_id=tenant_id, scan_id__in=latest_scan_ids
         )
         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),
             failed_findings=Coalesce(Sum("failed_findings"), 0),
@@ -4863,12 +4912,71 @@ class OverviewViewSet(BaseRLSViewSet):
             }
 
         response_data = [
-            {
-                "attack_surface_type": key,
-                **value,
-                "check_ids": attack_surface_check_ids.get(key, []),
+            {"attack_surface_type": key, **value} for key, value in results.items()
+        ]
+
+        return Response(
+            self.get_serializer(response_data, many=True).data,
+            status=status.HTTP_200_OK,
+        )
+
+    @action(detail=False, methods=["get"], url_name="categories")
+    def categories(self, request):
+        tenant_id = request.tenant_id
+        provider_filters = self._extract_provider_filters_from_params()
+        latest_scan_ids = self._latest_scan_ids_for_allowed_providers(
+            tenant_id, provider_filters
+        )
+
+        base_queryset = ScanCategorySummary.objects.filter(
+            tenant_id=tenant_id, scan_id__in=latest_scan_ids
+        )
+        provider_filter_keys = {
+            "provider_id",
+            "provider_id__in",
+            "provider_type",
+            "provider_type__in",
+        }
+        filtered_queryset = self._apply_filterset(
+            base_queryset, CategoryOverviewFilter, exclude_keys=provider_filter_keys
+        )
+
+        aggregation = (
+            filtered_queryset.values("category", "severity")
+            .annotate(
+                total=Coalesce(Sum("total_findings"), 0),
+                failed=Coalesce(Sum("failed_findings"), 0),
+                new_failed=Coalesce(Sum("new_failed_findings"), 0),
+            )
+            .order_by("category", "severity")
+        )
+
+        category_data = defaultdict(
+            lambda: {
+                "total_findings": 0,
+                "failed_findings": 0,
+                "new_failed_findings": 0,
+                "severity": {
+                    "informational": 0,
+                    "low": 0,
+                    "medium": 0,
+                    "high": 0,
+                    "critical": 0,
+                },
             }
-            for key, value in results.items()
+        )
+
+        for row in aggregation:
+            cat = row["category"]
+            sev = row["severity"]
+            category_data[cat]["total_findings"] += row["total"]
+            category_data[cat]["failed_findings"] += row["failed"]
+            category_data[cat]["new_failed_findings"] += row["new_failed"]
+            if sev in category_data[cat]["severity"]:
+                category_data[cat]["severity"][sev] = row["failed"]
+
+        response_data = [
+            {"category": cat, **data} for cat, data in sorted(category_data.items())
         ]
 
         return Response(

api/src/backend/config/guniconf.py

@@ -19,8 +19,6 @@ PORT = env("DJANGO_PORT", default=8000)
 
 # Server settings
 bind = f"{BIND_ADDRESS}:{PORT}"
-# TODO: Remove after the category filter is implemented
-limit_request_line = 0
 
 workers = env.int("DJANGO_WORKERS", default=multiprocessing.cpu_count() * 2 + 1)
 reload = DEBUG

api/src/backend/conftest.py

@@ -11,7 +11,10 @@ from django.urls import reverse
 from django_celery_results.models import TaskResult
 from rest_framework import status
 from rest_framework.test import APIClient
-from tasks.jobs.backfill import backfill_resource_scan_summaries
+from tasks.jobs.backfill import (
+    backfill_resource_scan_summaries,
+    backfill_scan_category_summaries,
+)
 
 from api.db_utils import rls_transaction
 from api.models import (
@@ -36,6 +39,7 @@ from api.models import (
     SAMLConfiguration,
     SAMLDomainIndex,
     Scan,
+    ScanCategorySummary,
     ScanSummary,
     StateChoices,
     StatusChoices,
@@ -1271,6 +1275,113 @@ def latest_scan_finding(authenticated_client, providers_fixture, resources_fixtu
     return finding
 
 
+@pytest.fixture(scope="function")
+def findings_with_categories(scans_fixture, resources_fixture):
+    scan = scans_fixture[0]
+    resource = resources_fixture[0]
+
+    finding = Finding.objects.create(
+        tenant_id=scan.tenant_id,
+        uid="finding_with_categories_1",
+        scan=scan,
+        delta=None,
+        status=Status.FAIL,
+        status_extended="test status",
+        impact=Severity.critical,
+        impact_extended="test impact",
+        severity=Severity.critical,
+        raw_result={"status": Status.FAIL},
+        check_id="genai_check",
+        check_metadata={"CheckId": "genai_check"},
+        categories=["gen-ai", "security"],
+        first_seen_at="2024-01-02T00:00:00Z",
+    )
+    finding.add_resources([resource])
+    backfill_resource_scan_summaries(str(scan.tenant_id), str(scan.id))
+    return finding
+
+
+@pytest.fixture(scope="function")
+def findings_with_multiple_categories(scans_fixture, resources_fixture):
+    scan = scans_fixture[0]
+    resource1, resource2 = resources_fixture[:2]
+
+    finding1 = Finding.objects.create(
+        tenant_id=scan.tenant_id,
+        uid="finding_multi_cat_1",
+        scan=scan,
+        delta=None,
+        status=Status.FAIL,
+        status_extended="test status",
+        impact=Severity.critical,
+        impact_extended="test impact",
+        severity=Severity.critical,
+        raw_result={"status": Status.FAIL},
+        check_id="genai_check",
+        check_metadata={"CheckId": "genai_check"},
+        categories=["gen-ai", "security"],
+        first_seen_at="2024-01-02T00:00:00Z",
+    )
+    finding1.add_resources([resource1])
+
+    finding2 = Finding.objects.create(
+        tenant_id=scan.tenant_id,
+        uid="finding_multi_cat_2",
+        scan=scan,
+        delta=None,
+        status=Status.FAIL,
+        status_extended="test status 2",
+        impact=Severity.high,
+        impact_extended="test impact 2",
+        severity=Severity.high,
+        raw_result={"status": Status.FAIL},
+        check_id="iam_check",
+        check_metadata={"CheckId": "iam_check"},
+        categories=["iam", "security"],
+        first_seen_at="2024-01-02T00:00:00Z",
+    )
+    finding2.add_resources([resource2])
+
+    backfill_resource_scan_summaries(str(scan.tenant_id), str(scan.id))
+    return finding1, finding2
+
+
+@pytest.fixture(scope="function")
+def latest_scan_finding_with_categories(
+    authenticated_client, providers_fixture, resources_fixture
+):
+    provider = providers_fixture[0]
+    tenant_id = str(providers_fixture[0].tenant_id)
+    resource = resources_fixture[0]
+    scan = Scan.objects.create(
+        name="latest completed scan with categories",
+        provider=provider,
+        trigger=Scan.TriggerChoices.MANUAL,
+        state=StateChoices.COMPLETED,
+        tenant_id=tenant_id,
+    )
+    finding = Finding.objects.create(
+        tenant_id=tenant_id,
+        uid="latest_finding_with_categories",
+        scan=scan,
+        delta="new",
+        status=Status.FAIL,
+        status_extended="test status",
+        impact=Severity.critical,
+        impact_extended="test impact",
+        severity=Severity.critical,
+        raw_result={"status": Status.FAIL},
+        check_id="genai_iam_check",
+        check_metadata={"CheckId": "genai_iam_check"},
+        categories=["gen-ai", "iam"],
+        first_seen_at="2024-01-02T00:00:00Z",
+    )
+    finding.add_resources([resource])
+    backfill_resource_scan_summaries(tenant_id, str(scan.id))
+    backfill_scan_category_summaries(tenant_id, str(scan.id))
+    return finding
+
+
 @pytest.fixture(scope="function")
 def latest_scan_resource(authenticated_client, providers_fixture):
     provider = providers_fixture[0]
@@ -1485,6 +1596,30 @@ def create_attack_surface_overview():
     return _create
 
 
+@pytest.fixture
+def create_scan_category_summary():
+    def _create(
+        tenant,
+        scan,
+        category,
+        severity,
+        total_findings=10,
+        failed_findings=5,
+        new_failed_findings=2,
+    ):
+        return ScanCategorySummary.objects.create(
+            tenant=tenant,
+            scan=scan,
+            category=category,
+            severity=severity,
+            total_findings=total_findings,
+            failed_findings=failed_findings,
+            new_failed_findings=new_failed_findings,
+        )
+
+    return _create
+
+
 def get_authorization_header(access_token: str) -> dict:
     return {"Authorization": f"Bearer {access_token}"}
 

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

@@ -3,6 +3,7 @@ from datetime import timedelta
 
 from django.db.models import Sum
 from django.utils import timezone
+from tasks.jobs.scan import aggregate_category_counts
 
 from api.db_router import READ_REPLICA_ALIAS
 from api.db_utils import rls_transaction
@@ -10,10 +11,12 @@ from api.models import (
     ComplianceOverviewSummary,
     ComplianceRequirementOverview,
     DailySeveritySummary,
+    Finding,
     Resource,
     ResourceFindingMapping,
     ResourceScanSummary,
     Scan,
+    ScanCategorySummary,
     ScanSummary,
     StateChoices,
 )
@@ -274,3 +277,67 @@ def backfill_daily_severity_summaries(tenant_id: str, days: int = None):
         "updated": updated_count,
         "total_days": len(latest_scans_by_day),
     }
+
+
+def backfill_scan_category_summaries(tenant_id: str, scan_id: str):
+    """
+    Backfill ScanCategorySummary for a completed scan.
+
+    Aggregates category counts from all findings in the scan and creates
+    one ScanCategorySummary row per (category, severity) combination.
+
+    Args:
+        tenant_id: Target tenant UUID
+        scan_id: Scan UUID to backfill
+
+    Returns:
+        dict: Status indicating whether backfill was performed
+    """
+    with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
+        if ScanCategorySummary.objects.filter(
+            tenant_id=tenant_id, scan_id=scan_id
+        ).exists():
+            return {"status": "already backfilled"}
+
+        if not Scan.objects.filter(
+            tenant_id=tenant_id,
+            id=scan_id,
+            state__in=(StateChoices.COMPLETED, StateChoices.FAILED),
+        ).exists():
+            return {"status": "scan is not completed"}
+
+        category_counts: dict[tuple[str, str], dict[str, int]] = {}
+        for finding in Finding.all_objects.filter(
+            tenant_id=tenant_id, scan_id=scan_id
+        ).values("categories", "severity", "status", "delta", "muted"):
+            aggregate_category_counts(
+                categories=finding.get("categories") or [],
+                severity=finding.get("severity"),
+                status=finding.get("status"),
+                delta=finding.get("delta"),
+                muted=finding.get("muted", False),
+                cache=category_counts,
+            )
+
+        if not category_counts:
+            return {"status": "no categories to backfill"}
+
+    category_summaries = [
+        ScanCategorySummary(
+            tenant_id=tenant_id,
+            scan_id=scan_id,
+            category=category,
+            severity=severity,
+            total_findings=counts["total"],
+            failed_findings=counts["failed"],
+            new_failed_findings=counts["new_failed"],
+        )
+        for (category, severity), counts in category_counts.items()
+    ]
+
+    with rls_transaction(tenant_id):
+        ScanCategorySummary.objects.bulk_create(
+            category_summaries, batch_size=500, ignore_conflicts=True
+        )
+
+    return {"status": "backfilled", "categories_count": len(category_counts)}

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

@@ -8,6 +8,7 @@ from collections import defaultdict
 from datetime import datetime, timezone
 from typing import Any
 
+import sentry_sdk
 from celery.utils.log import get_task_logger
 from config.env import env
 from config.settings.celery import CELERY_DEADLOCK_ATTEMPTS
@@ -39,6 +40,7 @@ from api.models import (
     ResourceScanSummary,
     ResourceTag,
     Scan,
+    ScanCategorySummary,
     ScanSummary,
     StateChoices,
 )
@@ -77,7 +79,6 @@ FINDINGS_MICRO_BATCH_SIZE = env.int("DJANGO_FINDINGS_MICRO_BATCH_SIZE", default=
 # Controls how many rows each ORM bulk_create/bulk_update call sends to Postgres
 SCAN_DB_BATCH_SIZE = env.int("DJANGO_SCAN_DB_BATCH_SIZE", default=500)
 
-
 ATTACK_SURFACE_PROVIDER_COMPATIBILITY = {
     "internet-exposed": None,  # Compatible with all providers
     "secrets": None,  # Compatible with all providers
@@ -88,6 +89,40 @@ ATTACK_SURFACE_PROVIDER_COMPATIBILITY = {
 _ATTACK_SURFACE_MAPPING_CACHE: dict[str, dict] = {}
 
 
+def aggregate_category_counts(
+    categories: list[str],
+    severity: str,
+    status: str,
+    delta: str | None,
+    muted: bool,
+    cache: dict[tuple[str, str], dict[str, int]],
+) -> None:
+    """
+    Increment category counters in-place for a finding.
+
+    Args:
+        categories: List of categories from finding metadata.
+        severity: Severity level (e.g., "high", "medium").
+        status: Finding status as string ("FAIL", "PASS").
+        delta: Delta value as string ("new", "changed") or None.
+        muted: Whether the finding is muted.
+        cache: Dict {(category, severity): {"total", "failed", "new_failed"}} to update.
+    """
+    is_failed = status == "FAIL" and not muted
+    is_new_failed = is_failed and delta == "new"
+
+    for cat in categories:
+        key = (cat, severity)
+        if key not in cache:
+            cache[key] = {"total": 0, "failed": 0, "new_failed": 0}
+        if not muted:
+            cache[key]["total"] += 1
+        if is_failed:
+            cache[key]["failed"] += 1
+        if is_new_failed:
+            cache[key]["new_failed"] += 1
+
+
 def _get_attack_surface_mapping_from_provider(provider_type: str) -> dict:
     global _ATTACK_SURFACE_MAPPING_CACHE
 
@@ -398,6 +433,7 @@ def _process_finding_micro_batch(
     unique_resources: set,
     scan_resource_cache: set,
     mute_rules_cache: dict,
+    scan_categories_cache: dict[tuple[str, str], dict[str, int]],
 ) -> None:
     """
     Process a micro-batch of findings and persist them using bulk operations.
@@ -418,6 +454,7 @@ def _process_finding_micro_batch(
         unique_resources: Set tracking (uid, region) pairs seen in the scan.
         scan_resource_cache: Set of tuples used to create `ResourceScanSummary` rows.
         mute_rules_cache: Map of finding UID -> mute reason gathered before the scan.
+        scan_categories_cache: Dict tracking category counts {(category, severity): {"total", "failed", "new_failed"}}.
     """
     # Accumulate objects for bulk operations
     findings_to_create = []
@@ -573,11 +610,12 @@ def _process_finding_micro_batch(
             resource_failed_findings_cache[resource_uid] += 1
 
         # Create finding object (don't save yet)
+        check_metadata = finding.get_metadata()
         finding_instance = Finding(
             tenant_id=tenant_id,
             uid=finding_uid,
             delta=delta,
-            check_metadata=finding.get_metadata(),
+            check_metadata=check_metadata,
             status=status,
             status_extended=finding.status_extended,
             severity=finding.severity,
@@ -590,6 +628,7 @@ def _process_finding_micro_batch(
             muted_at=datetime.now(tz=timezone.utc) if is_muted else None,
             muted_reason=muted_reason,
             compliance=finding.compliance,
+            categories=check_metadata.get("categories", []) or [],
         )
         findings_to_create.append(finding_instance)
         resource_denormalized_data.append((finding_instance, resource_instance))
@@ -604,6 +643,16 @@ def _process_finding_micro_batch(
             )
         )
 
+        # Track categories with counts for ScanCategorySummary by (category, severity)
+        aggregate_category_counts(
+            categories=check_metadata.get("categories", []) or [],
+            severity=finding.severity.value,
+            status=status.value,
+            delta=delta.value if delta else None,
+            muted=is_muted,
+            cache=scan_categories_cache,
+        )
+
     # Bulk operations within single transaction
     with rls_transaction(tenant_id):
         # Bulk create findings
@@ -703,6 +752,7 @@ def perform_prowler_scan(
     exception = None
     unique_resources = set()
     scan_resource_cache: set[tuple[str, str, str, str]] = set()
+    scan_categories_cache: dict[tuple[str, str], dict[str, int]] = {}
     start_time = time.time()
     exc = None
 
@@ -792,6 +842,7 @@ def perform_prowler_scan(
                     unique_resources=unique_resources,
                     scan_resource_cache=scan_resource_cache,
                     mute_rules_cache=mute_rules_cache,
+                    scan_categories_cache=scan_categories_cache,
                 )
 
             # Update scan progress
@@ -851,13 +902,33 @@ def perform_prowler_scan(
                 resource_scan_summaries, batch_size=500, ignore_conflicts=True
             )
     except Exception as filter_exception:
-        import sentry_sdk
-
         sentry_sdk.capture_exception(filter_exception)
         logger.error(
             f"Error storing filter values for scan {scan_id}: {filter_exception}"
         )
 
+    try:
+        if scan_categories_cache:
+            category_summaries = [
+                ScanCategorySummary(
+                    tenant_id=tenant_id,
+                    scan_id=scan_id,
+                    category=category,
+                    severity=severity,
+                    total_findings=counts["total"],
+                    failed_findings=counts["failed"],
+                    new_failed_findings=counts["new_failed"],
+                )
+                for (category, severity), counts in scan_categories_cache.items()
+            ]
+            with rls_transaction(tenant_id):
+                ScanCategorySummary.objects.bulk_create(
+                    category_summaries, batch_size=500, ignore_conflicts=True
+                )
+    except Exception as cat_exception:
+        sentry_sdk.capture_exception(cat_exception)
+        logger.error(f"Error storing categories for scan {scan_id}: {cat_exception}")
+
     serializer = ScanTaskSerializer(instance=scan_instance)
     return serializer.data
 

api/src/backend/tasks/tasks.py

@@ -12,6 +12,7 @@ from tasks.jobs.backfill import (
     backfill_compliance_summaries,
     backfill_daily_severity_summaries,
     backfill_resource_scan_summaries,
+    backfill_scan_category_summaries,
 )
 from tasks.jobs.connection import (
     check_integration_connection,
@@ -534,6 +535,21 @@ def backfill_daily_severity_summaries_task(tenant_id: str, days: int = None):
     return backfill_daily_severity_summaries(tenant_id=tenant_id, days=days)
 
 
+@shared_task(name="backfill-scan-category-summaries", queue="backfill")
+@handle_provider_deletion
+def backfill_scan_category_summaries_task(tenant_id: str, scan_id: str):
+    """
+    Backfill ScanCategorySummary for a completed scan.
+
+    Aggregates unique categories from findings and creates a summary row.
+
+    Args:
+        tenant_id (str): The tenant identifier.
+        scan_id (str): The scan identifier.
+    """
+    return backfill_scan_category_summaries(tenant_id=tenant_id, scan_id=scan_id)
+
+
 @shared_task(base=RLSTask, name="scan-compliance-overviews", queue="compliance")
 @handle_provider_deletion
 def create_compliance_requirements_task(tenant_id: str, scan_id: str):

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

@@ -4,14 +4,19 @@ import pytest
 from tasks.jobs.backfill import (
     backfill_compliance_summaries,
     backfill_resource_scan_summaries,
+    backfill_scan_category_summaries,
 )
 
 from api.models import (
     ComplianceOverviewSummary,
+    Finding,
     ResourceScanSummary,
     Scan,
+    ScanCategorySummary,
     StateChoices,
 )
+from prowler.lib.check.models import Severity
+from prowler.lib.outputs.finding import Status
 
 
 @pytest.fixture(scope="function")
@@ -46,6 +51,45 @@ def get_not_completed_scans(providers_fixture):
     return scan_1, scan_2
 
 
+@pytest.fixture(scope="function")
+def findings_with_categories_fixture(scans_fixture, resources_fixture):
+    scan = scans_fixture[0]
+    resource = resources_fixture[0]
+
+    finding = Finding.objects.create(
+        tenant_id=scan.tenant_id,
+        uid="finding_with_categories",
+        scan=scan,
+        delta="new",
+        status=Status.FAIL,
+        status_extended="test status",
+        impact=Severity.critical,
+        impact_extended="test impact",
+        severity=Severity.critical,
+        raw_result={"status": Status.FAIL},
+        check_id="test_check",
+        check_metadata={"CheckId": "test_check"},
+        categories=["gen-ai", "security"],
+        first_seen_at="2024-01-02T00:00:00Z",
+    )
+    finding.add_resources([resource])
+    return finding
+
+
+@pytest.fixture(scope="function")
+def scan_category_summary_fixture(scans_fixture):
+    scan = scans_fixture[0]
+    return ScanCategorySummary.objects.create(
+        tenant_id=scan.tenant_id,
+        scan=scan,
+        category="existing-category",
+        severity=Severity.critical,
+        total_findings=1,
+        failed_findings=0,
+        new_failed_findings=0,
+    )
+
+
 @pytest.mark.django_db
 class TestBackfillResourceScanSummaries:
     def test_already_backfilled(self, resource_scan_summary_data):
@@ -172,3 +216,47 @@ class TestBackfillComplianceSummaries:
             assert summary.requirements_failed == expected_counts["requirements_failed"]
             assert summary.requirements_manual == expected_counts["requirements_manual"]
             assert summary.total_requirements == expected_counts["total_requirements"]
+
+
+@pytest.mark.django_db
+class TestBackfillScanCategorySummaries:
+    def test_already_backfilled(self, scan_category_summary_fixture):
+        tenant_id = scan_category_summary_fixture.tenant_id
+        scan_id = scan_category_summary_fixture.scan_id
+
+        result = backfill_scan_category_summaries(str(tenant_id), str(scan_id))
+
+        assert result == {"status": "already backfilled"}
+
+    def test_not_completed_scan(self, get_not_completed_scans):
+        for scan in get_not_completed_scans:
+            result = backfill_scan_category_summaries(str(scan.tenant_id), str(scan.id))
+            assert result == {"status": "scan is not completed"}
+
+    def test_no_categories_to_backfill(self, scans_fixture):
+        scan = scans_fixture[1]  # Failed scan with no findings
+        result = backfill_scan_category_summaries(str(scan.tenant_id), str(scan.id))
+        assert result == {"status": "no categories to backfill"}
+
+    def test_successful_backfill(self, findings_with_categories_fixture):
+        finding = findings_with_categories_fixture
+        tenant_id = str(finding.tenant_id)
+        scan_id = str(finding.scan_id)
+
+        result = backfill_scan_category_summaries(tenant_id, scan_id)
+
+        # 2 categories × 1 severity = 2 rows
+        assert result == {"status": "backfilled", "categories_count": 2}
+
+        summaries = ScanCategorySummary.objects.filter(
+            tenant_id=tenant_id, scan_id=scan_id
+        )
+        assert summaries.count() == 2
+        categories = set(summaries.values_list("category", flat=True))
+        assert categories == {"gen-ai", "security"}
+
+        for summary in summaries:
+            assert summary.severity == Severity.critical
+            assert summary.total_findings == 1
+            assert summary.failed_findings == 1
+            assert summary.new_failed_findings == 1

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

@@ -20,6 +20,7 @@ from tasks.jobs.scan import (
     _process_finding_micro_batch,
     _store_resources,
     aggregate_attack_surface,
+    aggregate_category_counts,
     aggregate_findings,
     create_compliance_requirements,
     perform_prowler_scan,
@@ -1377,6 +1378,7 @@ class TestProcessFindingMicroBatch:
         unique_resources: set[tuple[str, str]] = set()
         scan_resource_cache: set[tuple[str, str, str, str]] = set()
         mute_rules_cache = {}
+        scan_categories_cache: dict[tuple[str, str], dict[str, int]] = {}
 
         with (
             patch("tasks.jobs.scan.rls_transaction", new=noop_rls_transaction),
@@ -1394,6 +1396,7 @@ class TestProcessFindingMicroBatch:
                 unique_resources,
                 scan_resource_cache,
                 mute_rules_cache,
+                scan_categories_cache,
             )
 
         created_finding = Finding.objects.get(uid=finding.uid)
@@ -1486,6 +1489,7 @@ class TestProcessFindingMicroBatch:
         unique_resources: set[tuple[str, str]] = set()
         scan_resource_cache: set[tuple[str, str, str, str]] = set()
         mute_rules_cache = {finding.uid: "Muted via rule"}
+        scan_categories_cache: dict[tuple[str, str], dict[str, int]] = {}
 
         with (
             patch("tasks.jobs.scan.rls_transaction", new=noop_rls_transaction),
@@ -1503,6 +1507,7 @@ class TestProcessFindingMicroBatch:
                 unique_resources,
                 scan_resource_cache,
                 mute_rules_cache,
+                scan_categories_cache,
             )
 
         existing_resource.refresh_from_db()
@@ -1610,6 +1615,7 @@ class TestProcessFindingMicroBatch:
         unique_resources: set[tuple[str, str]] = set()
         scan_resource_cache: set[tuple[str, str, str, str]] = set()
         mute_rules_cache = {}
+        scan_categories_cache: dict[tuple[str, str], dict[str, int]] = {}
 
         with (
             patch("tasks.jobs.scan.rls_transaction", new=noop_rls_transaction),
@@ -1628,6 +1634,7 @@ class TestProcessFindingMicroBatch:
                 unique_resources,
                 scan_resource_cache,
                 mute_rules_cache,
+                scan_categories_cache,
             )
 
         # Verify the long UID finding was NOT created
@@ -1648,6 +1655,118 @@ class TestProcessFindingMicroBatch:
             for call in warning_calls
         )
 
+    def test_process_finding_micro_batch_tracks_categories(
+        self, tenants_fixture, scans_fixture
+    ):
+        tenant = tenants_fixture[0]
+        scan = scans_fixture[0]
+        provider = scan.provider
+
+        finding1 = FakeFinding(
+            uid="finding-cat-1",
+            status=StatusChoices.PASS,
+            status_extended="all good",
+            severity=Severity.low,
+            check_id="genai_check",
+            resource_uid="arn:aws:bedrock:::model/test",
+            resource_name="test-model",
+            region="us-east-1",
+            service_name="bedrock",
+            resource_type="model",
+            resource_tags={},
+            resource_metadata={},
+            resource_details={},
+            partition="aws",
+            raw={},
+            compliance={},
+            metadata={"categories": ["gen-ai", "security"]},
+            muted=False,
+        )
+
+        finding2 = FakeFinding(
+            uid="finding-cat-2",
+            status=StatusChoices.FAIL,
+            status_extended="bad",
+            severity=Severity.high,
+            check_id="iam_check",
+            resource_uid="arn:aws:iam:::user/test",
+            resource_name="test-user",
+            region="us-east-1",
+            service_name="iam",
+            resource_type="user",
+            resource_tags={},
+            resource_metadata={},
+            resource_details={},
+            partition="aws",
+            raw={},
+            compliance={},
+            metadata={"categories": ["security", "iam"]},
+            muted=False,
+        )
+
+        resource_cache = {}
+        tag_cache = {}
+        last_status_cache = {}
+        resource_failed_findings_cache = {}
+        unique_resources: set[tuple[str, str]] = set()
+        scan_resource_cache: set[tuple[str, str, str, str]] = set()
+        mute_rules_cache = {}
+        scan_categories_cache: dict[tuple[str, str], dict[str, int]] = {}
+
+        with (
+            patch("tasks.jobs.scan.rls_transaction", new=noop_rls_transaction),
+            patch("api.db_utils.rls_transaction", new=noop_rls_transaction),
+        ):
+            _process_finding_micro_batch(
+                str(tenant.id),
+                [finding1, finding2],
+                scan,
+                provider,
+                resource_cache,
+                tag_cache,
+                last_status_cache,
+                resource_failed_findings_cache,
+                unique_resources,
+                scan_resource_cache,
+                mute_rules_cache,
+                scan_categories_cache,
+            )
+
+        # finding1: PASS, severity=low, categories=["gen-ai", "security"]
+        # finding2: FAIL, severity=high, categories=["security", "iam"]
+        # Keys are (category, severity) tuples
+        assert set(scan_categories_cache.keys()) == {
+            ("gen-ai", "low"),
+            ("security", "low"),
+            ("security", "high"),
+            ("iam", "high"),
+        }
+        assert scan_categories_cache[("gen-ai", "low")] == {
+            "total": 1,
+            "failed": 0,
+            "new_failed": 0,
+        }
+        assert scan_categories_cache[("security", "low")] == {
+            "total": 1,
+            "failed": 0,
+            "new_failed": 0,
+        }
+        assert scan_categories_cache[("security", "high")] == {
+            "total": 1,
+            "failed": 1,
+            "new_failed": 1,
+        }
+        assert scan_categories_cache[("iam", "high")] == {
+            "total": 1,
+            "failed": 1,
+            "new_failed": 1,
+        }
+
+        created_finding1 = Finding.objects.get(uid="finding-cat-1")
+        created_finding2 = Finding.objects.get(uid="finding-cat-2")
+        assert set(created_finding1.categories) == {"gen-ai", "security"}
+        assert set(created_finding2.categories) == {"security", "iam"}
+
 
 @pytest.mark.django_db
 class TestCreateComplianceRequirements:
@@ -3756,3 +3875,150 @@ class TestAggregateAttackSurface:
             aggregate_attack_surface(str(tenant.id), str(scan.id))
 
         mock_select_related.assert_called_once_with("provider")
+
+
+class TestAggregateCategoryCounts:
+    """Test aggregate_category_counts helper function."""
+
+    def test_aggregate_category_counts_basic(self):
+        """Test basic category counting for a non-muted PASS finding."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=["security", "iam"],
+            severity="high",
+            status="PASS",
+            delta=None,
+            muted=False,
+            cache=cache,
+        )
+
+        assert ("security", "high") in cache
+        assert ("iam", "high") in cache
+        assert cache[("security", "high")] == {"total": 1, "failed": 0, "new_failed": 0}
+        assert cache[("iam", "high")] == {"total": 1, "failed": 0, "new_failed": 0}
+
+    def test_aggregate_category_counts_fail_not_muted(self):
+        """Test category counting for a non-muted FAIL finding."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=["security"],
+            severity="critical",
+            status="FAIL",
+            delta=None,
+            muted=False,
+            cache=cache,
+        )
+
+        assert cache[("security", "critical")] == {
+            "total": 1,
+            "failed": 1,
+            "new_failed": 0,
+        }
+
+    def test_aggregate_category_counts_new_fail(self):
+        """Test category counting for a new FAIL finding (delta='new')."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=["gen-ai"],
+            severity="high",
+            status="FAIL",
+            delta="new",
+            muted=False,
+            cache=cache,
+        )
+
+        assert cache[("gen-ai", "high")] == {"total": 1, "failed": 1, "new_failed": 1}
+
+    def test_aggregate_category_counts_muted_finding(self):
+        """Test that muted findings are excluded from all counts."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=["security"],
+            severity="high",
+            status="FAIL",
+            delta="new",
+            muted=True,
+            cache=cache,
+        )
+
+        assert cache[("security", "high")] == {"total": 0, "failed": 0, "new_failed": 0}
+
+    def test_aggregate_category_counts_accumulates(self):
+        """Test that multiple calls accumulate counts."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+
+        # First finding: PASS
+        aggregate_category_counts(
+            categories=["security"],
+            severity="high",
+            status="PASS",
+            delta=None,
+            muted=False,
+            cache=cache,
+        )
+
+        # Second finding: FAIL (new)
+        aggregate_category_counts(
+            categories=["security"],
+            severity="high",
+            status="FAIL",
+            delta="new",
+            muted=False,
+            cache=cache,
+        )
+
+        # Third finding: FAIL (changed)
+        aggregate_category_counts(
+            categories=["security"],
+            severity="high",
+            status="FAIL",
+            delta="changed",
+            muted=False,
+            cache=cache,
+        )
+
+        assert cache[("security", "high")] == {"total": 3, "failed": 2, "new_failed": 1}
+
+    def test_aggregate_category_counts_empty_categories(self):
+        """Test with empty categories list."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=[],
+            severity="high",
+            status="FAIL",
+            delta="new",
+            muted=False,
+            cache=cache,
+        )
+
+        assert cache == {}
+
+    def test_aggregate_category_counts_changed_delta(self):
+        """Test that changed delta increments failed but not new_failed."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=["iam"],
+            severity="medium",
+            status="FAIL",
+            delta="changed",
+            muted=False,
+            cache=cache,
+        )
+
+        assert cache[("iam", "medium")] == {"total": 1, "failed": 1, "new_failed": 0}
+
+    def test_aggregate_category_counts_multiple_categories_single_finding(self):
+        """Test single finding with multiple categories."""
+        cache: dict[tuple[str, str], dict[str, int]] = {}
+        aggregate_category_counts(
+            categories=["security", "compliance", "data-protection"],
+            severity="low",
+            status="FAIL",
+            delta="new",
+            muted=False,
+            cache=cache,
+        )
+
+        assert len(cache) == 3
+        for cat in ["security", "compliance", "data-protection"]:
+            assert cache[(cat, "low")] == {"total": 1, "failed": 1, "new_failed": 1}