feat(resources): latest and metadata endpoints and performance (#8112)

api/CHANGELOG.md

@@ -5,9 +5,16 @@ All notable changes to the **Prowler API** are documented in this file.
 ## [v1.10.0] (Prowler UNRELEASED)
 
 ### Added
-
 - SSO with SAML support [(#8175)](https://github.com/prowler-cloud/prowler/pull/8175)
+- `GET /resources/metadata`, `GET /resources/metadata/latest` and `GET /resources/latest` to expose resource metadata and latest scan results [(#8112)](https://github.com/prowler-cloud/prowler/pull/8112)
+
+### Changed
 - `/processors` endpoints to post-process findings. Currently, only the Mutelist processor is supported to allow to mute findings.
+- Optimized the underlying queries for resources endpoints [(#8112)](https://github.com/prowler-cloud/prowler/pull/8112)
+- Optimized include parameters for resources view [(#8229)](https://github.com/prowler-cloud/prowler/pull/8229)
+
+### Fixed
+- Search filter for findings and resources [(#8112)](https://github.com/prowler-cloud/prowler/pull/8112)
 
 ### Security
 

api/pyproject.toml

@@ -38,7 +38,7 @@ name = "prowler-api"
 package-mode = false
 # Needed for the SDK compatibility
 requires-python = ">=3.11,<3.13"
-version = "1.9.0"
+version = "1.10.0"
 
 [project.scripts]
 celery = "src.backend.config.settings.celery"

api/src/backend/api/filters.py

@@ -1,5 +1,6 @@
 from datetime import date, datetime, timedelta, timezone
 
+from dateutil.parser import parse
 from django.conf import settings
 from django.db.models import Q
 from django_filters.rest_framework import (
@@ -339,6 +340,8 @@ class ResourceFilter(ProviderRelationshipFilterSet):
     tags = CharFilter(method="filter_tag")
     inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
     updated_at = DateFilter(field_name="updated_at", lookup_expr="date")
+    scan = UUIDFilter(field_name="provider__scan", lookup_expr="exact")
+    scan__in = UUIDInFilter(field_name="provider__scan", lookup_expr="in")
 
     class Meta:
         model = Resource
@@ -353,6 +356,82 @@ class ResourceFilter(ProviderRelationshipFilterSet):
             "updated_at": ["gte", "lte"],
         }
 
+    def filter_queryset(self, queryset):
+        if not (self.data.get("scan") or self.data.get("scan__in")) and not (
+            self.data.get("updated_at")
+            or self.data.get("updated_at__date")
+            or self.data.get("updated_at__gte")
+            or self.data.get("updated_at__lte")
+        ):
+            raise ValidationError(
+                [
+                    {
+                        "detail": "At least one date filter is required: filter[updated_at], filter[updated_at.gte], "
+                        "or filter[updated_at.lte].",
+                        "status": 400,
+                        "source": {"pointer": "/data/attributes/updated_at"},
+                        "code": "required",
+                    }
+                ]
+            )
+
+        gte_date = (
+            parse(self.data.get("updated_at__gte")).date()
+            if self.data.get("updated_at__gte")
+            else datetime.now(timezone.utc).date()
+        )
+        lte_date = (
+            parse(self.data.get("updated_at__lte")).date()
+            if self.data.get("updated_at__lte")
+            else datetime.now(timezone.utc).date()
+        )
+
+        if abs(lte_date - gte_date) > timedelta(
+            days=settings.FINDINGS_MAX_DAYS_IN_RANGE
+        ):
+            raise ValidationError(
+                [
+                    {
+                        "detail": f"The date range cannot exceed {settings.FINDINGS_MAX_DAYS_IN_RANGE} days.",
+                        "status": 400,
+                        "source": {"pointer": "/data/attributes/updated_at"},
+                        "code": "invalid",
+                    }
+                ]
+            )
+
+        return super().filter_queryset(queryset)
+
+    def filter_tag_key(self, queryset, name, value):
+        return queryset.filter(Q(tags__key=value) | Q(tags__key__icontains=value))
+
+    def filter_tag_value(self, queryset, name, value):
+        return queryset.filter(Q(tags__value=value) | Q(tags__value__icontains=value))
+
+    def filter_tag(self, queryset, name, value):
+        # We won't know what the user wants to filter on just based on the value,
+        # and we don't want to build special filtering logic for every possible
+        # provider tag spec, so we'll just do a full text search
+        return queryset.filter(tags__text_search=value)
+
+
+class LatestResourceFilter(ProviderRelationshipFilterSet):
+    tag_key = CharFilter(method="filter_tag_key")
+    tag_value = CharFilter(method="filter_tag_value")
+    tag = CharFilter(method="filter_tag")
+    tags = CharFilter(method="filter_tag")
+
+    class Meta:
+        model = Resource
+        fields = {
+            "provider": ["exact", "in"],
+            "uid": ["exact", "icontains"],
+            "name": ["exact", "icontains"],
+            "region": ["exact", "icontains", "in"],
+            "service": ["exact", "icontains", "in"],
+            "type": ["exact", "icontains", "in"],
+        }
+
     def filter_tag_key(self, queryset, name, value):
         return queryset.filter(Q(tags__key=value) | Q(tags__key__icontains=value))
 

api/src/backend/api/migrations/0036_rfm_tenant_finding_index_partitions.py

@@ -0,0 +1,30 @@
+from functools import partial
+
+from django.db import migrations
+
+from api.db_utils import create_index_on_partitions, drop_index_on_partitions
+
+
+class Migration(migrations.Migration):
+    atomic = False
+
+    dependencies = [
+        ("api", "0035_finding_muted_reason"),
+    ]
+
+    operations = [
+        migrations.RunPython(
+            partial(
+                create_index_on_partitions,
+                parent_table="resource_finding_mappings",
+                index_name="rfm_tenant_finding_idx",
+                columns="tenant_id, finding_id",
+                method="BTREE",
+            ),
+            reverse_code=partial(
+                drop_index_on_partitions,
+                parent_table="resource_finding_mappings",
+                index_name="rfm_tenant_finding_idx",
+            ),
+        ),
+    ]

api/src/backend/api/migrations/0037_rfm_tenant_finding_index_parent.py

@@ -0,0 +1,17 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0036_rfm_tenant_finding_index_partitions"),
+    ]
+
+    operations = [
+        migrations.AddIndex(
+            model_name="resourcefindingmapping",
+            index=models.Index(
+                fields=["tenant_id", "finding_id"],
+                name="rfm_tenant_finding_idx",
+            ),
+        ),
+    ]

api/src/backend/api/migrations/0038_resource_failed_findings_count.py

@@ -0,0 +1,15 @@
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0037_rfm_tenant_finding_index_parent"),
+    ]
+
+    operations = [
+        migrations.AddField(
+            model_name="resource",
+            name="failed_findings_count",
+            field=models.IntegerField(default=0),
+        )
+    ]

api/src/backend/api/migrations/0039_resource_resources_failed_findings_idx.py

@@ -0,0 +1,20 @@
+from django.contrib.postgres.operations import AddIndexConcurrently
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    atomic = False
+
+    dependencies = [
+        ("api", "0038_resource_failed_findings_count"),
+    ]
+
+    operations = [
+        AddIndexConcurrently(
+            model_name="resource",
+            index=models.Index(
+                fields=["tenant_id", "-failed_findings_count", "id"],
+                name="resources_failed_findings_idx",
+            ),
+        ),
+    ]

api/src/backend/api/models.py

@@ -561,6 +561,8 @@ class Resource(RowLevelSecurityProtectedModel):
     details = models.TextField(blank=True, null=True)
     partition = models.TextField(blank=True, null=True)
 
+    failed_findings_count = models.IntegerField(default=0)
+
     # Relationships
     tags = models.ManyToManyField(
         ResourceTag,
@@ -607,6 +609,10 @@ class Resource(RowLevelSecurityProtectedModel):
                 fields=["tenant_id", "provider_id"],
                 name="resources_tenant_provider_idx",
             ),
+            models.Index(
+                fields=["tenant_id", "-failed_findings_count", "id"],
+                name="resources_failed_findings_idx",
+            ),
         ]
 
         constraints = [
@@ -849,6 +855,12 @@ class ResourceFindingMapping(PostgresPartitionedModel, RowLevelSecurityProtected
         #   - tenant_id
         #   - id
 
+        indexes = [
+            models.Index(
+                fields=["tenant_id", "finding_id"],
+                name="rfm_tenant_finding_idx",
+            ),
+        ]
         constraints = [
             models.UniqueConstraint(
                 fields=("tenant_id", "resource_id", "finding_id"),

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

@@ -2966,12 +2966,21 @@ class TestTaskViewSet:
 @pytest.mark.django_db
 class TestResourceViewSet:
     def test_resources_list_none(self, authenticated_client):
-        response = authenticated_client.get(reverse("resource-list"))
+        response = authenticated_client.get(
+            reverse("resource-list"), {"filter[updated_at]": TODAY}
+        )
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()["data"]) == 0
 
-    def test_resources_list(self, authenticated_client, resources_fixture):
+    def test_resources_list_no_date_filter(self, authenticated_client):
         response = authenticated_client.get(reverse("resource-list"))
+        assert response.status_code == status.HTTP_400_BAD_REQUEST
+        assert response.json()["errors"][0]["code"] == "required"
+
+    def test_resources_list(self, authenticated_client, resources_fixture):
+        response = authenticated_client.get(
+            reverse("resource-list"), {"filter[updated_at]": TODAY}
+        )
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()["data"]) == len(resources_fixture)
 
@@ -2992,7 +3001,8 @@ class TestResourceViewSet:
         findings_fixture,
     ):
         response = authenticated_client.get(
-            reverse("resource-list"), {"include": include_values}
+            reverse("resource-list"),
+            {"include": include_values, "filter[updated_at]": TODAY},
         )
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()["data"]) == len(resources_fixture)
@@ -3020,8 +3030,9 @@ class TestResourceViewSet:
                 ("region.icontains", "west", 1),
                 ("service", "ec2", 2),
                 ("service.icontains", "ec", 2),
-                ("inserted_at.gte", "2024-01-01 00:00:00", 3),
-                ("updated_at.lte", "2024-01-01 00:00:00", 0),
+                ("inserted_at.gte", today_after_n_days(-1), 3),
+                ("updated_at.gte", today_after_n_days(-1), 3),
+                ("updated_at.lte", today_after_n_days(1), 3),
                 ("type.icontains", "prowler", 2),
                 # provider filters
                 ("provider_type", "aws", 3),
@@ -3041,7 +3052,8 @@ class TestResourceViewSet:
                 ("tags", "multi word", 1),
                 # full text search on resource
                 ("search", "arn", 3),
-                ("search", "def1", 1),
+                # To improve search efficiency, full text search is not fully applicable
+                # ("search", "def1", 1),
                 # full text search on resource tags
                 ("search", "multi word", 1),
                 ("search", "key2", 2),
@@ -3056,14 +3068,42 @@ class TestResourceViewSet:
         filter_value,
         expected_count,
     ):
+        filters = {f"filter[{filter_name}]": filter_value}
+        if "updated_at" not in filter_name:
+            filters["filter[updated_at]"] = TODAY
         response = authenticated_client.get(
             reverse("resource-list"),
-            {f"filter[{filter_name}]": filter_value},
+            filters,
         )
 
         assert response.status_code == status.HTTP_200_OK
         assert len(response.json()["data"]) == expected_count
 
+    def test_resource_filter_by_scan_id(
+        self, authenticated_client, resources_fixture, scans_fixture
+    ):
+        response = authenticated_client.get(
+            reverse("resource-list"),
+            {"filter[scan]": scans_fixture[0].id},
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == 2
+
+    def test_resource_filter_by_scan_id_in(
+        self, authenticated_client, resources_fixture, scans_fixture
+    ):
+        response = authenticated_client.get(
+            reverse("resource-list"),
+            {
+                "filter[scan.in]": [
+                    scans_fixture[0].id,
+                    scans_fixture[1].id,
+                ]
+            },
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == 2
+
     def test_resource_filter_by_provider_id_in(
         self, authenticated_client, resources_fixture
     ):
@@ -3073,7 +3113,8 @@ class TestResourceViewSet:
                 "filter[provider.in]": [
                     resources_fixture[0].provider.id,
                     resources_fixture[1].provider.id,
-                ]
+                ],
+                "filter[updated_at]": TODAY,
             },
         )
         assert response.status_code == status.HTTP_200_OK
@@ -3110,13 +3151,13 @@ class TestResourceViewSet:
     )
     def test_resources_sort(self, authenticated_client, sort_field):
         response = authenticated_client.get(
-            reverse("resource-list"), {"sort": sort_field}
+            reverse("resource-list"), {"filter[updated_at]": TODAY, "sort": sort_field}
         )
         assert response.status_code == status.HTTP_200_OK
 
     def test_resources_sort_invalid(self, authenticated_client):
         response = authenticated_client.get(
-            reverse("resource-list"), {"sort": "invalid"}
+            reverse("resource-list"), {"filter[updated_at]": TODAY, "sort": "invalid"}
         )
         assert response.status_code == status.HTTP_400_BAD_REQUEST
         assert response.json()["errors"][0]["code"] == "invalid"
@@ -3149,6 +3190,100 @@ class TestResourceViewSet:
         )
         assert response.status_code == status.HTTP_404_NOT_FOUND
 
+    def test_resources_metadata_retrieve(
+        self, authenticated_client, resources_fixture, backfill_scan_metadata_fixture
+    ):
+        resource_1, *_ = resources_fixture
+        response = authenticated_client.get(
+            reverse("resource-metadata"),
+            {"filter[updated_at]": resource_1.updated_at.strftime("%Y-%m-%d")},
+        )
+        data = response.json()
+
+        expected_services = {"ec2", "s3"}
+        expected_regions = {"us-east-1", "eu-west-1"}
+        expected_resource_types = {"prowler-test"}
+
+        assert data["data"]["type"] == "resources-metadata"
+        assert data["data"]["id"] is None
+        assert set(data["data"]["attributes"]["services"]) == expected_services
+        assert set(data["data"]["attributes"]["regions"]) == expected_regions
+        assert set(data["data"]["attributes"]["types"]) == expected_resource_types
+
+    def test_resources_metadata_resource_filter_retrieve(
+        self, authenticated_client, resources_fixture, backfill_scan_metadata_fixture
+    ):
+        resource_1, *_ = resources_fixture
+        response = authenticated_client.get(
+            reverse("resource-metadata"),
+            {
+                "filter[region]": "eu-west-1",
+                "filter[updated_at]": resource_1.updated_at.strftime("%Y-%m-%d"),
+            },
+        )
+        data = response.json()
+
+        expected_services = {"s3"}
+        expected_regions = {"eu-west-1"}
+        expected_resource_types = {"prowler-test"}
+
+        assert data["data"]["type"] == "resources-metadata"
+        assert data["data"]["id"] is None
+        assert set(data["data"]["attributes"]["services"]) == expected_services
+        assert set(data["data"]["attributes"]["regions"]) == expected_regions
+        assert set(data["data"]["attributes"]["types"]) == expected_resource_types
+
+    def test_resources_metadata_future_date(self, authenticated_client):
+        response = authenticated_client.get(
+            reverse("resource-metadata"),
+            {"filter[updated_at]": "2048-01-01"},
+        )
+        data = response.json()
+        assert data["data"]["type"] == "resources-metadata"
+        assert data["data"]["id"] is None
+        assert data["data"]["attributes"]["services"] == []
+        assert data["data"]["attributes"]["regions"] == []
+        assert data["data"]["attributes"]["types"] == []
+
+    def test_resources_metadata_invalid_date(self, authenticated_client):
+        response = authenticated_client.get(
+            reverse("resource-metadata"),
+            {"filter[updated_at]": "2048-01-011"},
+        )
+        assert response.json() == {
+            "errors": [
+                {
+                    "detail": "Enter a valid date.",
+                    "status": "400",
+                    "source": {"pointer": "/data/attributes/updated_at"},
+                    "code": "invalid",
+                }
+            ]
+        }
+
+    def test_resources_latest(self, authenticated_client, latest_scan_resource):
+        response = authenticated_client.get(
+            reverse("resource-latest"),
+        )
+        assert response.status_code == status.HTTP_200_OK
+        assert len(response.json()["data"]) == 1
+        assert (
+            response.json()["data"][0]["attributes"]["uid"] == latest_scan_resource.uid
+        )
+
+    def test_resources_metadata_latest(
+        self, authenticated_client, latest_scan_resource
+    ):
+        response = authenticated_client.get(
+            reverse("resource-metadata_latest"),
+        )
+        assert response.status_code == status.HTTP_200_OK
+        attributes = response.json()["data"]["attributes"]
+
+        assert attributes["services"] == [latest_scan_resource.service]
+        assert attributes["regions"] == [latest_scan_resource.region]
+        assert attributes["types"] == [latest_scan_resource.type]
+
 
 @pytest.mark.django_db
 class TestFindingViewSet:
@@ -3247,7 +3382,7 @@ class TestFindingViewSet:
                 ("search", "dev-qa", 1),
                 ("search", "orange juice", 1),
                 # full text search on resource
-                ("search", "ec2", 2),
+                ("search", "ec2", 1),
                 # full text search on finding tags (disabled for now)
                 # ("search", "value2", 2),
                 # Temporary disabled until we implement tag filtering in the UI

api/src/backend/api/v1/mixins.py

@@ -24,20 +24,32 @@ class PaginateByPkMixin:
         request,  # noqa: F841
         base_queryset,
         manager,
-        select_related: list[str] | None = None,
-        prefetch_related: list[str] | None = None,
+        select_related: list | None = None,
+        prefetch_related: list | None = None,
     ) -> Response:
+        """
+        Paginate a queryset by primary key.
+
+        This method is useful when you want to paginate a queryset that has been
+        filtered or annotated in a way that would be lost if you used the default
+        pagination method.
+        """
         pk_list = base_queryset.values_list("id", flat=True)
         page = self.paginate_queryset(pk_list)
         if page is None:
             return Response(self.get_serializer(base_queryset, many=True).data)
 
         queryset = manager.filter(id__in=page)
+
         if select_related:
             queryset = queryset.select_related(*select_related)
         if prefetch_related:
             queryset = queryset.prefetch_related(*prefetch_related)
 
+        # Optimize tags loading, if applicable
+        if hasattr(self, "_optimize_tags_loading"):
+            queryset = self._optimize_tags_loading(queryset)
+
         queryset = sorted(queryset, key=lambda obj: page.index(obj.id))
 
         serialized = self.get_serializer(queryset, many=True).data

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

@@ -9,6 +9,7 @@ from drf_spectacular.utils import extend_schema_field
 from jwt.exceptions import InvalidKeyError
 from rest_framework.validators import UniqueTogetherValidator
 from rest_framework_json_api import serializers
+from rest_framework_json_api.relations import SerializerMethodResourceRelatedField
 from rest_framework_json_api.serializers import ValidationError
 from rest_framework_simplejwt.exceptions import TokenError
 from rest_framework_simplejwt.serializers import TokenObtainPairSerializer
@@ -999,8 +1000,12 @@ class ResourceSerializer(RLSSerializer):
 
     tags = serializers.SerializerMethodField()
     type_ = serializers.CharField(read_only=True)
+    failed_findings_count = serializers.IntegerField(read_only=True)
 
-    findings = serializers.ResourceRelatedField(many=True, read_only=True)
+    findings = SerializerMethodResourceRelatedField(
+        many=True,
+        read_only=True,
+    )
 
     class Meta:
         model = Resource
@@ -1016,6 +1021,7 @@ class ResourceSerializer(RLSSerializer):
             "tags",
             "provider",
             "findings",
+            "failed_findings_count",
             "url",
         ]
         extra_kwargs = {
@@ -1037,6 +1043,10 @@ class ResourceSerializer(RLSSerializer):
         }
     )
     def get_tags(self, obj):
+        # Use prefetched tags if available to avoid N+1 queries
+        if hasattr(obj, "prefetched_tags"):
+            return {tag.key: tag.value for tag in obj.prefetched_tags}
+        # Fallback to the original method if prefetch is not available
         return obj.get_tags(self.context.get("tenant_id"))
 
     def get_fields(self):
@@ -1046,6 +1056,13 @@ class ResourceSerializer(RLSSerializer):
         fields["type"] = type_
         return fields
 
+    def get_findings(self, obj):
+        return (
+            obj.latest_findings
+            if hasattr(obj, "latest_findings")
+            else obj.findings.all()
+        )
+
 
 class ResourceIncludeSerializer(RLSSerializer):
     """
@@ -1082,6 +1099,10 @@ class ResourceIncludeSerializer(RLSSerializer):
         }
     )
     def get_tags(self, obj):
+        # Use prefetched tags if available to avoid N+1 queries
+        if hasattr(obj, "prefetched_tags"):
+            return {tag.key: tag.value for tag in obj.prefetched_tags}
+        # Fallback to the original method if prefetch is not available
         return obj.get_tags(self.context.get("tenant_id"))
 
     def get_fields(self):
@@ -1092,6 +1113,17 @@ class ResourceIncludeSerializer(RLSSerializer):
         return fields
 
 
+class ResourceMetadataSerializer(serializers.Serializer):
+    services = serializers.ListField(child=serializers.CharField(), allow_empty=True)
+    regions = serializers.ListField(child=serializers.CharField(), allow_empty=True)
+    types = 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.")
+
+    class Meta:
+        resource_name = "resources-metadata"
+
+
 class FindingSerializer(RLSSerializer):
     """
     Serializer for the Finding model.

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

@@ -22,7 +22,7 @@ from django.conf import settings as django_settings
 from django.contrib.postgres.aggregates import ArrayAgg
 from django.contrib.postgres.search import SearchQuery
 from django.db import transaction
-from django.db.models import Count, Exists, F, OuterRef, Prefetch, Q, Sum
+from django.db.models import Count, F, Prefetch, Q, Sum
 from django.db.models.functions import Coalesce
 from django.http import HttpResponse
 from django.shortcuts import redirect
@@ -78,6 +78,7 @@ from api.filters import (
     IntegrationFilter,
     InvitationFilter,
     LatestFindingFilter,
+    LatestResourceFilter,
     MembershipFilter,
     ProcessorFilter,
     ProviderFilter,
@@ -108,6 +109,7 @@ from api.models import (
     Resource,
     ResourceFindingMapping,
     ResourceScanSummary,
+    ResourceTag,
     Role,
     RoleProviderGroupRelationship,
     SAMLConfiguration,
@@ -167,6 +169,7 @@ from api.v1.serializers import (
     ProviderSecretUpdateSerializer,
     ProviderSerializer,
     ProviderUpdateSerializer,
+    ResourceMetadataSerializer,
     ResourceSerializer,
     RoleCreateSerializer,
     RoleProviderGroupRelationshipSerializer,
@@ -290,7 +293,7 @@ class SchemaView(SpectacularAPIView):
 
     def get(self, request, *args, **kwargs):
         spectacular_settings.TITLE = "Prowler API"
-        spectacular_settings.VERSION = "1.9.0"
+        spectacular_settings.VERSION = "1.10.0"
         spectacular_settings.DESCRIPTION = (
             "Prowler API specification.\n\nThis file is auto-generated."
         )
@@ -1883,6 +1886,14 @@ class TaskViewSet(BaseRLSViewSet):
         summary="List all resources",
         description="Retrieve a list of all resources with options for filtering by various criteria. Resources are "
         "objects that are discovered by Prowler. They can be anything from a single host to a whole VPC.",
+        parameters=[
+            OpenApiParameter(
+                name="filter[updated_at]",
+                description="At least one of the variations of the `filter[updated_at]` filter must be provided.",
+                required=True,
+                type=OpenApiTypes.DATE,
+            )
+        ],
     ),
     retrieve=extend_schema(
         tags=["Resource"],
@@ -1890,15 +1901,43 @@ class TaskViewSet(BaseRLSViewSet):
         description="Fetch detailed information about a specific resource by their ID. A Resource is an object that "
         "is discovered by Prowler. It can be anything from a single host to a whole VPC.",
     ),
+    metadata=extend_schema(
+        tags=["Resource"],
+        summary="Retrieve metadata values from resources",
+        description="Fetch unique metadata values from a set of resources. This is useful for dynamic filtering.",
+        parameters=[
+            OpenApiParameter(
+                name="filter[updated_at]",
+                description="At least one of the variations of the `filter[updated_at]` filter must be provided.",
+                required=True,
+                type=OpenApiTypes.DATE,
+            )
+        ],
+        filters=True,
+    ),
+    latest=extend_schema(
+        tags=["Resource"],
+        summary="List the latest resources",
+        description="Retrieve a list of the latest resources from the latest scans for each provider with options for "
+        "filtering by various criteria.",
+        filters=True,
+    ),
+    metadata_latest=extend_schema(
+        tags=["Resource"],
+        summary="Retrieve metadata values from the latest resources",
+        description="Fetch unique metadata values from a set of resources from the latest scans for each provider. "
+        "This is useful for dynamic filtering.",
+        filters=True,
+    ),
 )
 @method_decorator(CACHE_DECORATOR, name="list")
 @method_decorator(CACHE_DECORATOR, name="retrieve")
-class ResourceViewSet(BaseRLSViewSet):
-    queryset = Resource.objects.all()
+class ResourceViewSet(PaginateByPkMixin, BaseRLSViewSet):
+    queryset = Resource.all_objects.all()
     serializer_class = ResourceSerializer
     http_method_names = ["get"]
     filterset_class = ResourceFilter
-    ordering = ["-inserted_at"]
+    ordering = ["-failed_findings_count", "-updated_at"]
     ordering_fields = [
         "provider_uid",
         "uid",
@@ -1909,6 +1948,14 @@ class ResourceViewSet(BaseRLSViewSet):
         "inserted_at",
         "updated_at",
     ]
+    prefetch_for_includes = {
+        "__all__": [],
+        "provider": [
+            Prefetch(
+                "provider", queryset=Provider.all_objects.select_related("resources")
+            )
+        ],
+    }
     # RBAC required permissions (implicit -> MANAGE_PROVIDERS enable unlimited visibility or check the visibility of
     # the provider through the provider group)
     required_permissions = []
@@ -1917,41 +1964,257 @@ class ResourceViewSet(BaseRLSViewSet):
         user_roles = get_role(self.request.user)
         if user_roles.unlimited_visibility:
             # User has unlimited visibility, return all scans
-            queryset = Resource.objects.filter(tenant_id=self.request.tenant_id)
+            queryset = Resource.all_objects.filter(tenant_id=self.request.tenant_id)
         else:
             # User lacks permission, filter providers based on provider groups associated with the role
-            queryset = Resource.objects.filter(
+            queryset = Resource.all_objects.filter(
                 tenant_id=self.request.tenant_id, provider__in=get_providers(user_roles)
             )
 
         search_value = self.request.query_params.get("filter[search]", None)
         if search_value:
-            # Django's ORM will build a LEFT JOIN and OUTER JOIN on the "through" table, resulting in duplicates
-            # The duplicates then require a `distinct` query
             search_query = SearchQuery(
                 search_value, config="simple", search_type="plain"
             )
             queryset = queryset.filter(
-                Q(tags__key=search_value)
-                | Q(tags__value=search_value)
-                | Q(tags__text_search=search_query)
-                | Q(tags__key__contains=search_value)
-                | Q(tags__value__contains=search_value)
-                | Q(uid=search_value)
-                | Q(name=search_value)
-                | Q(region=search_value)
-                | Q(service=search_value)
-                | Q(type=search_value)
-                | Q(text_search=search_query)
-                | Q(uid__contains=search_value)
-                | Q(name__contains=search_value)
-                | Q(region__contains=search_value)
-                | Q(service__contains=search_value)
-                | Q(type__contains=search_value)
+                Q(text_search=search_query) | Q(tags__text_search=search_query)
             ).distinct()
 
         return queryset
 
+    def _optimize_tags_loading(self, queryset):
+        """Optimize tags loading with prefetch_related to avoid N+1 queries"""
+        # Use prefetch_related to load all tags in a single query
+        return queryset.prefetch_related(
+            Prefetch(
+                "tags",
+                queryset=ResourceTag.objects.filter(
+                    tenant_id=self.request.tenant_id
+                ).select_related(),
+                to_attr="prefetched_tags",
+            )
+        )
+
+    def get_serializer_class(self):
+        if self.action in ["metadata", "metadata_latest"]:
+            return ResourceMetadataSerializer
+        return super().get_serializer_class()
+
+    def get_filterset_class(self):
+        if self.action in ["latest", "metadata_latest"]:
+            return LatestResourceFilter
+        return ResourceFilter
+
+    def filter_queryset(self, queryset):
+        # Do not apply filters when retrieving specific resource
+        if self.action == "retrieve":
+            return queryset
+        return super().filter_queryset(queryset)
+
+    def list(self, request, *args, **kwargs):
+        filtered_queryset = self.filter_queryset(self.get_queryset())
+        return self.paginate_by_pk(
+            request,
+            filtered_queryset,
+            manager=Resource.all_objects,
+            select_related=["provider"],
+            prefetch_related=["findings"],
+        )
+
+    def retrieve(self, request, *args, **kwargs):
+        queryset = self._optimize_tags_loading(self.get_queryset())
+        instance = get_object_or_404(queryset, pk=kwargs.get("pk"))
+        mapping_ids = list(
+            ResourceFindingMapping.objects.filter(
+                resource=instance, tenant_id=request.tenant_id
+            ).values_list("finding_id", flat=True)
+        )
+        latest_findings = (
+            Finding.all_objects.filter(id__in=mapping_ids, tenant_id=request.tenant_id)
+            .order_by("uid", "-inserted_at")
+            .distinct("uid")
+        )
+        setattr(instance, "latest_findings", latest_findings)
+        serializer = self.get_serializer(instance)
+        return Response(serializer.data, status=status.HTTP_200_OK)
+
+    @action(detail=False, methods=["get"], url_name="latest")
+    def latest(self, request):
+        tenant_id = request.tenant_id
+        filtered_queryset = self.filter_queryset(self.get_queryset())
+
+        latest_scan_ids = (
+            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
+            .order_by("provider_id", "-inserted_at")
+            .distinct("provider_id")
+            .values_list("id", flat=True)
+        )
+        filtered_queryset = filtered_queryset.filter(
+            tenant_id=tenant_id, provider__scan__in=latest_scan_ids
+        )
+
+        return self.paginate_by_pk(
+            request,
+            filtered_queryset,
+            manager=Resource.all_objects,
+            select_related=["provider"],
+            prefetch_related=["findings"],
+        )
+
+    @action(detail=False, methods=["get"], url_name="metadata")
+    def metadata(self, request):
+        # Force filter validation
+        self.filter_queryset(self.get_queryset())
+
+        tenant_id = request.tenant_id
+        query_params = request.query_params
+
+        queryset = ResourceScanSummary.objects.filter(tenant_id=tenant_id)
+
+        if scans := query_params.get("filter[scan__in]") or query_params.get(
+            "filter[scan]"
+        ):
+            queryset = queryset.filter(scan_id__in=scans.split(","))
+        else:
+            exact = query_params.get("filter[inserted_at]")
+            gte = query_params.get("filter[inserted_at__gte]")
+            lte = query_params.get("filter[inserted_at__lte]")
+
+            date_filters = {}
+            if exact:
+                date = parse_date(exact)
+                datetime_start = datetime.combine(
+                    date, datetime.min.time(), tzinfo=timezone.utc
+                )
+                datetime_end = datetime_start + timedelta(days=1)
+                date_filters["scan_id__gte"] = uuid7_start(
+                    datetime_to_uuid7(datetime_start)
+                )
+                date_filters["scan_id__lt"] = uuid7_start(
+                    datetime_to_uuid7(datetime_end)
+                )
+            else:
+                if gte:
+                    date_start = parse_date(gte)
+                    datetime_start = datetime.combine(
+                        date_start, datetime.min.time(), tzinfo=timezone.utc
+                    )
+                    date_filters["scan_id__gte"] = uuid7_start(
+                        datetime_to_uuid7(datetime_start)
+                    )
+                if lte:
+                    date_end = parse_date(lte)
+                    datetime_end = datetime.combine(
+                        date_end + timedelta(days=1),
+                        datetime.min.time(),
+                        tzinfo=timezone.utc,
+                    )
+                    date_filters["scan_id__lt"] = uuid7_start(
+                        datetime_to_uuid7(datetime_end)
+                    )
+
+            if date_filters:
+                queryset = queryset.filter(**date_filters)
+
+        if service_filter := query_params.get("filter[service]") or query_params.get(
+            "filter[service__in]"
+        ):
+            queryset = queryset.filter(service__in=service_filter.split(","))
+        if region_filter := query_params.get("filter[region]") or query_params.get(
+            "filter[region__in]"
+        ):
+            queryset = queryset.filter(region__in=region_filter.split(","))
+        if resource_type_filter := query_params.get("filter[type]") or query_params.get(
+            "filter[type__in]"
+        ):
+            queryset = queryset.filter(
+                resource_type__in=resource_type_filter.split(",")
+            )
+
+        services = list(
+            queryset.values_list("service", flat=True).distinct().order_by("service")
+        )
+        regions = list(
+            queryset.values_list("region", flat=True).distinct().order_by("region")
+        )
+        resource_types = list(
+            queryset.values_list("resource_type", flat=True)
+            .exclude(resource_type__isnull=True)
+            .exclude(resource_type__exact="")
+            .distinct()
+            .order_by("resource_type")
+        )
+
+        result = {
+            "services": services,
+            "regions": regions,
+            "types": resource_types,
+        }
+
+        serializer = self.get_serializer(data=result)
+        serializer.is_valid(raise_exception=True)
+        return Response(serializer.data)
+
+    @action(
+        detail=False,
+        methods=["get"],
+        url_name="metadata_latest",
+        url_path="metadata/latest",
+    )
+    def metadata_latest(self, request):
+        tenant_id = request.tenant_id
+        query_params = request.query_params
+
+        latest_scans_queryset = (
+            Scan.all_objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
+            .order_by("provider_id", "-inserted_at")
+            .distinct("provider_id")
+        )
+
+        queryset = ResourceScanSummary.objects.filter(
+            tenant_id=tenant_id,
+            scan_id__in=latest_scans_queryset.values_list("id", flat=True),
+        )
+
+        if service_filter := query_params.get("filter[service]") or query_params.get(
+            "filter[service__in]"
+        ):
+            queryset = queryset.filter(service__in=service_filter.split(","))
+        if region_filter := query_params.get("filter[region]") or query_params.get(
+            "filter[region__in]"
+        ):
+            queryset = queryset.filter(region__in=region_filter.split(","))
+        if resource_type_filter := query_params.get("filter[type]") or query_params.get(
+            "filter[type__in]"
+        ):
+            queryset = queryset.filter(
+                resource_type__in=resource_type_filter.split(",")
+            )
+
+        services = list(
+            queryset.values_list("service", flat=True).distinct().order_by("service")
+        )
+        regions = list(
+            queryset.values_list("region", flat=True).distinct().order_by("region")
+        )
+        resource_types = list(
+            queryset.values_list("resource_type", flat=True)
+            .exclude(resource_type__isnull=True)
+            .exclude(resource_type__exact="")
+            .distinct()
+            .order_by("resource_type")
+        )
+
+        result = {
+            "services": services,
+            "regions": regions,
+            "types": resource_types,
+        }
+
+        serializer = self.get_serializer(data=result)
+        serializer.is_valid(raise_exception=True)
+        return Response(serializer.data)
+
 
 @extend_schema_view(
     list=extend_schema(
@@ -2070,17 +2333,7 @@ class FindingViewSet(PaginateByPkMixin, BaseRLSViewSet):
                 search_value, config="simple", search_type="plain"
             )
 
-            resource_match = Resource.all_objects.filter(
-                text_search=search_query,
-                id__in=ResourceFindingMapping.objects.filter(
-                    resource_id=OuterRef("pk"),
-                    tenant_id=tenant_id,
-                ).values("resource_id"),
-            )
-
-            queryset = queryset.filter(
-                Q(text_search=search_query) | Q(Exists(resource_match))
-            )
+            queryset = queryset.filter(text_search=search_query)
 
         return queryset
 

api/src/backend/conftest.py

@@ -29,6 +29,7 @@ from api.models import (
     ProviderSecret,
     Resource,
     ResourceTag,
+    ResourceTagMapping,
     Role,
     SAMLConfiguration,
     SAMLDomainIndex,
@@ -654,6 +655,7 @@ def findings_fixture(scans_fixture, resources_fixture):
         check_metadata={
             "CheckId": "test_check_id",
             "Description": "test description apple sauce",
+            "servicename": "ec2",
         },
         first_seen_at="2024-01-02T00:00:00Z",
     )
@@ -680,6 +682,7 @@ def findings_fixture(scans_fixture, resources_fixture):
         check_metadata={
             "CheckId": "test_check_id",
             "Description": "test description orange juice",
+            "servicename": "s3",
         },
         first_seen_at="2024-01-02T00:00:00Z",
         muted=True,
@@ -1135,6 +1138,69 @@ def latest_scan_finding(authenticated_client, providers_fixture, resources_fixtu
     return finding
 
 
+@pytest.fixture(scope="function")
+def latest_scan_resource(authenticated_client, providers_fixture):
+    provider = providers_fixture[0]
+    tenant_id = str(providers_fixture[0].tenant_id)
+    scan = Scan.objects.create(
+        name="latest completed scan for resource",
+        provider=provider,
+        trigger=Scan.TriggerChoices.MANUAL,
+        state=StateChoices.COMPLETED,
+        tenant_id=tenant_id,
+    )
+    resource = Resource.objects.create(
+        tenant_id=tenant_id,
+        provider=provider,
+        uid="latest_resource_uid",
+        name="Latest Resource",
+        region="us-east-1",
+        service="ec2",
+        type="instance",
+        metadata='{"test": "metadata"}',
+        details='{"test": "details"}',
+    )
+
+    resource_tag = ResourceTag.objects.create(
+        tenant_id=tenant_id,
+        key="environment",
+        value="test",
+    )
+    ResourceTagMapping.objects.create(
+        tenant_id=tenant_id,
+        resource=resource,
+        tag=resource_tag,
+    )
+
+    finding = Finding.objects.create(
+        tenant_id=tenant_id,
+        uid="test_finding_uid_latest",
+        scan=scan,
+        delta="new",
+        status=Status.FAIL,
+        status_extended="test status extended ",
+        impact=Severity.critical,
+        impact_extended="test impact extended",
+        severity=Severity.critical,
+        raw_result={
+            "status": Status.FAIL,
+            "impact": Severity.critical,
+            "severity": Severity.critical,
+        },
+        tags={"test": "latest"},
+        check_id="test_check_id_latest",
+        check_metadata={
+            "CheckId": "test_check_id_latest",
+            "Description": "test description latest",
+        },
+        first_seen_at="2024-01-02T00:00:00Z",
+    )
+    finding.add_resources([resource])
+
+    backfill_resource_scan_summaries(tenant_id, str(scan.id))
+    return resource
+
+
 @pytest.fixture
 def saml_setup(tenants_fixture):
     tenant_id = tenants_fixture[0].id

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

@@ -6,7 +6,7 @@ from datetime import datetime, timezone
 from celery.utils.log import get_task_logger
 from config.settings.celery import CELERY_DEADLOCK_ATTEMPTS
 from django.db import IntegrityError, OperationalError
-from django.db.models import Case, Count, IntegerField, Sum, When
+from django.db.models import Case, Count, IntegerField, OuterRef, Subquery, Sum, When
 from tasks.utils import CustomEncoder
 
 from api.compliance import (
@@ -376,12 +376,16 @@ def perform_prowler_scan(
 def aggregate_findings(tenant_id: str, scan_id: str):
     """
     Aggregates findings for a given scan and stores the results in the ScanSummary table.
+    Also updates the failed_findings_count for each resource based on the latest findings.
 
     This function retrieves all findings associated with a given `scan_id` and calculates various
     metrics such as counts of failed, passed, and muted findings, as well as their deltas (new,
     changed, unchanged). The results are grouped by `check_id`, `service`, `severity`, and `region`.
     These aggregated metrics are then stored in the `ScanSummary` table.
 
+    Additionally, it updates the failed_findings_count field for each resource based on the most
+    recent findings for each finding.uid.
+
     Args:
         tenant_id (str): The ID of the tenant to which the scan belongs.
         scan_id (str): The ID of the scan for which findings need to be aggregated.
@@ -401,6 +405,8 @@ def aggregate_findings(tenant_id: str, scan_id: str):
         - muted_new: Muted findings with a delta of 'new'.
         - muted_changed: Muted findings with a delta of 'changed'.
     """
+    _update_resource_failed_findings_count(tenant_id, scan_id)
+
     with rls_transaction(tenant_id):
         findings = Finding.objects.filter(tenant_id=tenant_id, scan_id=scan_id)
 
@@ -525,6 +531,53 @@ def aggregate_findings(tenant_id: str, scan_id: str):
         ScanSummary.objects.bulk_create(scan_aggregations, batch_size=3000)
 
 
+def _update_resource_failed_findings_count(tenant_id: str, scan_id: str):
+    """
+    Update the failed_findings_count field for resources based on the latest findings.
+
+    This function calculates the number of failed findings for each resource by:
+    1. Getting the latest finding for each finding.uid
+    2. Counting failed findings per resource
+    3. Updating the failed_findings_count field for each resource
+
+    Args:
+        tenant_id (str): The ID of the tenant to which the scan belongs.
+        scan_id (str): The ID of the scan for which to update resource counts.
+    """
+
+    with rls_transaction(tenant_id):
+        scan = Scan.objects.get(pk=scan_id)
+        provider_id = scan.provider_id
+
+        resources = list(
+            Resource.all_objects.filter(tenant_id=tenant_id, provider_id=provider_id)
+        )
+
+    # For each resource, calculate failed findings count based on latest findings
+    for resource in resources:
+        with rls_transaction(tenant_id):
+            # Get the latest finding for each finding.uid that affects this resource
+            latest_findings_subquery = (
+                Finding.all_objects.filter(
+                    tenant_id=tenant_id, uid=OuterRef("uid"), resources=resource
+                )
+                .order_by("-inserted_at")
+                .values("id")[:1]
+            )
+
+            # Count failed findings from the latest findings
+            failed_count = Finding.all_objects.filter(
+                tenant_id=tenant_id,
+                resources=resource,
+                id__in=Subquery(latest_findings_subquery),
+                status=FindingStatus.FAIL,
+                muted=False,
+            ).count()
+
+            resource.failed_findings_count = failed_count
+            resource.save(update_fields=["failed_findings_count"])
+
+
 def create_compliance_requirements(tenant_id: str, scan_id: str):
     """
     Create detailed compliance requirement overview records for a scan.

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

@@ -7,6 +7,7 @@ import pytest
 from tasks.jobs.scan import (
     _create_finding_delta,
     _store_resources,
+    _update_resource_failed_findings_count,
     create_compliance_requirements,
     perform_prowler_scan,
 )
@@ -1166,3 +1167,85 @@ class TestCreateComplianceRequirements:
             assert len(req_2_objects) == 2
             assert all(obj.requirement_status == "PASS" for obj in req_1_objects)
             assert all(obj.requirement_status == "FAIL" for obj in req_2_objects)
+
+
+@pytest.mark.django_db
+class TestUpdateResourceFailedFindingsCount:
+    @patch("api.models.Resource.all_objects.filter")
+    @patch("api.models.Finding.all_objects.filter")
+    def test_failed_findings_count_update(
+        self,
+        mock_finding_filter,
+        mock_resource_filter,
+        tenants_fixture,
+        scans_fixture,
+        providers_fixture,
+    ):
+        tenant = tenants_fixture[0]
+        scan = scans_fixture[0]
+        provider = providers_fixture[0]
+
+        scan.provider = provider
+        scan.save()
+
+        tenant_id = str(tenant.id)
+        scan_id = str(scan.id)
+
+        resource1 = MagicMock()
+        resource1.uid = "res-1"
+        resource1.failed_findings_count = None
+        resource1.save = MagicMock()
+
+        resource2 = MagicMock()
+        resource2.uid = "res-2"
+        resource2.failed_findings_count = None
+        resource2.save = MagicMock()
+
+        mock_resource_filter.return_value = [resource1, resource2]
+
+        fake_subquery_qs = MagicMock()
+        fake_subquery_qs.order_by.return_value = fake_subquery_qs
+        fake_subquery_qs.values.return_value = fake_subquery_qs
+        fake_subquery_qs.__getitem__.return_value = fake_subquery_qs
+
+        def finding_filter_side_effect(*args, **kwargs):
+            if "status" in kwargs:
+                qs_count = MagicMock()
+                if kwargs.get("resources") == resource1:
+                    qs_count.count.return_value = 3
+                else:
+                    qs_count.count.return_value = 0
+                return qs_count
+            return fake_subquery_qs
+
+        mock_finding_filter.side_effect = finding_filter_side_effect
+
+        _update_resource_failed_findings_count(tenant_id, scan_id)
+
+        # resource1 should have been updated to 3
+        assert resource1.failed_findings_count == 3
+        resource1.save.assert_called_once_with(update_fields=["failed_findings_count"])
+
+        # resource2 should have been updated to 0
+        assert resource2.failed_findings_count == 0
+        resource2.save.assert_called_once_with(update_fields=["failed_findings_count"])
+
+    @patch("api.models.Resource.all_objects.filter", return_value=[])
+    @patch("api.models.Finding.all_objects.filter")
+    def test_no_resources_no_error(
+        self,
+        mock_finding_filter,
+        mock_resource_filter,
+        tenants_fixture,
+        scans_fixture,
+        providers_fixture,
+    ):
+        tenant = tenants_fixture[0]
+        scan = scans_fixture[0]
+        provider = providers_fixture[0]
+        scan.provider = provider
+        scan.save()
+
+        _update_resource_failed_findings_count(str(tenant.id), str(scan.id))
+
+        mock_finding_filter.assert_not_called()