mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-10-29 14:54:30 +00:00
refactor(api): split finding-groups status from muted state (#10630)
This commit is contained in:
1 parent
63174caf98
commit
e4b2950436
11 files changed
+853
-93
No files matched your search
@@ -1803,7 +1803,12 @@ def aggregate_finding_group_summaries(tenant_id: str, scan_id: str):
|
||||
output_field=IntegerField(),
|
||||
)
|
||||
|
||||
# Aggregate findings by check_id for this scan
|
||||
# Aggregate findings by check_id for this scan.
|
||||
# `pass_count`, `fail_count` and `manual_count` count *every* finding
|
||||
# in this group, regardless of mute state, so the aggregated `status`
|
||||
# always reflects the underlying check outcome (FAIL > PASS > MANUAL)
|
||||
# even when the group is fully muted. The orthogonal `muted` flag is
|
||||
# what tells whether the group has any actionable (non-muted) findings.
|
||||
aggregated = (
|
||||
Finding.objects.filter(
|
||||
tenant_id=tenant_id,
|
||||
@@ -1812,11 +1817,52 @@ def aggregate_finding_group_summaries(tenant_id: str, scan_id: str):
|
||||
.values("check_id")
|
||||
.annotate(
|
||||
severity_order=Max(severity_case),
|
||||
pass_count=Count("id", filter=Q(status="PASS", muted=False)),
|
||||
fail_count=Count("id", filter=Q(status="FAIL", muted=False)),
|
||||
pass_count=Count("id", filter=Q(status="PASS")),
|
||||
fail_count=Count("id", filter=Q(status="FAIL")),
|
||||
manual_count=Count("id", filter=Q(status="MANUAL")),
|
||||
pass_muted_count=Count("id", filter=Q(status="PASS", muted=True)),
|
||||
fail_muted_count=Count("id", filter=Q(status="FAIL", muted=True)),
|
||||
manual_muted_count=Count("id", filter=Q(status="MANUAL", muted=True)),
|
||||
muted_count=Count("id", filter=Q(muted=True)),
|
||||
nonmuted_count=Count("id", filter=Q(muted=False)),
|
||||
new_count=Count("id", filter=Q(delta="new", muted=False)),
|
||||
changed_count=Count("id", filter=Q(delta="changed", muted=False)),
|
||||
new_fail_count=Count(
|
||||
"id", filter=Q(delta="new", status="FAIL", muted=False)
|
||||
),
|
||||
new_fail_muted_count=Count(
|
||||
"id", filter=Q(delta="new", status="FAIL", muted=True)
|
||||
),
|
||||
new_pass_count=Count(
|
||||
"id", filter=Q(delta="new", status="PASS", muted=False)
|
||||
),
|
||||
new_pass_muted_count=Count(
|
||||
"id", filter=Q(delta="new", status="PASS", muted=True)
|
||||
),
|
||||
new_manual_count=Count(
|
||||
"id", filter=Q(delta="new", status="MANUAL", muted=False)
|
||||
),
|
||||
new_manual_muted_count=Count(
|
||||
"id", filter=Q(delta="new", status="MANUAL", muted=True)
|
||||
),
|
||||
changed_fail_count=Count(
|
||||
"id", filter=Q(delta="changed", status="FAIL", muted=False)
|
||||
),
|
||||
changed_fail_muted_count=Count(
|
||||
"id", filter=Q(delta="changed", status="FAIL", muted=True)
|
||||
),
|
||||
changed_pass_count=Count(
|
||||
"id", filter=Q(delta="changed", status="PASS", muted=False)
|
||||
),
|
||||
changed_pass_muted_count=Count(
|
||||
"id", filter=Q(delta="changed", status="PASS", muted=True)
|
||||
),
|
||||
changed_manual_count=Count(
|
||||
"id", filter=Q(delta="changed", status="MANUAL", muted=False)
|
||||
),
|
||||
changed_manual_muted_count=Count(
|
||||
"id", filter=Q(delta="changed", status="MANUAL", muted=True)
|
||||
),
|
||||
resources_total=Count("resources__id", distinct=True),
|
||||
resources_fail=Count(
|
||||
"resources__id",
|
||||
@@ -1895,9 +1941,26 @@ def aggregate_finding_group_summaries(tenant_id: str, scan_id: str):
|
||||
severity_order=row["severity_order"] or 1,
|
||||
pass_count=row["pass_count"],
|
||||
fail_count=row["fail_count"],
|
||||
manual_count=row["manual_count"],
|
||||
pass_muted_count=row["pass_muted_count"],
|
||||
fail_muted_count=row["fail_muted_count"],
|
||||
manual_muted_count=row["manual_muted_count"],
|
||||
muted_count=row["muted_count"],
|
||||
muted=row["nonmuted_count"] == 0,
|
||||
new_count=row["new_count"],
|
||||
changed_count=row["changed_count"],
|
||||
new_fail_count=row["new_fail_count"],
|
||||
new_fail_muted_count=row["new_fail_muted_count"],
|
||||
new_pass_count=row["new_pass_count"],
|
||||
new_pass_muted_count=row["new_pass_muted_count"],
|
||||
new_manual_count=row["new_manual_count"],
|
||||
new_manual_muted_count=row["new_manual_muted_count"],
|
||||
changed_fail_count=row["changed_fail_count"],
|
||||
changed_fail_muted_count=row["changed_fail_muted_count"],
|
||||
changed_pass_count=row["changed_pass_count"],
|
||||
changed_pass_muted_count=row["changed_pass_muted_count"],
|
||||
changed_manual_count=row["changed_manual_count"],
|
||||
changed_manual_muted_count=row["changed_manual_muted_count"],
|
||||
resources_total=row["resources_total"],
|
||||
resources_fail=row["resources_fail"],
|
||||
first_seen_at=row["agg_first_seen_at"],
|
||||
@@ -1917,9 +1980,26 @@ def aggregate_finding_group_summaries(tenant_id: str, scan_id: str):
|
||||
"severity_order",
|
||||
"pass_count",
|
||||
"fail_count",
|
||||
"manual_count",
|
||||
"pass_muted_count",
|
||||
"fail_muted_count",
|
||||
"manual_muted_count",
|
||||
"muted_count",
|
||||
"muted",
|
||||
"new_count",
|
||||
"changed_count",
|
||||
"new_fail_count",
|
||||
"new_fail_muted_count",
|
||||
"new_pass_count",
|
||||
"new_pass_muted_count",
|
||||
"new_manual_count",
|
||||
"new_manual_muted_count",
|
||||
"changed_fail_count",
|
||||
"changed_fail_muted_count",
|
||||
"changed_pass_count",
|
||||
"changed_pass_muted_count",
|
||||
"changed_manual_count",
|
||||
"changed_manual_muted_count",
|
||||
"resources_total",
|
||||
"resources_fail",
|
||||
"first_seen_at",
|
||||
|
||||
@@ -771,26 +771,49 @@ def aggregate_finding_group_summaries_task(tenant_id: str, scan_id: str):
|
||||
)
|
||||
@set_tenant(keep_tenant=True)
|
||||
def reaggregate_all_finding_group_summaries_task(tenant_id: str):
|
||||
"""Reaggregate finding group summaries for all providers' latest completed scans."""
|
||||
latest_scan_ids = list(
|
||||
Scan.objects.filter(tenant_id=tenant_id, state=StateChoices.COMPLETED)
|
||||
.order_by("provider_id", "-completed_at", "-inserted_at")
|
||||
.distinct("provider_id")
|
||||
.values_list("id", flat=True)
|
||||
"""Reaggregate finding group summaries for every (provider, day) combination.
|
||||
|
||||
Mirrors the unbounded scope of `mute_historical_findings_task`: that task
|
||||
rewrites every Finding row whose UID matches a mute rule, with no time
|
||||
limit. To keep the daily summaries consistent with that update, this task
|
||||
re-runs the aggregator on the latest completed scan of every (provider,
|
||||
day) pair that exists in the database. Tasks are dispatched in parallel
|
||||
via a Celery group so the wallclock scales with the worker pool, not with
|
||||
the number of pairs.
|
||||
"""
|
||||
completed_scans = list(
|
||||
Scan.objects.filter(
|
||||
tenant_id=tenant_id,
|
||||
state=StateChoices.COMPLETED,
|
||||
completed_at__isnull=False,
|
||||
)
|
||||
.order_by("-completed_at")
|
||||
.values("id", "completed_at", "provider_id")
|
||||
)
|
||||
if latest_scan_ids:
|
||||
|
||||
# Keep the latest scan per (provider, day) pair so the daily summary row
|
||||
# the aggregator writes is the most recent snapshot of that day for that
|
||||
# provider. Iterating from most recent to oldest means the first scan we
|
||||
# see for a given key wins.
|
||||
latest_scans: dict[tuple, str] = {}
|
||||
for scan in completed_scans:
|
||||
key = (scan["provider_id"], scan["completed_at"].date())
|
||||
if key not in latest_scans:
|
||||
latest_scans[key] = str(scan["id"])
|
||||
|
||||
scan_ids = list(latest_scans.values())
|
||||
if scan_ids:
|
||||
logger.info(
|
||||
"Reaggregating finding group summaries for %d scans: %s",
|
||||
len(latest_scan_ids),
|
||||
latest_scan_ids,
|
||||
"Reaggregating finding group summaries for %d scans (provider x day)",
|
||||
len(scan_ids),
|
||||
)
|
||||
group(
|
||||
aggregate_finding_group_summaries_task.si(
|
||||
tenant_id=tenant_id, scan_id=str(scan_id)
|
||||
tenant_id=tenant_id, scan_id=scan_id
|
||||
)
|
||||
for scan_id in latest_scan_ids
|
||||
for scan_id in scan_ids
|
||||
).apply_async()
|
||||
return {"scans_reaggregated": len(latest_scan_ids)}
|
||||
return {"scans_reaggregated": len(scan_ids)}
|
||||
|
||||
|
||||
@shared_task(base=RLSTask, name="lighthouse-connection-check")
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
import uuid
|
||||
from contextlib import contextmanager
|
||||
from datetime import datetime, timezone
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import openai
|
||||
@@ -2362,35 +2362,96 @@ class TestReaggregateAllFindingGroupSummaries:
|
||||
@patch("tasks.tasks.group")
|
||||
@patch("tasks.tasks.aggregate_finding_group_summaries_task")
|
||||
@patch("tasks.tasks.Scan.objects.filter")
|
||||
def test_dispatches_subtasks_for_each_provider(
|
||||
def test_dispatches_subtasks_for_each_provider_per_day(
|
||||
self, mock_scan_filter, mock_agg_task, mock_group
|
||||
):
|
||||
scan_id_1 = uuid.uuid4()
|
||||
scan_id_2 = uuid.uuid4()
|
||||
provider_id_1 = uuid.uuid4()
|
||||
provider_id_2 = uuid.uuid4()
|
||||
scan_id_today_p1 = uuid.uuid4()
|
||||
scan_id_yesterday_p1 = uuid.uuid4()
|
||||
scan_id_today_p2 = uuid.uuid4()
|
||||
today = datetime.now(tz=timezone.utc)
|
||||
yesterday = today - timedelta(days=1)
|
||||
|
||||
mock_group_result = MagicMock()
|
||||
mock_group.side_effect = lambda gen: (list(gen), mock_group_result)[1]
|
||||
|
||||
mock_scan_filter.return_value.order_by.return_value.distinct.return_value.values_list.return_value = [
|
||||
scan_id_1,
|
||||
scan_id_2,
|
||||
mock_scan_filter.return_value.order_by.return_value.values.return_value = [
|
||||
{
|
||||
"id": scan_id_today_p1,
|
||||
"completed_at": today,
|
||||
"provider_id": provider_id_1,
|
||||
},
|
||||
{
|
||||
"id": scan_id_today_p2,
|
||||
"completed_at": today,
|
||||
"provider_id": provider_id_2,
|
||||
},
|
||||
{
|
||||
"id": scan_id_yesterday_p1,
|
||||
"completed_at": yesterday,
|
||||
"provider_id": provider_id_1,
|
||||
},
|
||||
]
|
||||
|
||||
result = reaggregate_all_finding_group_summaries_task(tenant_id=self.tenant_id)
|
||||
|
||||
assert result == {"scans_reaggregated": 2}
|
||||
assert mock_agg_task.si.call_count == 2
|
||||
assert result == {"scans_reaggregated": 3}
|
||||
assert mock_agg_task.si.call_count == 3
|
||||
mock_agg_task.si.assert_any_call(
|
||||
tenant_id=self.tenant_id, scan_id=str(scan_id_1)
|
||||
tenant_id=self.tenant_id, scan_id=str(scan_id_today_p1)
|
||||
)
|
||||
mock_agg_task.si.assert_any_call(
|
||||
tenant_id=self.tenant_id, scan_id=str(scan_id_2)
|
||||
tenant_id=self.tenant_id, scan_id=str(scan_id_today_p2)
|
||||
)
|
||||
mock_agg_task.si.assert_any_call(
|
||||
tenant_id=self.tenant_id, scan_id=str(scan_id_yesterday_p1)
|
||||
)
|
||||
mock_group_result.apply_async.assert_called_once()
|
||||
|
||||
@patch("tasks.tasks.group")
|
||||
@patch("tasks.tasks.aggregate_finding_group_summaries_task")
|
||||
@patch("tasks.tasks.Scan.objects.filter")
|
||||
def test_dedupes_scans_to_latest_per_provider_per_day(
|
||||
self, mock_scan_filter, mock_agg_task, mock_group
|
||||
):
|
||||
"""When several scans run on the same day for the same provider, only
|
||||
the latest one is dispatched (matching the daily summary unique key)."""
|
||||
provider_id = uuid.uuid4()
|
||||
latest_scan_today = uuid.uuid4()
|
||||
earlier_scan_today = uuid.uuid4()
|
||||
today_late = datetime.now(tz=timezone.utc)
|
||||
today_early = today_late - timedelta(hours=4)
|
||||
|
||||
mock_group_result = MagicMock()
|
||||
mock_group.side_effect = lambda gen: (list(gen), mock_group_result)[1]
|
||||
|
||||
# Returned ordered by `-completed_at`, so the most recent comes first.
|
||||
mock_scan_filter.return_value.order_by.return_value.values.return_value = [
|
||||
{
|
||||
"id": latest_scan_today,
|
||||
"completed_at": today_late,
|
||||
"provider_id": provider_id,
|
||||
},
|
||||
{
|
||||
"id": earlier_scan_today,
|
||||
"completed_at": today_early,
|
||||
"provider_id": provider_id,
|
||||
},
|
||||
]
|
||||
|
||||
result = reaggregate_all_finding_group_summaries_task(tenant_id=self.tenant_id)
|
||||
|
||||
assert result == {"scans_reaggregated": 1}
|
||||
mock_agg_task.si.assert_called_once_with(
|
||||
tenant_id=self.tenant_id, scan_id=str(latest_scan_today)
|
||||
)
|
||||
mock_group_result.apply_async.assert_called_once()
|
||||
|
||||
@patch("tasks.tasks.group")
|
||||
@patch("tasks.tasks.Scan.objects.filter")
|
||||
def test_no_completed_scans_skips_dispatch(self, mock_scan_filter, mock_group):
|
||||
mock_scan_filter.return_value.order_by.return_value.distinct.return_value.values_list.return_value = []
|
||||
mock_scan_filter.return_value.order_by.return_value.values.return_value = []
|
||||
|
||||
result = reaggregate_all_finding_group_summaries_task(tenant_id=self.tenant_id)
|
||||
|
||||
|
||||
Reference in new issue
Block a user