feat(api): add findings severity timeseries endpoint (#9363)

api/CHANGELOG.md

@@ -6,6 +6,7 @@ All notable changes to the **Prowler API** are documented in this file.
 
 ### Added
 - New endpoint to retrieve an overview of the attack surfaces [(#9309)](https://github.com/prowler-cloud/prowler/pull/9309)
+- New endpoint `GET /api/v1/overviews/findings_severity/timeseries` to retrieve daily aggregated findings by severity level [(#9363)](https://github.com/prowler-cloud/prowler/pull/9363)
 - Lighthouse AI support for Amazon Bedrock API key [(#9343)](https://github.com/prowler-cloud/prowler/pull/9343)
 - Exception handler for provider deletions during scans [(#9414)](https://github.com/prowler-cloud/prowler/pull/9414)
 - Support to use admin credentials through the read replica database [(#9440)](https://github.com/prowler-cloud/prowler/pull/9440)

api/src/backend/api/filters.py

@@ -25,6 +25,7 @@ from api.db_utils import (
 from api.models import (
     AttackSurfaceOverview,
     ComplianceRequirementOverview,
+    DailySeveritySummary,
     Finding,
     Integration,
     Invitation,
@@ -795,6 +796,68 @@ class ScanSummaryFilter(FilterSet):
         }
 
 
+class DailySeveritySummaryFilter(FilterSet):
+    """Filter for findings_severity/timeseries endpoint."""
+
+    MAX_DATE_RANGE_DAYS = 365
+
+    provider_id = UUIDFilter(field_name="provider_id", lookup_expr="exact")
+    provider_id__in = UUIDInFilter(field_name="provider_id", lookup_expr="in")
+    provider_type = ChoiceFilter(
+        field_name="provider__provider", choices=Provider.ProviderChoices.choices
+    )
+    provider_type__in = ChoiceInFilter(
+        field_name="provider__provider", choices=Provider.ProviderChoices.choices
+    )
+    date_from = DateFilter(method="filter_noop")
+    date_to = DateFilter(method="filter_noop")
+
+    class Meta:
+        model = DailySeveritySummary
+        fields = ["provider_id"]
+
+    def filter_noop(self, queryset, name, value):
+        return queryset
+
+    def filter_queryset(self, queryset):
+        if not self.data.get("date_from"):
+            raise ValidationError(
+                [
+                    {
+                        "detail": "This query parameter is required.",
+                        "status": "400",
+                        "source": {"pointer": "filter[date_from]"},
+                        "code": "required",
+                    }
+                ]
+            )
+
+        today = date.today()
+        date_from = self.form.cleaned_data.get("date_from")
+        date_to = min(self.form.cleaned_data.get("date_to") or today, today)
+
+        if (date_to - date_from).days > self.MAX_DATE_RANGE_DAYS:
+            raise ValidationError(
+                [
+                    {
+                        "detail": f"Date range cannot exceed {self.MAX_DATE_RANGE_DAYS} days.",
+                        "status": "400",
+                        "source": {"pointer": "filter[date_from]"},
+                        "code": "invalid",
+                    }
+                ]
+            )
+
+        # View access
+        self.request._date_from = date_from
+        self.request._date_to = date_to
+
+        # Apply date filter (only lte for fill-forward logic)
+        queryset = queryset.filter(date__lte=date_to)
+
+        return super().filter_queryset(queryset)
+
+
 class ScanSummarySeverityFilter(ScanSummaryFilter):
     """Filter for findings_severity ScanSummary endpoint - includes status filters"""
 

api/src/backend/api/migrations/0061_daily_severity_summary.py

@@ -0,0 +1,96 @@
+# Generated by Django 5.1.14 on 2025-12-03 13:38
+
+import uuid
+
+import django.db.models.deletion
+from django.db import migrations, models
+
+import api.rls
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0060_attack_surface_overview"),
+    ]
+
+    operations = [
+        migrations.CreateModel(
+            name="DailySeveritySummary",
+            fields=[
+                (
+                    "id",
+                    models.UUIDField(
+                        default=uuid.uuid4,
+                        editable=False,
+                        primary_key=True,
+                        serialize=False,
+                    ),
+                ),
+                ("date", models.DateField()),
+                ("critical", models.IntegerField(default=0)),
+                ("high", models.IntegerField(default=0)),
+                ("medium", models.IntegerField(default=0)),
+                ("low", models.IntegerField(default=0)),
+                ("informational", models.IntegerField(default=0)),
+                ("muted", models.IntegerField(default=0)),
+                (
+                    "provider",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="daily_severity_summaries",
+                        related_query_name="daily_severity_summary",
+                        to="api.provider",
+                    ),
+                ),
+                (
+                    "scan",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        related_name="daily_severity_summaries",
+                        related_query_name="daily_severity_summary",
+                        to="api.scan",
+                    ),
+                ),
+                (
+                    "tenant",
+                    models.ForeignKey(
+                        on_delete=django.db.models.deletion.CASCADE,
+                        to="api.tenant",
+                    ),
+                ),
+            ],
+            options={
+                "db_table": "daily_severity_summaries",
+                "abstract": False,
+            },
+        ),
+        migrations.AddIndex(
+            model_name="dailyseveritysummary",
+            index=models.Index(
+                fields=["tenant_id", "id"],
+                name="dss_tenant_id_idx",
+            ),
+        ),
+        migrations.AddIndex(
+            model_name="dailyseveritysummary",
+            index=models.Index(
+                fields=["tenant_id", "provider_id"],
+                name="dss_tenant_provider_idx",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="dailyseveritysummary",
+            constraint=models.UniqueConstraint(
+                fields=("tenant_id", "provider", "date"),
+                name="unique_daily_severity_summary",
+            ),
+        ),
+        migrations.AddConstraint(
+            model_name="dailyseveritysummary",
+            constraint=api.rls.RowLevelSecurityConstraint(
+                "tenant_id",
+                name="rls_on_dailyseveritysummary",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ),
+    ]

api/src/backend/api/models.py

@@ -1500,6 +1500,65 @@ class ScanSummary(RowLevelSecurityProtectedModel):
         resource_name = "scan-summaries"
 
 
+class DailySeveritySummary(RowLevelSecurityProtectedModel):
+    """
+    Pre-aggregated daily severity counts per provider.
+    Used by findings_severity/timeseries endpoint for efficient queries.
+    """
+
+    objects = ActiveProviderManager()
+
+    id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
+    date = models.DateField()
+
+    provider = models.ForeignKey(
+        Provider,
+        on_delete=models.CASCADE,
+        related_name="daily_severity_summaries",
+        related_query_name="daily_severity_summary",
+    )
+    scan = models.ForeignKey(
+        Scan,
+        on_delete=models.CASCADE,
+        related_name="daily_severity_summaries",
+        related_query_name="daily_severity_summary",
+    )
+
+    # Aggregated fail counts by severity
+    critical = models.IntegerField(default=0)
+    high = models.IntegerField(default=0)
+    medium = models.IntegerField(default=0)
+    low = models.IntegerField(default=0)
+    informational = models.IntegerField(default=0)
+    muted = models.IntegerField(default=0)
+
+    class Meta(RowLevelSecurityProtectedModel.Meta):
+        db_table = "daily_severity_summaries"
+
+        constraints = [
+            models.UniqueConstraint(
+                fields=("tenant_id", "provider", "date"),
+                name="unique_daily_severity_summary",
+            ),
+            RowLevelSecurityConstraint(
+                field="tenant_id",
+                name="rls_on_%(class)s",
+                statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
+            ),
+        ]
+
+        indexes = [
+            models.Index(
+                fields=["tenant_id", "id"],
+                name="dss_tenant_id_idx",
+            ),
+            models.Index(
+                fields=["tenant_id", "provider_id"],
+                name="dss_tenant_provider_idx",
+            ),
+        ]
+
+
 class Integration(RowLevelSecurityProtectedModel):
     class IntegrationChoices(models.TextChoices):
         AMAZON_S3 = "amazon_s3", _("Amazon S3")

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

@@ -4940,6 +4940,154 @@ paths:
               schema:
                 $ref: '#/components/schemas/OverviewSeverityResponse'
           description: ''
+  /api/v1/overviews/findings_severity/timeseries:
+    get:
+      operationId: overviews_findings_severity_timeseries_retrieve
+      description: Retrieve daily aggregated findings data grouped by severity levels
+        over a date range. Returns one data point per day with counts of failed findings
+        by severity (critical, high, medium, low, informational) and muted findings.
+        Days without scans are filled forward with the most recent known values. Use
+        date_from (required) and date_to filters to specify the range.
+      summary: Get findings severity data over time
+      parameters:
+      - in: query
+        name: fields[findings-severity-timeseries]
+        schema:
+          type: array
+          items:
+            type: string
+            enum:
+            - id
+            - critical
+            - high
+            - medium
+            - low
+            - informational
+            - muted
+            - scan_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[date_from]
+        schema:
+          type: string
+      - in: query
+        name: filter[date_to]
+        schema:
+          type: string
+      - 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
+          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
+            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: 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
+            - critical
+            - -critical
+            - high
+            - -high
+            - medium
+            - -medium
+            - low
+            - -low
+            - informational
+            - -informational
+            - muted
+            - -muted
+            - scan_ids
+            - -scan_ids
+        explode: false
+      tags:
+      - Overview
+      security:
+      - JWT or API Key: []
+      responses:
+        '200':
+          content:
+            application/vnd.api+json:
+              schema:
+                $ref: '#/components/schemas/FindingsSeverityOverTimeResponse'
+          description: ''
   /api/v1/overviews/providers:
     get:
       operationId: overviews_providers_retrieve
@@ -11099,6 +11247,60 @@ components:
           $ref: '#/components/schemas/Finding'
       required:
       - data
+    FindingsSeverityOverTime:
+      type: object
+      required:
+      - 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:
+          - findings-severity-over-time
+        id: {}
+        attributes:
+          type: object
+          properties:
+            id:
+              type: string
+              format: date
+            critical:
+              type: integer
+            high:
+              type: integer
+            medium:
+              type: integer
+            low:
+              type: integer
+            informational:
+              type: integer
+            muted:
+              type: integer
+            scan_ids:
+              type: array
+              items:
+                type: string
+                format: uuid
+          required:
+          - id
+          - critical
+          - high
+          - medium
+          - low
+          - informational
+          - muted
+          - scan_ids
+    FindingsSeverityOverTimeResponse:
+      type: object
+      properties:
+        data:
+          $ref: '#/components/schemas/FindingsSeverityOverTime'
+      required:
+      - data
     Integration:
       type: object
       required:

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

@@ -3,7 +3,7 @@ import io
 import json
 import os
 import tempfile
-from datetime import datetime, timedelta, timezone
+from datetime import date, datetime, timedelta, timezone
 from decimal import Decimal
 from pathlib import Path
 from types import SimpleNamespace
@@ -38,6 +38,7 @@ from api.models import (
     AttackSurfaceOverview,
     ComplianceOverviewSummary,
     ComplianceRequirementOverview,
+    DailySeveritySummary,
     Finding,
     Integration,
     Invitation,
@@ -6984,6 +6985,270 @@ class TestOverviewViewSet:
         assert combined_attributes["medium"] == 4
         assert combined_attributes["critical"] == 3
 
+    def test_overview_findings_severity_timeseries_requires_date_from(
+        self, authenticated_client
+    ):
+        response = authenticated_client.get(
+            reverse("overview-findings_severity_timeseries")
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "date_from" in response.json()["errors"][0]["source"]["pointer"]
+
+    def test_overview_findings_severity_timeseries_invalid_date_format(
+        self, authenticated_client
+    ):
+        response = authenticated_client.get(
+            reverse("overview-findings_severity_timeseries"),
+            {"filter[date_from]": "invalid-date"},
+        )
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert "Enter a valid date." in response.json()["errors"][0]["detail"]
+
+    def test_overview_findings_severity_timeseries_empty_data(
+        self, authenticated_client
+    ):
+        response = authenticated_client.get(
+            reverse("overview-findings_severity_timeseries"),
+            {
+                "filter[date_from]": "2024-01-01",
+                "filter[date_to]": "2024-01-03",
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        # Should return 3 days with fill-forward (all zeros since no data)
+        assert len(data) == 3
+        for item in data:
+            assert item["attributes"]["critical"] == 0
+            assert item["attributes"]["high"] == 0
+            assert item["attributes"]["medium"] == 0
+            assert item["attributes"]["low"] == 0
+            assert item["attributes"]["informational"] == 0
+            assert item["attributes"]["muted"] == 0
+            assert item["attributes"]["scan_ids"] == []
+
+    def test_overview_findings_severity_timeseries_with_data(
+        self, authenticated_client, tenants_fixture, providers_fixture
+    ):
+        tenant = tenants_fixture[0]
+        provider1, provider2, *_ = providers_fixture
+
+        # Create scan for day 1
+        scan1 = Scan.objects.create(
+            name="severity-over-time-scan-1",
+            provider=provider1,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+            completed_at=datetime(2024, 1, 1, 12, 0, 0, tzinfo=timezone.utc),
+        )
+
+        # Create scan for day 3
+        scan3 = Scan.objects.create(
+            name="severity-over-time-scan-3",
+            provider=provider1,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+            completed_at=datetime(2024, 1, 3, 12, 0, 0, tzinfo=timezone.utc),
+        )
+
+        # Create DailySeveritySummary for day 1
+        DailySeveritySummary.objects.create(
+            tenant=tenant,
+            provider=provider1,
+            scan=scan1,
+            date=date(2024, 1, 1),
+            critical=10,
+            high=20,
+            medium=30,
+            low=40,
+            informational=50,
+            muted=5,
+        )
+
+        # Create DailySeveritySummary for day 3
+        DailySeveritySummary.objects.create(
+            tenant=tenant,
+            provider=provider1,
+            scan=scan3,
+            date=date(2024, 1, 3),
+            critical=15,
+            high=25,
+            medium=35,
+            low=45,
+            informational=55,
+            muted=10,
+        )
+
+        response = authenticated_client.get(
+            reverse("overview-findings_severity_timeseries"),
+            {
+                "filter[date_from]": "2024-01-01",
+                "filter[date_to]": "2024-01-03",
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 3
+
+        # Day 1 - actual data (id is the date)
+        assert data[0]["id"] == "2024-01-01"
+        assert data[0]["attributes"]["critical"] == 10
+        assert data[0]["attributes"]["high"] == 20
+        assert data[0]["attributes"]["scan_ids"] == [str(scan1.id)]
+
+        # Day 2 - fill forward from day 1 (no data for this day)
+        assert data[1]["id"] == "2024-01-02"
+        assert data[1]["attributes"]["critical"] == 10
+        assert data[1]["attributes"]["high"] == 20
+        assert data[1]["attributes"]["scan_ids"] == [str(scan1.id)]
+
+        # Day 3 - actual data
+        assert data[2]["id"] == "2024-01-03"
+        assert data[2]["attributes"]["critical"] == 15
+        assert data[2]["attributes"]["high"] == 25
+        assert data[2]["attributes"]["scan_ids"] == [str(scan3.id)]
+
+    def test_overview_findings_severity_timeseries_aggregates_providers(
+        self, authenticated_client, tenants_fixture, providers_fixture
+    ):
+        tenant = tenants_fixture[0]
+        provider1, provider2, *_ = providers_fixture
+
+        # Same day, different providers
+        scan1 = Scan.objects.create(
+            name="severity-over-time-scan-p1",
+            provider=provider1,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+            completed_at=datetime(2024, 2, 1, 12, 0, 0, tzinfo=timezone.utc),
+        )
+        scan2 = Scan.objects.create(
+            name="severity-over-time-scan-p2",
+            provider=provider2,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+            completed_at=datetime(2024, 2, 1, 14, 0, 0, tzinfo=timezone.utc),
+        )
+
+        # Create DailySeveritySummary for provider1
+        DailySeveritySummary.objects.create(
+            tenant=tenant,
+            provider=provider1,
+            scan=scan1,
+            date=date(2024, 2, 1),
+            critical=10,
+            high=20,
+            medium=30,
+            low=40,
+            informational=50,
+            muted=5,
+        )
+
+        # Create DailySeveritySummary for provider2
+        DailySeveritySummary.objects.create(
+            tenant=tenant,
+            provider=provider2,
+            scan=scan2,
+            date=date(2024, 2, 1),
+            critical=5,
+            high=10,
+            medium=15,
+            low=20,
+            informational=25,
+            muted=3,
+        )
+
+        response = authenticated_client.get(
+            reverse("overview-findings_severity_timeseries"),
+            {
+                "filter[date_from]": "2024-02-01",
+                "filter[date_to]": "2024-02-01",
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+
+        # Should aggregate both providers
+        assert data[0]["attributes"]["critical"] == 15  # 10 + 5
+        assert data[0]["attributes"]["high"] == 30  # 20 + 10
+        assert data[0]["attributes"]["medium"] == 45  # 30 + 15
+        assert data[0]["attributes"]["low"] == 60  # 40 + 20
+        assert data[0]["attributes"]["informational"] == 75  # 50 + 25
+        assert data[0]["attributes"]["muted"] == 8  # 5 + 3
+        # scan_ids should contain both scans (order may vary)
+        assert set(data[0]["attributes"]["scan_ids"]) == {str(scan1.id), str(scan2.id)}
+
+    def test_overview_findings_severity_timeseries_provider_filter(
+        self, authenticated_client, tenants_fixture, providers_fixture
+    ):
+        tenant = tenants_fixture[0]
+        provider1, provider2, *_ = providers_fixture
+
+        scan1 = Scan.objects.create(
+            name="severity-over-time-filter-scan-p1",
+            provider=provider1,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+            completed_at=datetime(2024, 3, 1, 12, 0, 0, tzinfo=timezone.utc),
+        )
+        scan2 = Scan.objects.create(
+            name="severity-over-time-filter-scan-p2",
+            provider=provider2,
+            trigger=Scan.TriggerChoices.MANUAL,
+            state=StateChoices.COMPLETED,
+            tenant=tenant,
+            completed_at=datetime(2024, 3, 1, 14, 0, 0, tzinfo=timezone.utc),
+        )
+
+        # Provider 1 - critical=100
+        DailySeveritySummary.objects.create(
+            tenant=tenant,
+            provider=provider1,
+            scan=scan1,
+            date=date(2024, 3, 1),
+            critical=100,
+            high=0,
+            medium=0,
+            low=0,
+            informational=0,
+            muted=0,
+        )
+
+        # Provider 2 - critical=50
+        DailySeveritySummary.objects.create(
+            tenant=tenant,
+            provider=provider2,
+            scan=scan2,
+            date=date(2024, 3, 1),
+            critical=50,
+            high=0,
+            medium=0,
+            low=0,
+            informational=0,
+            muted=0,
+        )
+
+        # Filter by provider1 only
+        response = authenticated_client.get(
+            reverse("overview-findings_severity_timeseries"),
+            {
+                "filter[date_from]": "2024-03-01",
+                "filter[date_to]": "2024-03-01",
+                "filter[provider_id]": str(provider1.id),
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        data = response.json()["data"]
+        assert len(data) == 1
+        assert data[0]["attributes"]["critical"] == 100  # Only provider1
+        assert data[0]["attributes"]["scan_ids"] == [str(scan1.id)]
+
     def test_overview_attack_surface_no_data(self, authenticated_client):
         response = authenticated_client.get(reverse("overview-attack-surface"))
         assert response.status_code == status.HTTP_200_OK

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

@@ -2204,6 +2204,22 @@ class OverviewSeveritySerializer(BaseSerializerV1):
         resource_name = "findings-severity-overview"
 
 
+class FindingsSeverityOverTimeSerializer(BaseSerializerV1):
+    """Serializer for daily findings severity trend data."""
+
+    id = serializers.DateField(source="date")
+    critical = serializers.IntegerField()
+    high = serializers.IntegerField()
+    medium = serializers.IntegerField()
+    low = serializers.IntegerField()
+    informational = serializers.IntegerField()
+    muted = serializers.IntegerField()
+    scan_ids = serializers.ListField(child=serializers.UUIDField())
+
+    class JSONAPIMeta:
+        resource_name = "findings-severity-over-time"
+
+
 class OverviewServiceSerializer(BaseSerializerV1):
     id = serializers.CharField(source="service")
     total = serializers.IntegerField()

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

@@ -102,6 +102,7 @@ from api.filters import (
     AttackSurfaceOverviewFilter,
     ComplianceOverviewFilter,
     CustomDjangoFilterBackend,
+    DailySeveritySummaryFilter,
     FindingFilter,
     IntegrationFilter,
     IntegrationJiraFindingsFilter,
@@ -131,6 +132,7 @@ from api.models import (
     AttackSurfaceOverview,
     ComplianceOverviewSummary,
     ComplianceRequirementOverview,
+    DailySeveritySummary,
     Finding,
     Integration,
     Invitation,
@@ -184,6 +186,7 @@ from api.v1.serializers import (
     FindingDynamicFilterSerializer,
     FindingMetadataSerializer,
     FindingSerializer,
+    FindingsSeverityOverTimeSerializer,
     IntegrationCreateSerializer,
     IntegrationJiraDispatchSerializer,
     IntegrationSerializer,
@@ -4009,6 +4012,16 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
         ),
         filters=True,
     ),
+    findings_severity_timeseries=extend_schema(
+        summary="Get findings severity data over time",
+        description=(
+            "Retrieve daily aggregated findings data grouped by severity levels over a date range. "
+            "Returns one data point per day with counts of failed findings by severity (critical, high, "
+            "medium, low, informational) and muted findings. Days without scans are filled forward with "
+            "the most recent known values. Use date_from (required) and date_to filters to specify the range."
+        ),
+        filters=True,
+    ),
     attack_surface=extend_schema(
         summary="Get attack surface overview",
         description="Retrieve aggregated attack surface metrics from latest completed scans per provider.",
@@ -4057,7 +4070,16 @@ class OverviewViewSet(BaseRLSViewSet):
         if not role.unlimited_visibility:
             self.allowed_providers = providers
 
-        return ScanSummary.all_objects.filter(tenant_id=self.request.tenant_id)
+        tenant_id = self.request.tenant_id
+
+        # Return appropriate queryset per action
+        if self.action == "findings_severity_timeseries":
+            qs = DailySeveritySummary.objects.filter(tenant_id=tenant_id)
+            if hasattr(self, "allowed_providers"):
+                qs = qs.filter(provider_id__in=self.allowed_providers)
+            return qs
+
+        return ScanSummary.all_objects.filter(tenant_id=tenant_id)
 
     def get_serializer_class(self):
         if self.action == "providers":
@@ -4068,6 +4090,8 @@ class OverviewViewSet(BaseRLSViewSet):
             return OverviewFindingSerializer
         elif self.action == "findings_severity":
             return OverviewSeveritySerializer
+        elif self.action == "findings_severity_timeseries":
+            return FindingsSeverityOverTimeSerializer
         elif self.action == "services":
             return OverviewServiceSerializer
         elif self.action == "regions":
@@ -4085,8 +4109,18 @@ class OverviewViewSet(BaseRLSViewSet):
             return ScanSummaryFilter
         elif self.action == "findings_severity":
             return ScanSummarySeverityFilter
+        elif self.action == "findings_severity_timeseries":
+            return DailySeveritySummaryFilter
         return None
 
+    def filter_queryset(self, queryset):
+        # Skip OrderingFilter for findings_severity_timeseries (no inserted_at field)
+        if self.action == "findings_severity_timeseries":
+            return CustomDjangoFilterBackend().filter_queryset(
+                self.request, queryset, self
+            )
+        return super().filter_queryset(queryset)
+
     @extend_schema(exclude=True)
     def list(self, request, *args, **kwargs):
         raise MethodNotAllowed(method="GET")
@@ -4363,6 +4397,108 @@ class OverviewViewSet(BaseRLSViewSet):
 
         return Response(serializer.data, status=status.HTTP_200_OK)
 
+    @action(
+        detail=False,
+        methods=["get"],
+        url_path="findings_severity/timeseries",
+        url_name="findings_severity_timeseries",
+    )
+    def findings_severity_timeseries(self, request):
+        """
+        Daily severity trends for charts. Uses DailySeveritySummary pre-aggregation.
+        Requires date_from filter.
+        """
+        # Get queryset with RBAC, provider, and date filters applied
+        # Date validation is handled by DailySeveritySummaryFilter
+        daily_qs = self.filter_queryset(self.get_queryset())
+
+        date_from = request._date_from
+        date_to = request._date_to
+
+        if not daily_qs.exists():
+            # No data matches filters - return zeros
+            result = self._generate_zero_result(date_from, date_to)
+            serializer = self.get_serializer(result, many=True)
+            return Response(serializer.data, status=status.HTTP_200_OK)
+
+        # Fetch all data for fill-forward logic
+        daily_summaries = list(
+            daily_qs.order_by("provider_id", "-date").values(
+                "provider_id",
+                "scan_id",
+                "date",
+                "critical",
+                "high",
+                "medium",
+                "low",
+                "informational",
+                "muted",
+            )
+        )
+
+        if not daily_summaries:
+            result = self._generate_zero_result(date_from, date_to)
+            serializer = self.get_serializer(result, many=True)
+            return Response(serializer.data, status=status.HTTP_200_OK)
+
+        # Build provider_data: {provider_id: [(date, data), ...]} sorted by date desc
+        provider_data = defaultdict(list)
+        for summary in daily_summaries:
+            provider_data[summary["provider_id"]].append(summary)
+
+        # For each day, find the latest data per provider and sum values
+        result = []
+        current_date = date_from
+        while current_date <= date_to:
+            day_totals = {
+                "critical": 0,
+                "high": 0,
+                "medium": 0,
+                "low": 0,
+                "informational": 0,
+                "muted": 0,
+            }
+            day_scan_ids = []
+
+            for provider_id, summaries in provider_data.items():
+                # Find the latest data for this provider <= current_date
+                for summary in summaries:  # Already sorted by date desc
+                    if summary["date"] <= current_date:
+                        day_totals["critical"] += summary["critical"] or 0
+                        day_totals["high"] += summary["high"] or 0
+                        day_totals["medium"] += summary["medium"] or 0
+                        day_totals["low"] += summary["low"] or 0
+                        day_totals["informational"] += summary["informational"] or 0
+                        day_totals["muted"] += summary["muted"] or 0
+                        day_scan_ids.append(summary["scan_id"])
+                        break  # Found the latest data for this provider
+
+            result.append(
+                {"date": current_date, "scan_ids": day_scan_ids, **day_totals}
+            )
+            current_date += timedelta(days=1)
+
+        serializer = self.get_serializer(result, many=True)
+        return Response(serializer.data, status=status.HTTP_200_OK)
+
+    def _generate_zero_result(self, date_from, date_to):
+        """Generate a list of zero-filled results for each date in range."""
+        result = []
+        current_date = date_from
+        zero_values = {
+            "critical": 0,
+            "high": 0,
+            "medium": 0,
+            "low": 0,
+            "informational": 0,
+            "muted": 0,
+            "scan_ids": [],
+        }
+        while current_date <= date_to:
+            result.append({"date": current_date, **zero_values})
+            current_date += timedelta(days=1)
+        return result
+
     @extend_schema(
         summary="Get ThreatScore snapshots",
         description=(

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

@@ -1,14 +1,18 @@
 from collections import defaultdict
 
+from django.db.models import Sum
+
 from api.db_router import READ_REPLICA_ALIAS
 from api.db_utils import rls_transaction
 from api.models import (
     ComplianceOverviewSummary,
     ComplianceRequirementOverview,
+    DailySeveritySummary,
     Resource,
     ResourceFindingMapping,
     ResourceScanSummary,
     Scan,
+    ScanSummary,
     StateChoices,
 )
 
@@ -175,3 +179,100 @@ def backfill_compliance_summaries(tenant_id: str, scan_id: str):
             )
 
     return {"status": "backfilled", "inserted": len(summary_objects)}
+
+
+def backfill_daily_severity_summaries(tenant_id: str, days: int = None):
+    """
+    Backfill DailySeveritySummary from completed scans.
+    Groups by provider+date, keeps latest scan per day.
+    """
+    from datetime import timedelta
+
+    from django.utils import timezone
+
+    created_count = 0
+    updated_count = 0
+
+    with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
+        scan_filter = {
+            "tenant_id": tenant_id,
+            "state": StateChoices.COMPLETED,
+            "completed_at__isnull": False,
+        }
+
+        if days is not None:
+            cutoff_date = timezone.now() - timedelta(days=days)
+            scan_filter["completed_at__gte"] = cutoff_date
+
+        completed_scans = (
+            Scan.objects.filter(**scan_filter)
+            .order_by("provider_id", "-completed_at")
+            .values("id", "provider_id", "completed_at")
+        )
+
+        if not completed_scans:
+            return {"status": "no scans to backfill"}
+
+        # Keep only latest scan per provider/day
+        latest_scans_by_day = {}
+        for scan in completed_scans:
+            key = (scan["provider_id"], scan["completed_at"].date())
+            if key not in latest_scans_by_day:
+                latest_scans_by_day[key] = scan
+
+    # Process each provider/day
+    for (provider_id, scan_date), scan in latest_scans_by_day.items():
+        scan_id = scan["id"]
+
+        with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
+            severity_totals = (
+                ScanSummary.objects.filter(
+                    tenant_id=tenant_id,
+                    scan_id=scan_id,
+                )
+                .values("severity")
+                .annotate(total_fail=Sum("fail"), total_muted=Sum("muted"))
+            )
+
+            severity_data = {
+                "critical": 0,
+                "high": 0,
+                "medium": 0,
+                "low": 0,
+                "informational": 0,
+                "muted": 0,
+            }
+
+            for row in severity_totals:
+                severity = row["severity"]
+                if severity in severity_data:
+                    severity_data[severity] = row["total_fail"] or 0
+                severity_data["muted"] += row["total_muted"] or 0
+
+        with rls_transaction(tenant_id):
+            _, created = DailySeveritySummary.objects.update_or_create(
+                tenant_id=tenant_id,
+                provider_id=provider_id,
+                date=scan_date,
+                defaults={
+                    "scan_id": scan_id,
+                    "critical": severity_data["critical"],
+                    "high": severity_data["high"],
+                    "medium": severity_data["medium"],
+                    "low": severity_data["low"],
+                    "informational": severity_data["informational"],
+                    "muted": severity_data["muted"],
+                },
+            )
+
+            if created:
+                created_count += 1
+            else:
+                updated_count += 1
+
+    return {
+        "status": "backfilled",
+        "created": created_count,
+        "updated": updated_count,
+        "total_days": len(latest_scans_by_day),
+    }

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

@@ -29,6 +29,7 @@ from api.models import (
     AttackSurfaceOverview,
     ComplianceOverviewSummary,
     ComplianceRequirementOverview,
+    DailySeveritySummary,
     Finding,
     MuteRule,
     Processor,
@@ -1348,3 +1349,72 @@ def aggregate_attack_surface(tenant_id: str, scan_id: str):
             )
     else:
         logger.info(f"No attack surface overview records created for scan {scan_id}")
+
+
+def aggregate_daily_severity(tenant_id: str, scan_id: str):
+    """Aggregate scan severity counts into DailySeveritySummary (one record per provider/day)."""
+    with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
+        scan = Scan.objects.filter(
+            tenant_id=tenant_id,
+            id=scan_id,
+            state=StateChoices.COMPLETED,
+        ).first()
+
+        if not scan:
+            logger.warning(f"Scan {scan_id} not found or not completed")
+            return {"status": "scan is not completed"}
+
+        provider_id = scan.provider_id
+        scan_date = scan.completed_at.date()
+
+        severity_totals = (
+            ScanSummary.objects.filter(
+                tenant_id=tenant_id,
+                scan_id=scan_id,
+            )
+            .values("severity")
+            .annotate(total_fail=Sum("fail"), total_muted=Sum("muted"))
+        )
+
+        severity_data = {
+            "critical": 0,
+            "high": 0,
+            "medium": 0,
+            "low": 0,
+            "informational": 0,
+            "muted": 0,
+        }
+
+        for row in severity_totals:
+            severity = row["severity"]
+            if severity in severity_data:
+                severity_data[severity] = row["total_fail"] or 0
+            severity_data["muted"] += row["total_muted"] or 0
+
+    with rls_transaction(tenant_id):
+        summary, created = DailySeveritySummary.objects.update_or_create(
+            tenant_id=tenant_id,
+            provider_id=provider_id,
+            date=scan_date,
+            defaults={
+                "scan_id": scan_id,
+                "critical": severity_data["critical"],
+                "high": severity_data["high"],
+                "medium": severity_data["medium"],
+                "low": severity_data["low"],
+                "informational": severity_data["informational"],
+                "muted": severity_data["muted"],
+            },
+        )
+
+    action = "created" if created else "updated"
+    logger.info(
+        f"Daily severity summary {action} for provider {provider_id} on {scan_date}"
+    )
+
+    return {
+        "status": action,
+        "provider_id": str(provider_id),
+        "date": str(scan_date),
+        "severity_data": severity_data,
+    }

api/src/backend/tasks/tasks.py

@@ -10,6 +10,7 @@ from config.django.base import DJANGO_FINDINGS_BATCH_SIZE, DJANGO_TMP_OUTPUT_DIR
 from django_celery_beat.models import PeriodicTask
 from tasks.jobs.backfill import (
     backfill_compliance_summaries,
+    backfill_daily_severity_summaries,
     backfill_resource_scan_summaries,
 )
 from tasks.jobs.connection import (
@@ -38,6 +39,7 @@ from tasks.jobs.muting import mute_historical_findings
 from tasks.jobs.report import generate_compliance_reports_job
 from tasks.jobs.scan import (
     aggregate_attack_surface,
+    aggregate_daily_severity,
     aggregate_findings,
     create_compliance_requirements,
     perform_prowler_scan,
@@ -75,8 +77,11 @@ def _perform_scan_complete_tasks(tenant_id: str, scan_id: str, provider_id: str)
     )
     chain(
         perform_scan_summary_task.si(tenant_id=tenant_id, scan_id=scan_id),
-        generate_outputs_task.si(
-            scan_id=scan_id, provider_id=provider_id, tenant_id=tenant_id
+        group(
+            aggregate_daily_severity_task.si(tenant_id=tenant_id, scan_id=scan_id),
+            generate_outputs_task.si(
+                scan_id=scan_id, provider_id=provider_id, tenant_id=tenant_id
+            ),
         ),
         group(
             # Use optimized task that generates both reports with shared queries
@@ -523,6 +528,12 @@ def backfill_compliance_summaries_task(tenant_id: str, scan_id: str):
     return backfill_compliance_summaries(tenant_id=tenant_id, scan_id=scan_id)
 
 
+@shared_task(name="backfill-daily-severity-summaries", queue="backfill")
+def backfill_daily_severity_summaries_task(tenant_id: str, days: int = None):
+    """Backfill DailySeveritySummary from historical scans. Use days param to limit scope."""
+    return backfill_daily_severity_summaries(tenant_id=tenant_id, days=days)
+
+
 @shared_task(base=RLSTask, name="scan-compliance-overviews", queue="compliance")
 @handle_provider_deletion
 def create_compliance_requirements_task(tenant_id: str, scan_id: str):
@@ -556,6 +567,13 @@ def aggregate_attack_surface_task(tenant_id: str, scan_id: str):
     return aggregate_attack_surface(tenant_id=tenant_id, scan_id=scan_id)
 
 
+@shared_task(name="scan-daily-severity", queue="overview")
+@handle_provider_deletion
+def aggregate_daily_severity_task(tenant_id: str, scan_id: str):
+    """Aggregate scan severity into DailySeveritySummary for findings_severity/timeseries endpoint."""
+    return aggregate_daily_severity(tenant_id=tenant_id, scan_id=scan_id)
+
+
 @shared_task(base=RLSTask, name="lighthouse-connection-check")
 @set_tenant
 def check_lighthouse_connection_task(lighthouse_config_id: str, tenant_id: str = None):