fix(pdf): align ENS report requirement status (#10270)

This commit is contained in:
Pedro Martín
2026-03-06 12:36:50 +01:00
committed by GitHub
parent 1a6285c6a0
commit 86daf7bc05
4 changed files with 139 additions and 84 deletions

View File

@@ -29,6 +29,7 @@ All notable changes to the **Prowler API** are documented in this file.
### 🐞 Fixed
- PDF compliance reports consistency with UI: exclude resourceless findings and fix ENS MANUAL status handling [(#10270)](https://github.com/prowler-cloud/prowler/pull/10270)
- Attack Paths: Orphaned temporary Neo4j databases are now cleaned up on scan failure and provider deletion [(#10101)](https://github.com/prowler-cloud/prowler/pull/10101)
- Attack Paths: scan no longer raises `DatabaseError` when provider is deleted mid-scan [(#10116)](https://github.com/prowler-cloud/prowler/pull/10116)
- Tenant compliance summaries recalculated after provider deletion [(#10172)](https://github.com/prowler-cloud/prowler/pull/10172)

View File

@@ -336,7 +336,6 @@ class ENSReportGenerator(BaseComplianceReportGenerator):
for req in data.requirements:
if req.status == StatusChoices.MANUAL:
continue
m = get_requirement_metadata(req.id, data.attributes_by_requirement_id)
if m:
marco = getattr(m, "Marco", "Otros")
@@ -365,9 +364,12 @@ class ENSReportGenerator(BaseComplianceReportGenerator):
elements.append(Paragraph(f"{categoria_name}", self.styles["h3"]))
for req in reqs:
status_indicator = (
"" if req["status"] == StatusChoices.PASS else ""
)
if req["status"] == StatusChoices.PASS:
status_indicator = ""
elif req["status"] == StatusChoices.MANUAL:
status_indicator = ""
else:
status_indicator = ""
nivel_badge = f"[{req['nivel'].upper()}]" if req["nivel"] else ""
elements.append(
Paragraph(
@@ -841,11 +843,14 @@ class ENSReportGenerator(BaseComplianceReportGenerator):
elements.append(Spacer(1, 0.15 * inch))
# Status and Nivel badges row
status_color = COLOR_HIGH_RISK # FAIL
status_text = str(req.status).upper()
status_color = (
COLOR_HIGH_RISK if req.status == StatusChoices.FAIL else COLOR_GRAY
)
nivel_color = nivel_colors.get(nivel, COLOR_GRAY)
badges_row1 = [
["State:", "FAIL", "", f"Nivel: {nivel.upper()}"],
["State:", status_text, "", f"Nivel: {nivel.upper()}"],
]
badges_table1 = Table(
badges_row1,

View File

@@ -35,19 +35,27 @@ def _aggregate_requirement_statistics_from_database(
}
"""
requirement_statistics_by_check_id = {}
# TODO: take into account that now the relation is 1 finding == 1 resource, review this when the logic changes
with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
aggregated_statistics_queryset = (
Finding.all_objects.filter(
tenant_id=tenant_id, scan_id=scan_id, muted=False
tenant_id=tenant_id,
scan_id=scan_id,
muted=False,
resources__provider__is_deleted=False,
)
.values("check_id")
.annotate(
total_findings=Count(
"id",
distinct=True,
filter=Q(status__in=[StatusChoices.PASS, StatusChoices.FAIL]),
),
passed_findings=Count("id", filter=Q(status=StatusChoices.PASS)),
passed_findings=Count(
"id",
distinct=True,
filter=Q(status=StatusChoices.PASS),
),
)
)

View File

@@ -29,7 +29,7 @@ from tasks.jobs.threatscore_utils import (
_load_findings_for_requirement_checks,
)
from api.models import Finding, StatusChoices
from api.models import Finding, Resource, ResourceFindingMapping, StatusChoices
from prowler.lib.check.models import Severity
matplotlib.use("Agg") # Use non-interactive backend for tests
@@ -39,43 +39,50 @@ matplotlib.use("Agg") # Use non-interactive backend for tests
class TestAggregateRequirementStatistics:
"""Test suite for _aggregate_requirement_statistics_from_database function."""
def _create_finding_with_resource(
self, tenant, scan, uid, check_id, status, severity=Severity.high
):
"""Helper to create a finding linked to a resource (matching scan processing behavior)."""
finding = Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid=uid,
check_id=check_id,
status=status,
severity=severity,
impact=severity,
check_metadata={},
raw_result={},
)
resource = Resource.objects.create(
tenant_id=tenant.id,
provider=scan.provider,
uid=f"resource-{uid}",
name=f"resource-{uid}",
region="us-east-1",
service="test",
type="test::resource",
)
ResourceFindingMapping.objects.create(
tenant_id=tenant.id,
finding=finding,
resource=resource,
)
return finding
def test_aggregates_findings_correctly(self, tenants_fixture, scans_fixture):
"""Verify correct pass/total counts per check are aggregated from database."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-1",
check_id="check_1",
status=StatusChoices.PASS,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant, scan, "finding-1", "check_1", StatusChoices.PASS
)
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-2",
check_id="check_1",
status=StatusChoices.FAIL,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant, scan, "finding-2", "check_1", StatusChoices.FAIL
)
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-3",
check_id="check_2",
status=StatusChoices.PASS,
severity=Severity.medium,
impact=Severity.medium,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant, scan, "finding-3", "check_2", StatusChoices.PASS, Severity.medium
)
result = _aggregate_requirement_statistics_from_database(
@@ -106,27 +113,11 @@ class TestAggregateRequirementStatistics:
tenant = tenants_fixture[0]
scan = scans_fixture[0]
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-1",
check_id="check_1",
status=StatusChoices.FAIL,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant, scan, "finding-1", "check_1", StatusChoices.FAIL
)
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-2",
check_id="check_1",
status=StatusChoices.FAIL,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant, scan, "finding-2", "check_1", StatusChoices.FAIL
)
result = _aggregate_requirement_statistics_from_database(
@@ -142,16 +133,12 @@ class TestAggregateRequirementStatistics:
scan = scans_fixture[0]
for i in range(5):
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid=f"finding-{i}",
check_id="check_1",
status=StatusChoices.PASS if i % 2 == 0 else StatusChoices.FAIL,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant,
scan,
f"finding-{i}",
"check_1",
StatusChoices.PASS if i % 2 == 0 else StatusChoices.FAIL,
)
result = _aggregate_requirement_statistics_from_database(
@@ -162,27 +149,43 @@ class TestAggregateRequirementStatistics:
assert result["check_1"]["total"] == 5
def test_mixed_statuses(self, tenants_fixture, scans_fixture):
"""Verify MANUAL status is counted in total but not passed."""
"""Verify MANUAL status is not counted in total or passed."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-1",
check_id="check_1",
status=StatusChoices.PASS,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
self._create_finding_with_resource(
tenant, scan, "finding-1", "check_1", StatusChoices.PASS
)
self._create_finding_with_resource(
tenant, scan, "finding-2", "check_1", StatusChoices.MANUAL
)
result = _aggregate_requirement_statistics_from_database(
str(tenant.id), str(scan.id)
)
# MANUAL findings are excluded from the aggregation query
# since it only counts PASS and FAIL statuses
assert result["check_1"]["passed"] == 1
assert result["check_1"]["total"] == 1
def test_excludes_findings_without_resources(self, tenants_fixture, scans_fixture):
"""Verify findings without resources are excluded from aggregation."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
# Finding WITH resource → should be counted
self._create_finding_with_resource(
tenant, scan, "finding-1", "check_1", StatusChoices.PASS
)
# Finding WITHOUT resource → should be EXCLUDED
Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-2",
check_id="check_1",
status=StatusChoices.MANUAL,
status=StatusChoices.FAIL,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
@@ -193,8 +196,46 @@ class TestAggregateRequirementStatistics:
str(tenant.id), str(scan.id)
)
# MANUAL findings are excluded from the aggregation query
# since it only counts PASS and FAIL statuses
assert result["check_1"]["passed"] == 1
assert result["check_1"]["total"] == 1
def test_multiple_resources_no_double_count(self, tenants_fixture, scans_fixture):
"""Verify a finding with multiple resources is only counted once."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
finding = Finding.objects.create(
tenant_id=tenant.id,
scan=scan,
uid="finding-1",
check_id="check_1",
status=StatusChoices.PASS,
severity=Severity.high,
impact=Severity.high,
check_metadata={},
raw_result={},
)
# Link two resources to the same finding
for i in range(2):
resource = Resource.objects.create(
tenant_id=tenant.id,
provider=scan.provider,
uid=f"resource-{i}",
name=f"resource-{i}",
region="us-east-1",
service="test",
type="test::resource",
)
ResourceFindingMapping.objects.create(
tenant_id=tenant.id,
finding=finding,
resource=resource,
)
result = _aggregate_requirement_statistics_from_database(
str(tenant.id), str(scan.id)
)
assert result["check_1"]["passed"] == 1
assert result["check_1"]["total"] == 1