fix(threatscore): improve the way ThreatScore is calculated (#8582)

Co-authored-by: Sergio Garcia <hello@mistercloudsec.com>
Co-authored-by: alejandrobailo <alejandrobailo94@gmail.com>

api/src/backend/api/migrations/0049_compliancerequirementoverview_passed_failed_findings.py

@@ -0,0 +1,21 @@
+# Generated by Django 5.1.12 on 2025-10-07 10:41
+
+from django.db import migrations, models
+
+
+class Migration(migrations.Migration):
+    dependencies = [
+        ("api", "0048_api_key"),
+    ]
+    operations = [
+        migrations.AddField(
+            model_name="compliancerequirementoverview",
+            name="passed_findings",
+            field=models.IntegerField(default=0),
+        ),
+        migrations.AddField(
+            model_name="compliancerequirementoverview",
+            name="total_findings",
+            field=models.IntegerField(default=0),
+        ),
+    ]

api/src/backend/api/models.py

@@ -1293,6 +1293,8 @@ class ComplianceRequirementOverview(RowLevelSecurityProtectedModel):
     passed_checks = models.IntegerField(default=0)
     failed_checks = models.IntegerField(default=0)
     total_checks = models.IntegerField(default=0)
+    passed_findings = models.IntegerField(default=0)
+    total_findings = models.IntegerField(default=0)
 
     scan = models.ForeignKey(
         Scan,

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

@@ -1991,6 +1991,17 @@ class ComplianceOverviewDetailSerializer(serializers.Serializer):
         resource_name = "compliance-requirements-details"
 
 
+class ComplianceOverviewDetailThreatscoreSerializer(ComplianceOverviewDetailSerializer):
+    """
+    Serializer for detailed compliance requirement information for Threatscore.
+
+    Includes additional fields specific to the Threatscore framework.
+    """
+
+    passed_findings = serializers.IntegerField()
+    total_findings = serializers.IntegerField()
+
+
 class ComplianceOverviewAttributesSerializer(serializers.Serializer):
     id = serializers.CharField()
     compliance_name = serializers.CharField()

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

@@ -142,6 +142,7 @@ from api.v1.mixins import PaginateByPkMixin, TaskManagementMixin
 from api.v1.serializers import (
     ComplianceOverviewAttributesSerializer,
     ComplianceOverviewDetailSerializer,
+    ComplianceOverviewDetailThreatscoreSerializer,
     ComplianceOverviewMetadataSerializer,
     ComplianceOverviewSerializer,
     FindingDynamicFilterSerializer,
@@ -3438,7 +3439,12 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
 
         all_requirements = (
             filtered_queryset.values(
-                "requirement_id", "framework", "version", "description"
+                "requirement_id",
+                "framework",
+                "version",
+                "description",
+                "passed_findings",
+                "total_findings",
             )
             .distinct()
             .annotate(
@@ -3463,6 +3469,8 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
             total_instances = requirement["total_instances"]
             passed_count = passed_counts.get(requirement_id, 0)
             is_manual = requirement["manual_count"] == total_instances
+            passed_findings = requirement["passed_findings"]
+            total_findings = requirement["total_findings"]
             if is_manual:
                 requirement_status = "MANUAL"
             elif passed_count == total_instances:
@@ -3477,10 +3485,19 @@ class ComplianceOverviewViewSet(BaseRLSViewSet, TaskManagementMixin):
                     "version": requirement["version"],
                     "description": requirement["description"],
                     "status": requirement_status,
+                    "passed_findings": passed_findings,
+                    "total_findings": total_findings,
                 }
             )
 
-        serializer = self.get_serializer(requirements_summary, many=True)
+        # Use different serializer for threatscore framework
+        if "threatscore" not in compliance_id:
+            serializer = self.get_serializer(requirements_summary, many=True)
+        else:
+            serializer = ComplianceOverviewDetailThreatscoreSerializer(
+                requirements_summary, many=True
+            )
+
         return Response(serializer.data, status=status.HTTP_200_OK)
 
     @action(detail=False, methods=["get"], url_name="attributes")

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

@@ -590,7 +590,7 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
         # Get check status data by region from findings
         findings = (
             Finding.all_objects.filter(scan_id=scan_id, muted=False)
-            .only("id", "check_id", "status")
+            .only("id", "check_id", "status", "compliance")
             .prefetch_related(
                 Prefetch(
                     "resources",
@@ -601,7 +601,9 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
             .iterator(chunk_size=1000)
         )
 
+        findings_count_by_compliance = {}
         check_status_by_region = {}
+        modeled_threatscore_compliance_id = "ProwlerThreatScore-1.0"
         with rls_transaction(tenant_id):
             for finding in findings:
                 for resource in finding.small_resources:
@@ -609,6 +611,27 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                     current_status = check_status_by_region.setdefault(region, {})
                     if current_status.get(finding.check_id) != "FAIL":
                         current_status[finding.check_id] = finding.status
+                    if modeled_threatscore_compliance_id in finding.compliance:
+                        for requirement_id in finding.compliance[
+                            modeled_threatscore_compliance_id
+                        ]:
+                            compliance_key = findings_count_by_compliance.setdefault(
+                                region, {}
+                            ).setdefault(
+                                modeled_threatscore_compliance_id.lower().replace(
+                                    "-", ""
+                                ),
+                                {},
+                            )
+                            if requirement_id not in compliance_key:
+                                compliance_key[requirement_id] = {
+                                    "total": 0,
+                                    "pass": 0,
+                                }
+
+                            compliance_key[requirement_id]["total"] += 1
+                            if finding.status == "PASS":
+                                compliance_key[requirement_id]["pass"] += 1
 
         try:
             # Try to get regions from provider
@@ -644,6 +667,13 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
         compliance_requirement_objects = []
         for region, compliance_data in compliance_overview_by_region.items():
             for compliance_id, compliance in compliance_data.items():
+                modeled_framework = (
+                    compliance["framework"].lower().replace("-", "").replace("_", "")
+                )
+                modeled_version = (
+                    compliance["version"].lower().replace("-", "").replace("_", "")
+                )
+                modeled_compliance_id = f"{modeled_framework}{modeled_version}"
                 # Create an overview record for each requirement within each compliance framework
                 for requirement_id, requirement in compliance["requirements"].items():
                     compliance_requirement_objects.append(
@@ -660,9 +690,16 @@ def create_compliance_requirements(tenant_id: str, scan_id: str):
                             failed_checks=requirement["checks_status"]["fail"],
                             total_checks=requirement["checks_status"]["total"],
                             requirement_status=requirement["status"],
+                            passed_findings=findings_count_by_compliance.get(region, {})
+                            .get(modeled_compliance_id, {})
+                            .get(requirement_id, {})
+                            .get("pass", 0),
+                            total_findings=findings_count_by_compliance.get(region, {})
+                            .get(modeled_compliance_id, {})
+                            .get(requirement_id, {})
+                            .get("total", 0),
                         )
                     )
-
         # Bulk create requirement records
         create_objects_in_batches(
             tenant_id, ComplianceRequirementOverview, compliance_requirement_objects