ben.carrasco/prowler
1
0
Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-07-23 20:42:02 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(mute-rules): Support simple muting in API (#9051)

Browse Source
This commit is contained in:
Víctor Fernández Poyatos
2025-10-31 10:49:17 +01:00
committed by GitHub
parent 5d840385df
commit 5d4415d090
15 changed files with 2874 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
api/CHANGELOG.md
+1
View File
@@ -10,6 +10,7 @@ All notable changes to the **Prowler API** are documented in this file.
- Support for configuring multiple LLM providers [(#8772)](https://github.com/prowler-cloud/prowler/pull/8772)
- Support C5 compliance framework for Azure provider [(#9081)](https://github.com/prowler-cloud/prowler/pull/9081)
- Support for Oracle Cloud Infrastructure (OCI) provider [(#8927)](https://github.com/prowler-cloud/prowler/pull/8927)
- Support muting findings based on simple rules with custom reason [(#9051)](https://github.com/prowler-cloud/prowler/pull/9051)
- Support C5 compliance framework for the GCP provider [(#9097)](https://github.com/prowler-cloud/prowler/pull/9097)
## [1.14.1] (Prowler 5.13.1)
api/src/backend/api/filters.py
+18
View File
@@ -30,6 +30,7 @@ from api.models import (
LighthouseProviderConfiguration,
LighthouseProviderModels,
Membership,
MuteRule,
OverviewStatusChoices,
PermissionChoices,
Processor,
@@ -980,3 +981,20 @@ class LighthouseProviderModelsFilter(FilterSet):
fields = {
"model_id": ["exact", "icontains", "in"],
}
class MuteRuleFilter(FilterSet):
inserted_at = DateFilter(field_name="inserted_at", lookup_expr="date")
updated_at = DateFilter(field_name="updated_at", lookup_expr="date")
created_by = UUIDFilter(field_name="created_by__id", lookup_expr="exact")
class Meta:
model = MuteRule
fields = {
"id": ["exact", "in"],
"name": ["exact", "icontains"],
"reason": ["icontains"],
"enabled": ["exact"],
"inserted_at": ["gte", "lte"],
"updated_at": ["gte", "lte"],
}
api/src/backend/api/migrations/0052_mute_rules.py
+117
View File
@@ -0,0 +1,117 @@
# Generated by Django 5.1.13 on 2025-10-22 11:56
import uuid
import django.contrib.postgres.fields
import django.core.validators
import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models
import api.rls
class Migration(migrations.Migration):
dependencies = [
("api", "0051_oraclecloud_provider"),
]
operations = [
migrations.CreateModel(
name="MuteRule",
fields=[
(
"id",
models.UUIDField(
default=uuid.uuid4,
editable=False,
primary_key=True,
serialize=False,
),
),
("inserted_at", models.DateTimeField(auto_now_add=True)),
("updated_at", models.DateTimeField(auto_now=True)),
(
"name",
models.CharField(
help_text="Human-readable name for this rule",
max_length=100,
validators=[django.core.validators.MinLengthValidator(3)],
),
),
(
"reason",
models.TextField(
help_text="Reason for muting",
max_length=500,
validators=[django.core.validators.MinLengthValidator(3)],
),
),
(
"enabled",
models.BooleanField(
default=True, help_text="Whether this rule is currently enabled"
),
),
(
"finding_uids",
django.contrib.postgres.fields.ArrayField(
base_field=models.CharField(max_length=255),
help_text="List of finding UIDs to mute",
size=None,
),
),
],
options={
"db_table": "mute_rules",
"abstract": False,
},
),
migrations.AddField(
model_name="finding",
name="muted_at",
field=models.DateTimeField(
blank=True, help_text="Timestamp when this finding was muted", null=True
),
),
migrations.AlterField(
model_name="tenantapikey",
name="name",
field=models.CharField(
max_length=100,
validators=[django.core.validators.MinLengthValidator(3)],
),
),
migrations.AddField(
model_name="muterule",
name="created_by",
field=models.ForeignKey(
help_text="User who created this rule",
null=True,
on_delete=django.db.models.deletion.SET_NULL,
related_name="created_mute_rules",
to=settings.AUTH_USER_MODEL,
),
),
migrations.AddField(
model_name="muterule",
name="tenant",
field=models.ForeignKey(
on_delete=django.db.models.deletion.CASCADE, to="api.tenant"
),
),
migrations.AddConstraint(
model_name="muterule",
constraint=api.rls.RowLevelSecurityConstraint(
"tenant_id",
name="rls_on_muterule",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
),
migrations.AddConstraint(
model_name="muterule",
constraint=models.UniqueConstraint(
fields=("tenant_id", "name"), name="unique_mute_rule_name_per_tenant"
),
),
]
api/src/backend/api/models.py
+56
View File
@@ -823,6 +823,9 @@ class Finding(PostgresPartitionedModel, RowLevelSecurityProtectedModel):
muted_reason = models.TextField(
blank=True, null=True, validators=[MinLengthValidator(3)], max_length=500
)
muted_at = models.DateTimeField(
null=True, blank=True, help_text="Timestamp when this finding was muted"
)
compliance = models.JSONField(default=dict, null=True, blank=True)
# Denormalize resource data for performance
@@ -1935,6 +1938,59 @@ class LighthouseConfiguration(RowLevelSecurityProtectedModel):
resource_name = "lighthouse-configurations"
class MuteRule(RowLevelSecurityProtectedModel):
id = models.UUIDField(primary_key=True, default=uuid4, editable=False)
inserted_at = models.DateTimeField(auto_now_add=True, editable=False)
updated_at = models.DateTimeField(auto_now=True, editable=False)
# Rule metadata
name = models.CharField(
max_length=100,
validators=[MinLengthValidator(3)],
help_text="Human-readable name for this rule",
)
reason = models.TextField(
validators=[MinLengthValidator(3)],
max_length=500,
help_text="Reason for muting",
)
enabled = models.BooleanField(
default=True, help_text="Whether this rule is currently enabled"
)
# Audit fields
created_by = models.ForeignKey(
User,
on_delete=models.SET_NULL,
null=True,
related_name="created_mute_rules",
help_text="User who created this rule",
)
# Rule criteria - array of finding UIDs
finding_uids = ArrayField(
models.CharField(max_length=255), help_text="List of finding UIDs to mute"
)
class Meta(RowLevelSecurityProtectedModel.Meta):
db_table = "mute_rules"
constraints = [
RowLevelSecurityConstraint(
field="tenant_id",
name="rls_on_%(class)s",
statements=["SELECT", "INSERT", "UPDATE", "DELETE"],
),
models.UniqueConstraint(
fields=("tenant_id", "name"),
name="unique_mute_rule_name_per_tenant",
),
]
class JSONAPIMeta:
resource_name = "mute-rules"
class Processor(RowLevelSecurityProtectedModel):
class ProcessorChoices(models.TextChoices):
MUTELIST = "mutelist", _("Mutelist")
api/src/backend/api/specs/v1.yaml
+602 -1
View File
@@ -4157,6 +4157,290 @@ paths:
task_args: null
metadata: null
description: ''
/api/v1/mute-rules:
get:
operationId: mute_rules_list
description: Retrieve a list of all mute rules with filtering options.
summary: List all mute rules
parameters:
- in: query
name: fields[mute-rules]
schema:
type: array
items:
type: string
enum:
- inserted_at
- updated_at
- name
- reason
- enabled
- created_by
- finding_uids
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[created_by]
schema:
type: string
format: uuid
- in: query
name: filter[enabled]
schema:
type: boolean
- in: query
name: filter[id]
schema:
type: string
format: uuid
- in: query
name: filter[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[inserted_at]
schema:
type: string
format: date
- in: query
name: filter[inserted_at__gte]
schema:
type: string
format: date-time
- in: query
name: filter[inserted_at__lte]
schema:
type: string
format: date-time
- in: query
name: filter[name]
schema:
type: string
- in: query
name: filter[name__icontains]
schema:
type: string
- in: query
name: filter[reason__icontains]
schema:
type: string
- name: filter[search]
required: false
in: query
description: A search term.
schema:
type: string
- in: query
name: filter[updated_at]
schema:
type: string
format: date
- in: query
name: filter[updated_at__gte]
schema:
type: string
format: date-time
- in: query
name: filter[updated_at__lte]
schema:
type: string
format: date-time
- in: query
name: include
schema:
type: array
items:
type: string
enum:
- created_by
description: include query parameter to allow the client to customize which
related resources should be returned.
explode: false
- name: page[number]
required: false
in: query
description: A page number within the paginated result set.
schema:
type: integer
- name: page[size]
required: false
in: query
description: Number of results to return per page.
schema:
type: integer
- 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:
- name
- -name
- enabled
- -enabled
- inserted_at
- -inserted_at
- updated_at
- -updated_at
explode: false
tags:
- Mute Rules
security:
- JWT or API Key: []
responses:
'200':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/PaginatedMuteRuleList'
description: ''
post:
operationId: mute_rules_create
description: Create a new mute rule by providing finding IDs, name, and reason.
The rule will immediately mute the selected findings and launch a background
task to mute all historical findings with matching UIDs.
summary: Create a new mute rule
tags:
- Mute Rules
requestBody:
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/MuteRuleCreateRequest'
application/x-www-form-urlencoded:
schema:
$ref: '#/components/schemas/MuteRuleCreateRequest'
multipart/form-data:
schema:
$ref: '#/components/schemas/MuteRuleCreateRequest'
required: true
security:
- JWT or API Key: []
responses:
'201':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/MuteRuleCreateResponse'
description: ''
/api/v1/mute-rules/{id}:
get:
operationId: mute_rules_retrieve
description: Fetch detailed information about a specific mute rule by ID.
summary: Retrieve a mute rule
parameters:
- in: query
name: fields[mute-rules]
schema:
type: array
items:
type: string
enum:
- inserted_at
- updated_at
- name
- reason
- enabled
- created_by
- finding_uids
description: endpoint return only specific fields in the response on a per-type
basis by including a fields[TYPE] query parameter.
explode: false
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this mute rule.
required: true
- in: query
name: include
schema:
type: array
items:
type: string
enum:
- created_by
description: include query parameter to allow the client to customize which
related resources should be returned.
explode: false
tags:
- Mute Rules
security:
- JWT or API Key: []
responses:
'200':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/MuteRuleResponse'
description: ''
patch:
operationId: mute_rules_partial_update
description: Update certain fields of an existing mute rule (e.g., name, reason,
enabled).
summary: Partially update a mute rule
parameters:
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this mute rule.
required: true
tags:
- Mute Rules
requestBody:
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/PatchedMuteRuleUpdateRequest'
application/x-www-form-urlencoded:
schema:
$ref: '#/components/schemas/PatchedMuteRuleUpdateRequest'
multipart/form-data:
schema:
$ref: '#/components/schemas/PatchedMuteRuleUpdateRequest'
required: true
security:
- JWT or API Key: []
responses:
'200':
content:
application/vnd.api+json:
schema:
$ref: '#/components/schemas/SerializerMetaclassResponse'
description: ''
delete:
operationId: mute_rules_destroy
description: 'Remove a mute rule from the system. Note: Previously muted findings
remain muted.'
summary: Delete a mute rule
parameters:
- in: path
name: id
schema:
type: string
format: uuid
description: A UUID string identifying this mute rule.
required: true
tags:
- Mute Rules
security:
- JWT or API Key: []
responses:
'204':
description: No response body
/api/v1/overviews/findings:
get:
operationId: overviews_findings_retrieve
@@ -12061,6 +12345,272 @@ components:
$ref: '#/components/schemas/Membership'
required:
- data
MuteRule:
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:
- mute-rules
id:
type: string
format: uuid
attributes:
type: object
properties:
inserted_at:
type: string
format: date-time
readOnly: true
updated_at:
type: string
format: date-time
readOnly: true
name:
type: string
description: Human-readable name for this rule
maxLength: 100
minLength: 3
reason:
type: string
description: Reason for muting
minLength: 3
maxLength: 500
enabled:
type: boolean
description: Whether this rule is currently enabled
finding_uids:
type: array
items:
type: string
readOnly: true
description: List of finding UIDs that are muted by this rule
required:
- name
- reason
relationships:
type: object
properties:
created_by:
type: object
properties:
data:
type: object
properties:
id:
type: string
format: uuid
type:
type: string
enum:
- users
title: Resource Type Name
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share common
attributes and relationships.
required:
- id
- type
required:
- data
description: The identifier of the related object.
title: Resource Identifier
nullable: true
MuteRuleCreate:
type: object
required:
- type
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:
- mute-rules
attributes:
type: object
properties:
inserted_at:
type: string
format: date-time
readOnly: true
updated_at:
type: string
format: date-time
readOnly: true
name:
type: string
description: Human-readable name for this rule
maxLength: 100
minLength: 3
reason:
type: string
description: Reason for muting
minLength: 3
maxLength: 500
enabled:
type: boolean
readOnly: true
description: Whether this rule is currently enabled
finding_ids:
type: array
items:
type: string
format: uuid
writeOnly: true
description: List of Finding IDs to mute (will be converted to UIDs)
finding_uids:
type: array
items:
type: string
readOnly: true
description: List of finding UIDs that are muted by this rule
required:
- name
- reason
- finding_ids
relationships:
type: object
properties:
created_by:
type: object
properties:
data:
type: object
properties:
id:
type: string
format: uuid
type:
type: string
enum:
- users
title: Resource Type Name
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share common
attributes and relationships.
required:
- id
- type
required:
- data
description: The identifier of the related object.
title: Resource Identifier
readOnly: true
nullable: true
MuteRuleCreateRequest:
type: object
properties:
data:
type: object
required:
- type
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:
- mute-rules
attributes:
type: object
properties:
inserted_at:
type: string
format: date-time
readOnly: true
updated_at:
type: string
format: date-time
readOnly: true
name:
type: string
minLength: 3
description: Human-readable name for this rule
maxLength: 100
reason:
type: string
minLength: 3
description: Reason for muting
maxLength: 500
enabled:
type: boolean
readOnly: true
description: Whether this rule is currently enabled
finding_ids:
type: array
items:
type: string
format: uuid
writeOnly: true
description: List of Finding IDs to mute (will be converted to UIDs)
finding_uids:
type: array
items:
type: string
minLength: 1
readOnly: true
description: List of finding UIDs that are muted by this rule
required:
- name
- reason
- finding_ids
relationships:
type: object
properties:
created_by:
type: object
properties:
data:
type: object
properties:
id:
type: string
format: uuid
type:
type: string
enum:
- users
title: Resource Type Name
description: The [type](https://jsonapi.org/format/#document-resource-object-identification)
member is used to describe resource objects that share
common attributes and relationships.
required:
- id
- type
required:
- data
description: The identifier of the related object.
title: Resource Identifier
readOnly: true
nullable: true
required:
- data
MuteRuleCreateResponse:
type: object
properties:
data:
$ref: '#/components/schemas/MuteRuleCreate'
required:
- data
MuteRuleResponse:
type: object
properties:
data:
$ref: '#/components/schemas/MuteRule'
required:
- data
OpenApiResponseResponse:
type: object
properties:
@@ -12399,6 +12949,15 @@ components:
$ref: '#/components/schemas/Membership'
required:
- data
PaginatedMuteRuleList:
type: object
properties:
data:
type: array
items:
$ref: '#/components/schemas/MuteRule'
required:
- data
PaginatedProcessorList:
type: object
properties:
@@ -12926,6 +13485,44 @@ components:
default_models: {}
required:
- data
PatchedMuteRuleUpdateRequest:
type: object
properties:
data:
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:
- mute-rules
id:
type: string
format: uuid
attributes:
type: object
properties:
name:
type: string
minLength: 3
description: Human-readable name for this rule
maxLength: 100
reason:
type: string
minLength: 3
description: Reason for muting
maxLength: 500
enabled:
type: boolean
description: Whether this rule is currently enabled
required:
- data
PatchedProcessorUpdateRequest:
type: object
properties:
@@ -17024,7 +17621,7 @@ components:
type: object
properties:
data:
$ref: '#/components/schemas/ProviderGroup'
$ref: '#/components/schemas/MuteRule'
required:
- data
Task:
@@ -18068,3 +18665,7 @@ tags:
- name: API Keys
description: Endpoints for API keys management. These can be used as an alternative
to JWT authorization.
- name: Mute Rules
description: Endpoints for simple mute rules management. These can be used as an
alternative to the Mutelist Processor if you need to mute specific findings across
your tenant with a specific reason.
api/src/backend/api/tests/test_views.py
+557
View File
@@ -34,6 +34,7 @@ from rest_framework.response import Response
from api.compliance import get_compliance_frameworks
from api.db_router import MainRouter
from api.models import (
Finding,
Integration,
Invitation,
LighthouseProviderConfiguration,
@@ -61,6 +62,8 @@ from api.models import (
from api.rls import Tenant
from api.v1.serializers import TokenSerializer
from api.v1.views import ComplianceOverviewViewSet, TenantFinishACSView
from prowler.lib.check.models import Severity
from prowler.lib.outputs.finding import Status
class TestViewSet:
@@ -9393,3 +9396,557 @@ class TestLighthouseProviderConfigViewSet:
# Unrelated entries should remain untouched
assert cfg.default_models.get("other") == "model-x"
@pytest.mark.django_db
class TestMuteRuleViewSet:
"""Tests for MuteRule endpoints."""
def test_mute_rules_list(self, authenticated_client, mute_rules_fixture):
"""Test listing all mute rules for the tenant."""
response = authenticated_client.get(reverse("mute-rule-list"))
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == len(mute_rules_fixture)
def test_mute_rules_list_empty(self, authenticated_client, tenants_fixture):
"""Test listing mute rules when none exist returns empty list."""
response = authenticated_client.get(reverse("mute-rule-list"))
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == 0
assert isinstance(data, list)
def test_mute_rules_list_default_ordering(
self, authenticated_client, mute_rules_fixture
):
"""Test that mute rules are ordered by -inserted_at by default."""
response = authenticated_client.get(reverse("mute-rule-list"))
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
if len(data) >= 2:
first_date = data[0]["attributes"]["inserted_at"]
second_date = data[1]["attributes"]["inserted_at"]
assert first_date >= second_date
def test_mute_rules_retrieve(self, authenticated_client, mute_rules_fixture):
"""Test retrieving a single mute rule by ID."""
mute_rule = mute_rules_fixture[0]
response = authenticated_client.get(
reverse("mute-rule-detail", kwargs={"pk": mute_rule.id})
)
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert data["id"] == str(mute_rule.id)
assert data["attributes"]["name"] == mute_rule.name
assert data["attributes"]["reason"] == mute_rule.reason
assert data["attributes"]["enabled"] == mute_rule.enabled
assert "finding_uids" in data["attributes"]
assert "inserted_at" in data["attributes"]
assert "updated_at" in data["attributes"]
def test_mute_rules_retrieve_invalid(self, authenticated_client):
"""Test retrieving non-existent mute rule returns 404."""
response = authenticated_client.get(
reverse(
"mute-rule-detail",
kwargs={"pk": "f498b103-c760-4785-9a3e-e23fafbb7b02"},
)
)
assert response.status_code == status.HTTP_404_NOT_FOUND
@pytest.mark.parametrize(
"filter_name, filter_value, expected_count",
(
[
("name", "Test Rule 1", 1),
("name.icontains", "rule", 2),
("reason.icontains", "security", 1),
("enabled", True, 1),
("enabled", False, 1),
]
),
)
def test_mute_rule_filters(
self,
authenticated_client,
mute_rules_fixture,
filter_name,
filter_value,
expected_count,
):
"""Test filtering mute rules by various fields."""
filters = {f"filter[{filter_name}]": filter_value}
response = authenticated_client.get(reverse("mute-rule-list"), filters)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) == expected_count
def test_mute_rule_filter_by_created_by(
self, authenticated_client, mute_rules_fixture, create_test_user
):
"""Test filtering mute rules by creator."""
response = authenticated_client.get(
reverse("mute-rule-list"),
{"filter[created_by]": create_test_user.id},
)
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == 2
def test_mute_rule_search(self, authenticated_client, mute_rules_fixture):
"""Test searching mute rules by name and reason."""
response = authenticated_client.get(
reverse("mute-rule-list"), {"filter[search]": "Rule 1"}
)
assert response.status_code == status.HTTP_200_OK
assert len(response.json()["data"]) == 1
@pytest.mark.parametrize(
"sort_field, first_index",
(
[
("name", 0),
("-name", 1),
("inserted_at", 0),
("-inserted_at", 1),
]
),
)
def test_mute_rule_ordering(
self, authenticated_client, mute_rules_fixture, sort_field, first_index
):
"""Test ordering mute rules by various fields."""
response = authenticated_client.get(
reverse("mute-rule-list"), {"sort": sort_field}
)
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == 2
assert data[0]["id"] == str(mute_rules_fixture[first_index].id)
@patch("tasks.tasks.mute_historical_findings_task.apply_async")
def test_mute_rules_create_valid(
self,
mock_task,
authenticated_client,
findings_fixture,
create_test_user,
):
"""Test creating a valid mute rule."""
finding_ids = [str(findings_fixture[0].id)]
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "New Mute Rule",
"reason": "Security exception approved",
"finding_ids": finding_ids,
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_201_CREATED
# Verify response contains the created mute rule
response_data = response.json()["data"]
assert response_data["type"] == "mute-rules"
assert response_data["attributes"]["name"] == "New Mute Rule"
assert response_data["attributes"]["reason"] == "Security exception approved"
# Verify the finding was immediately muted
from api.models import Finding
finding = Finding.objects.get(id=findings_fixture[0].id)
assert finding.muted is True
assert finding.muted_at is not None
assert finding.muted_reason == "Security exception approved"
# Verify background task was called
mock_task.assert_called_once()
@patch("tasks.tasks.mute_historical_findings_task.apply_async")
def test_mute_rules_create_converts_finding_ids_to_uids(
self,
mock_task,
authenticated_client,
findings_fixture,
):
"""Test that finding_ids are converted to finding UIDs."""
finding_ids = [str(findings_fixture[0].id), str(findings_fixture[1].id)]
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "UID Conversion Test",
"reason": "Testing UID conversion",
"finding_ids": finding_ids,
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_201_CREATED
# Verify finding_uids contains the UIDs, not IDs
from api.models import MuteRule
mute_rule = MuteRule.objects.get(name="UID Conversion Test")
expected_uids = [
findings_fixture[0].uid,
findings_fixture[1].uid,
]
assert set(mute_rule.finding_uids) == set(expected_uids)
@patch("tasks.tasks.mute_historical_findings_task.apply_async")
def test_mute_rules_deduplicates_uids(
self,
mock_task,
authenticated_client,
tenants_fixture,
providers_fixture,
scans_fixture,
):
"""Test that multiple findings with same UID result in only one UID in the rule."""
tenant = tenants_fixture[0]
scan = scans_fixture[0]
shared_uid = "prowler-aws-dedupe-test-001"
finding1 = Finding.objects.create(
tenant=tenant,
uid=shared_uid,
scan=scan,
status=Status.FAIL,
status_extended="test",
severity=Severity.high,
impact=Severity.high,
check_id="test_check",
check_metadata={"CheckId": "test_check"},
raw_result={},
)
finding2 = Finding.objects.create(
tenant=tenant,
uid=shared_uid,
scan=scan,
status=Status.FAIL,
status_extended="test",
severity=Severity.high,
impact=Severity.high,
check_id="test_check",
check_metadata={"CheckId": "test_check"},
raw_result={},
)
finding_ids = [str(finding1.id), str(finding2.id)]
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "Dedupe Test Rule",
"reason": "Testing UID deduplication",
"finding_ids": finding_ids,
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_201_CREATED
from api.models import MuteRule
mute_rule = MuteRule.objects.get(name="Dedupe Test Rule")
assert len(mute_rule.finding_uids) == 1
assert mute_rule.finding_uids[0] == shared_uid
finding1.refresh_from_db()
finding2.refresh_from_db()
assert finding1.muted is True
assert finding2.muted is True
@patch("tasks.tasks.mute_historical_findings_task.apply_async")
def test_mute_rules_create_overlap_detection_active(
self,
mock_task,
authenticated_client,
mute_rules_fixture,
findings_fixture,
):
"""Test that creating a rule with overlapping UIDs in active rule fails."""
# mute_rules_fixture[0] is active and has findings_fixture[0] UID
finding_ids = [str(findings_fixture[0].id)]
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "Overlapping Rule",
"reason": "This should fail",
"finding_ids": finding_ids,
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_409_CONFLICT
assert "errors" in response.json()
error_detail = response.json()["errors"][0]["detail"]
assert (
"already muted" in error_detail.lower() or "overlap" in error_detail.lower()
)
@patch("tasks.tasks.mute_historical_findings_task.apply_async")
def test_mute_rules_create_no_overlap_with_inactive(
self,
mock_task,
authenticated_client,
mute_rules_fixture,
findings_fixture,
):
"""Test that disabled rules don't prevent new rules with same UIDs."""
# mute_rules_fixture[1] is disabled
# Disable the enabled rule first
mute_rules_fixture[0].enabled = False
mute_rules_fixture[0].save()
finding_ids = [str(findings_fixture[0].id)]
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "Non-overlapping Rule",
"reason": "Inactive rules don't block",
"finding_ids": finding_ids,
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_201_CREATED
def test_mute_rules_create_invalid_empty_finding_ids(self, authenticated_client):
"""Test creating mute rule with empty finding_ids fails."""
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "Valid",
"reason": "Valid",
"finding_ids": [],
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "errors" in response.json()
assert (
response.json()["errors"][0]["source"]["pointer"]
== "/data/attributes/finding_ids"
)
@patch("tasks.tasks.mute_historical_findings_task.apply_async")
def test_mute_rules_create_invalid_finding_ids(
self, mock_task, authenticated_client
):
"""Test creating mute rule with non-existent finding IDs fails."""
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": "Invalid Findings",
"reason": "This should fail",
"finding_ids": ["f498b103-c760-4785-9a3e-e23fafbb7b02"],
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "errors" in response.json()
def test_mute_rules_create_duplicate_name(
self, authenticated_client, mute_rules_fixture
):
"""Test creating a mute rule with duplicate name fails."""
existing_name = mute_rules_fixture[0].name
data = {
"data": {
"type": "mute-rules",
"attributes": {
"name": existing_name,
"reason": "Duplicate name test",
"finding_ids": ["f498b103-c760-4785-9a3e-e23fafbb7b02"],
},
}
}
response = authenticated_client.post(
reverse("mute-rule-list"),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "errors" in response.json()
def test_mute_rules_update_name(self, authenticated_client, mute_rules_fixture):
"""Test updating mute rule name."""
mute_rule = mute_rules_fixture[0]
data = {
"data": {
"type": "mute-rules",
"id": str(mute_rule.id),
"attributes": {
"name": "Updated Name",
},
}
}
response = authenticated_client.patch(
reverse("mute-rule-detail", kwargs={"pk": mute_rule.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
response_data = response.json()["data"]
assert response_data["attributes"]["name"] == "Updated Name"
# Verify database was updated
mute_rule.refresh_from_db()
assert mute_rule.name == "Updated Name"
def test_mute_rules_update_reason(self, authenticated_client, mute_rules_fixture):
"""Test updating mute rule reason."""
mute_rule = mute_rules_fixture[0]
data = {
"data": {
"type": "mute-rules",
"id": str(mute_rule.id),
"attributes": {
"reason": "Updated reason for muting",
},
}
}
response = authenticated_client.patch(
reverse("mute-rule-detail", kwargs={"pk": mute_rule.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
response_data = response.json()["data"]
assert response_data["attributes"]["reason"] == "Updated reason for muting"
mute_rule.refresh_from_db()
assert mute_rule.reason == "Updated reason for muting"
def test_mute_rules_update_enabled(self, authenticated_client, mute_rules_fixture):
"""Test disabling a mute rule."""
mute_rule = mute_rules_fixture[0]
assert mute_rule.enabled is True
data = {
"data": {
"type": "mute-rules",
"id": str(mute_rule.id),
"attributes": {
"enabled": False,
},
}
}
response = authenticated_client.patch(
reverse("mute-rule-detail", kwargs={"pk": mute_rule.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_200_OK
response_data = response.json()["data"]
assert response_data["attributes"]["enabled"] is False
mute_rule.refresh_from_db()
assert mute_rule.enabled is False
def test_mute_rules_update_duplicate_name(
self, authenticated_client, mute_rules_fixture
):
"""Test updating mute rule with duplicate name fails."""
first_rule = mute_rules_fixture[0]
second_rule = mute_rules_fixture[1]
data = {
"data": {
"type": "mute-rules",
"id": str(second_rule.id),
"attributes": {
"name": first_rule.name,
},
}
}
response = authenticated_client.patch(
reverse("mute-rule-detail", kwargs={"pk": second_rule.id}),
data=json.dumps(data),
content_type="application/vnd.api+json",
)
assert response.status_code == status.HTTP_400_BAD_REQUEST
assert "errors" in response.json()
def test_mute_rules_delete(self, authenticated_client, mute_rules_fixture):
"""Test deleting a mute rule."""
mute_rule = mute_rules_fixture[0]
response = authenticated_client.delete(
reverse("mute-rule-detail", kwargs={"pk": mute_rule.id})
)
assert response.status_code == status.HTTP_204_NO_CONTENT
# Verify rule was deleted
from api.models import MuteRule
assert not MuteRule.objects.filter(id=mute_rule.id).exists()
def test_mute_rules_tenant_isolation(
self, authenticated_client, mute_rules_fixture, tenants_fixture
):
"""Test that users can only access mute rules from their tenant."""
# Create a second tenant with a mute rule
from api.models import MuteRule, Tenant
other_tenant = Tenant.objects.create(name="Other Tenant")
other_rule = MuteRule.objects.create(
tenant=other_tenant,
name="Other Tenant Rule",
reason="Should not be visible",
finding_uids=["test-uid"],
)
# Try to access other tenant's rule
response = authenticated_client.get(
reverse("mute-rule-detail", kwargs={"pk": other_rule.id})
)
assert response.status_code == status.HTTP_404_NOT_FOUND
# List should only show current tenant's rules
response = authenticated_client.get(reverse("mute-rule-list"))
assert response.status_code == status.HTTP_200_OK
data = response.json()["data"]
assert len(data) == len(mute_rules_fixture)
for rule_data in data:
assert rule_data["id"] != str(other_rule.id)
api/src/backend/api/v1/serializers.py
+174
View File
@@ -31,6 +31,7 @@ from api.models import (
LighthouseProviderModels,
LighthouseTenantConfiguration,
Membership,
MuteRule,
Processor,
Provider,
ProviderGroup,
@@ -3345,3 +3346,176 @@ class LighthouseProviderModelsUpdateSerializer(BaseWriteSerializer):
extra_kwargs = {
"id": {"read_only": True},
}
# Mute Rules
class MuteRuleSerializer(RLSSerializer):
"""
Serializer for reading MuteRule instances.
"""
finding_uids = serializers.ListField(
child=serializers.CharField(),
read_only=True,
help_text="List of finding UIDs that are muted by this rule",
)
class Meta:
model = MuteRule
fields = [
"id",
"inserted_at",
"updated_at",
"name",
"reason",
"enabled",
"created_by",
"finding_uids",
]
included_serializers = {
"created_by": "api.v1.serializers.UserIncludeSerializer",
}
class MuteRuleCreateSerializer(RLSSerializer, BaseWriteSerializer):
"""
Serializer for creating new MuteRule instances.
Accepts finding_ids in the request, converts them to UIDs, and stores in finding_uids.
"""
finding_ids = serializers.ListField(
child=serializers.UUIDField(),
write_only=True,
required=True,
help_text="List of Finding IDs to mute (will be converted to UIDs)",
)
finding_uids = serializers.ListField(
child=serializers.CharField(),
read_only=True,
help_text="List of finding UIDs that are muted by this rule",
)
class Meta:
model = MuteRule
fields = [
"id",
"inserted_at",
"updated_at",
"name",
"reason",
"enabled",
"created_by",
"finding_ids",
"finding_uids",
]
extra_kwargs = {
"id": {"read_only": True},
"inserted_at": {"read_only": True},
"updated_at": {"read_only": True},
"enabled": {"read_only": True},
"created_by": {"read_only": True},
}
def validate_name(self, value):
"""Validate that the name is unique within the tenant."""
tenant_id = self.context.get("tenant_id")
if MuteRule.objects.filter(tenant_id=tenant_id, name=value).exists():
raise ValidationError("A mute rule with this name already exists.")
return value
def validate_finding_ids(self, value):
"""Validate that all finding IDs exist and belong to the tenant."""
if not value:
raise ValidationError("At least one finding_id must be provided.")
tenant_id = self.context.get("tenant_id")
# Check that all findings exist and belong to this tenant
findings = Finding.all_objects.filter(tenant_id=tenant_id, id__in=value)
found_ids = set(findings.values_list("id", flat=True))
provided_ids = set(value)
missing_ids = provided_ids - found_ids
if missing_ids:
raise ValidationError(
f"The following finding IDs do not exist or do not belong to your tenant: {missing_ids}"
)
return value
def validate(self, data):
"""Validate the entire mute rule, including overlap detection."""
data = super().validate(data)
tenant_id = self.context.get("tenant_id")
finding_ids = data.get("finding_ids", [])
if not finding_ids:
return data
# Convert finding IDs to UIDs (deduplicate in case multiple findings have same UID)
findings = Finding.all_objects.filter(id__in=finding_ids, tenant_id=tenant_id)
finding_uids = list(set(findings.values_list("uid", flat=True)))
# Check for overlaps with existing enabled rules
existing_rules = MuteRule.objects.filter(tenant_id=tenant_id, enabled=True)
for rule in existing_rules:
overlap = set(finding_uids) & set(rule.finding_uids)
if overlap:
raise ConflictException(
detail=f"The following finding UIDs are already muted by rule '{rule.name}': {overlap}"
)
# Store finding_uids in validated_data for create
data["finding_uids"] = finding_uids
return data
def create(self, validated_data):
"""Create a new mute rule and set created_by."""
# Remove finding_ids from validated_data (we've already converted to finding_uids)
validated_data.pop("finding_ids", None)
# Set created_by to the current user
request = self.context.get("request")
if request and hasattr(request, "user"):
validated_data["created_by"] = request.user
return super().create(validated_data)
class MuteRuleUpdateSerializer(BaseWriteSerializer):
"""
Serializer for updating MuteRule instances.
"""
class Meta:
model = MuteRule
fields = [
"id",
"name",
"reason",
"enabled",
]
extra_kwargs = {
"id": {"read_only": True},
"name": {"required": False},
"reason": {"required": False},
"enabled": {"required": False},
}
def validate_name(self, value):
"""Validate that the name is unique within the tenant, excluding current instance."""
tenant_id = self.context.get("tenant_id")
if (
MuteRule.objects.filter(tenant_id=tenant_id, name=value)
.exclude(id=self.instance.id)
.exists()
):
raise ValidationError("A mute rule with this name already exists.")
return value
api/src/backend/api/v1/urls.py
+2 -1
View File
@@ -21,6 +21,7 @@ from api.v1.views import (
LighthouseProviderModelsViewSet,
LighthouseTenantConfigViewSet,
MembershipViewSet,
MuteRuleViewSet,
OverviewViewSet,
ProcessorViewSet,
ProviderGroupProvidersRelationshipView,
@@ -80,6 +81,7 @@ router.register(
LighthouseProviderModelsViewSet,
basename="lighthouse-models",
)
router.register(r"mute-rules", MuteRuleViewSet, basename="mute-rule")
tenants_router = routers.NestedSimpleRouter(router, r"tenants", lookup="tenant")
tenants_router.register(
@@ -150,7 +152,6 @@ urlpatterns = [
),
name="provider_group-providers-relationship",
),
# Lighthouse tenant config as singleton endpoint
path(
"lighthouse/configuration",
LighthouseTenantConfigViewSet.as_view(
api/src/backend/api/v1/views.py
+104
View File
@@ -66,6 +66,7 @@ from tasks.tasks import (
delete_provider_task,
delete_tenant_task,
jira_integration_task,
mute_historical_findings_task,
perform_scan_task,
refresh_lighthouse_provider_models_task,
)
@@ -90,6 +91,7 @@ from api.filters import (
LighthouseProviderConfigFilter,
LighthouseProviderModelsFilter,
MembershipFilter,
MuteRuleFilter,
ProcessorFilter,
ProviderFilter,
ProviderGroupFilter,
@@ -115,6 +117,7 @@ from api.models import (
LighthouseProviderModels,
LighthouseTenantConfiguration,
Membership,
MuteRule,
Processor,
Provider,
ProviderGroup,
@@ -175,6 +178,9 @@ from api.v1.serializers import (
LighthouseTenantConfigSerializer,
LighthouseTenantConfigUpdateSerializer,
MembershipSerializer,
MuteRuleCreateSerializer,
MuteRuleSerializer,
MuteRuleUpdateSerializer,
OverviewFindingSerializer,
OverviewProviderCountSerializer,
OverviewProviderSerializer,
@@ -414,6 +420,12 @@ class SchemaView(SpectacularAPIView):
"description": "Endpoints for API keys management. These can be used as an alternative to JWT "
"authorization.",
},
{
"name": "Mute Rules",
"description": "Endpoints for simple mute rules management. These can be used as an alternative to the"
" Mutelist Processor if you need to mute specific findings across your tenant with a "
"specific reason.",
},
]
return super().get(request, *args, **kwargs)
@@ -4705,3 +4717,95 @@ class TenantApiKeyViewSet(BaseRLSViewSet):
serializer = self.get_serializer(instance)
return Response(data=serializer.data, status=status.HTTP_200_OK)
# MuteRules
@extend_schema_view(
list=extend_schema(
tags=["Mute Rules"],
summary="List all mute rules",
description="Retrieve a list of all mute rules with filtering options.",
),
retrieve=extend_schema(
tags=["Mute Rules"],
summary="Retrieve a mute rule",
description="Fetch detailed information about a specific mute rule by ID.",
),
create=extend_schema(
tags=["Mute Rules"],
summary="Create a new mute rule",
description="Create a new mute rule by providing finding IDs, name, and reason. "
"The rule will immediately mute the selected findings and launch a background task "
"to mute all historical findings with matching UIDs.",
request=MuteRuleCreateSerializer,
),
partial_update=extend_schema(
tags=["Mute Rules"],
summary="Partially update a mute rule",
description="Update certain fields of an existing mute rule (e.g., name, reason, enabled).",
request=MuteRuleUpdateSerializer,
responses={200: MuteRuleSerializer},
),
destroy=extend_schema(
tags=["Mute Rules"],
summary="Delete a mute rule",
description="Remove a mute rule from the system. Note: Previously muted findings remain muted.",
),
)
class MuteRuleViewSet(BaseRLSViewSet):
queryset = MuteRule.objects.all()
serializer_class = MuteRuleSerializer
filterset_class = MuteRuleFilter
http_method_names = ["get", "post", "patch", "delete"]
search_fields = ["name", "reason"]
ordering = ["-inserted_at"]
ordering_fields = [
"name",
"enabled",
"inserted_at",
"updated_at",
]
required_permissions = [Permissions.MANAGE_SCANS]
def get_queryset(self):
queryset = MuteRule.objects.filter(tenant_id=self.request.tenant_id)
return queryset.select_related("created_by")
def get_serializer_class(self):
if self.action == "create":
return MuteRuleCreateSerializer
elif self.action == "partial_update":
return MuteRuleUpdateSerializer
return super().get_serializer_class()
def create(self, request, *args, **kwargs):
serializer = self.get_serializer(data=request.data)
serializer.is_valid(raise_exception=True)
# Create the mute rule
mute_rule = serializer.save()
tenant_id = str(request.tenant_id)
finding_ids = request.data.get("finding_ids", [])
# Immediately mute the selected findings
Finding.all_objects.filter(
id__in=finding_ids, tenant_id=tenant_id, muted=False
).update(
muted=True,
muted_at=mute_rule.inserted_at,
muted_reason=mute_rule.reason,
)
# Launch background task for historical muting
with transaction.atomic():
mute_historical_findings_task.apply_async(
kwargs={"tenant_id": tenant_id, "mute_rule_id": str(mute_rule.id)}
)
# Return the created mute rule
serializer = self.get_serializer(mute_rule)
return Response(
data=serializer.data,
status=status.HTTP_201_CREATED,
)
api/src/backend/conftest.py
+29
View File
@@ -23,6 +23,7 @@ from api.models import (
Invitation,
LighthouseConfiguration,
Membership,
MuteRule,
Processor,
Provider,
ProviderGroup,
@@ -1425,6 +1426,34 @@ def api_keys_fixture(tenants_fixture, create_test_user):
return [api_key1, api_key2, api_key3]
@pytest.fixture
def mute_rules_fixture(tenants_fixture, create_test_user, findings_fixture):
"""Create test mute rules for testing."""
tenant = tenants_fixture[0]
user = create_test_user
# Create two mute rules: one enabled, one disabled
mute_rule1 = MuteRule.objects.create(
tenant_id=tenant.id,
name="Test Rule 1",
reason="Security exception for testing",
enabled=True,
created_by=user,
finding_uids=[findings_fixture[0].uid],
)
mute_rule2 = MuteRule.objects.create(
tenant_id=tenant.id,
name="Test Rule 2",
reason="Compliance exception approved",
enabled=False,
created_by=user,
finding_uids=[findings_fixture[1].uid],
)
return mute_rule1, mute_rule2
def get_authorization_header(access_token: str) -> dict:
return {"Authorization": f"Bearer {access_token}"}
api/src/backend/tasks/jobs/muting.py
+64
View File
@@ -0,0 +1,64 @@
from celery.utils.log import get_task_logger
from config.django.base import DJANGO_FINDINGS_BATCH_SIZE
from tasks.utils import batched
from api.db_utils import rls_transaction
from api.models import Finding, MuteRule
logger = get_task_logger(__name__)
def mute_historical_findings(tenant_id: str, mute_rule_id: str):
"""
Mute historical findings that match the given mute rule.
This function processes findings in batches, updating their muted status
and adding the mute reason.
Args:
tenant_id (str): The tenant ID for RLS context
mute_rule_id (str): The ID of the mute rule to apply
Returns:
dict: Summary of the muting operation with findings_muted count
"""
findings_muted_count = 0
# Get the list of UIDs to mute and the reason
with rls_transaction(tenant_id):
mute_rule = MuteRule.objects.get(id=mute_rule_id, tenant_id=tenant_id)
finding_uids = mute_rule.finding_uids
mute_reason = mute_rule.reason
muted_at = mute_rule.inserted_at
# Query findings that match the UIDs and are not already muted
with rls_transaction(tenant_id):
findings_to_mute = Finding.objects.filter(
tenant_id=tenant_id, uid__in=finding_uids, muted=False
)
total_findings = findings_to_mute.count()
logger.info(
f"Processing {total_findings} findings for mute rule {mute_rule_id}"
)
if total_findings > 0:
for batch, is_last in batched(
findings_to_mute.iterator(), DJANGO_FINDINGS_BATCH_SIZE
):
batch_ids = [f.id for f in batch]
updated_count = Finding.all_objects.filter(
id__in=batch_ids, tenant_id=tenant_id
).update(
muted=True,
muted_at=muted_at,
muted_reason=mute_reason,
)
findings_muted_count += updated_count
logger.info(f"Muted {findings_muted_count} findings for rule {mute_rule_id}")
return {
"findings_muted": findings_muted_count,
"rule_id": mute_rule_id,
}
api/src/backend/tasks/jobs/scan.py
+31 -4
View File
@@ -30,6 +30,7 @@ from api.exceptions import ProviderConnectionError
from api.models import (
ComplianceRequirementOverview,
Finding,
MuteRule,
Processor,
Provider,
Resource,
@@ -301,6 +302,20 @@ def perform_prowler_scan(
logger.error(f"Error processing mutelist rules: {e}")
mutelist_processor = None
# Load enabled mute rules for this tenant
with rls_transaction(tenant_id, using=READ_REPLICA_ALIAS):
try:
active_mute_rules = MuteRule.objects.filter(
tenant_id=tenant_id, enabled=True
).values_list("finding_uids", "reason")
mute_rules_cache = {}
for finding_uids, reason in active_mute_rules:
for uid in finding_uids:
mute_rules_cache[uid] = reason
except Exception as e:
logger.error(f"Error loading mute rules: {e}")
mute_rules_cache = {}
try:
with rls_transaction(tenant_id):
try:
@@ -449,11 +464,22 @@ def perform_prowler_scan(
if not last_first_seen_at:
last_first_seen_at = datetime.now(tz=timezone.utc)
# If the finding is muted at this time the reason must be the configured Mutelist
muted_reason = "Muted by mutelist" if finding.muted else None
# Determine if finding should be muted and why
# Priority: mutelist processor (highest) > manual mute rules
is_muted = False
muted_reason = None
# Check mutelist processor first (highest priority)
if finding.muted:
is_muted = True
muted_reason = "Muted by mutelist"
# If not muted by mutelist, check manual mute rules
elif finding_uid in mute_rules_cache:
is_muted = True
muted_reason = mute_rules_cache[finding_uid]
# Increment failed_findings_count cache if the finding status is FAIL and not muted
if status == FindingStatus.FAIL and not finding.muted:
if status == FindingStatus.FAIL and not is_muted:
resource_uid = finding.resource_uid
resource_failed_findings_cache[resource_uid] += 1
@@ -472,7 +498,8 @@ def perform_prowler_scan(
check_id=finding.check_id,
scan=scan_instance,
first_seen_at=last_first_seen_at,
muted=finding.muted,
muted=is_muted,
muted_at=datetime.now(tz=timezone.utc) if is_muted else None,
muted_reason=muted_reason,
compliance=finding.compliance,
)
api/src/backend/tasks/tasks.py
+23
View File
@@ -31,6 +31,7 @@ from tasks.jobs.lighthouse_providers import (
check_lighthouse_provider_connection,
refresh_lighthouse_provider_models,
)
from tasks.jobs.muting import mute_historical_findings
from tasks.jobs.report import generate_threatscore_report_job
from tasks.jobs.scan import (
aggregate_findings,
@@ -681,3 +682,25 @@ def generate_threatscore_report_task(tenant_id: str, scan_id: str, provider_id:
return generate_threatscore_report_job(
tenant_id=tenant_id, scan_id=scan_id, provider_id=provider_id
)
@shared_task(name="findings-mute-historical")
def mute_historical_findings_task(tenant_id: str, mute_rule_id: str):
"""
Background task to mute all historical findings matching a mute rule.
This task processes findings in batches to avoid memory issues with large datasets.
It updates the Finding.muted, Finding.muted_at, and Finding.muted_reason fields
for all findings whose UID is in the mute rule's finding_uids list.
Args:
tenant_id (str): The tenant ID for RLS context.
mute_rule_id (str): The primary key of the MuteRule to apply.
Returns:
dict: A dictionary containing:
- 'findings_muted' (int): Total number of findings muted.
- 'rule_id' (str): The mute rule ID.
- 'status' (str): Final status ('completed').
"""
return mute_historical_findings(tenant_id, mute_rule_id)
api/src/backend/tasks/tests/test_muting.py
+532
View File
@@ -0,0 +1,532 @@
from datetime import datetime, timezone
from uuid import uuid4
import pytest
from django.core.exceptions import ObjectDoesNotExist
from tasks.jobs.muting import mute_historical_findings
from api.models import Finding, MuteRule
from prowler.lib.check.models import Severity
from prowler.lib.outputs.finding import Status
@pytest.mark.django_db
class TestMuteHistoricalFindings:
"""
Test suite for the mute_historical_findings function.
This class tests the batch processing of findings to update their muted status
based on MuteRule criteria.
"""
@pytest.fixture(scope="function")
def test_user(self, create_test_user):
"""Create a test user for mute rule creation."""
return create_test_user
@pytest.fixture(scope="function")
def mute_rule_with_findings(self, tenants_fixture, findings_fixture, test_user):
"""
Create a mute rule that targets the first finding in the fixture.
"""
tenant = tenants_fixture[0]
finding = findings_fixture[0]
mute_rule = MuteRule.objects.create(
tenant_id=tenant.id,
name="Test Mute Rule",
reason="Testing mute functionality",
enabled=True,
created_by=test_user,
finding_uids=[finding.uid],
)
return mute_rule
@pytest.fixture(scope="function")
def mute_rule_multiple_findings(self, scans_fixture, test_user):
"""
Create multiple unmuted findings and a mute rule targeting all of them.
"""
scan = scans_fixture[0]
tenant_id = scan.tenant_id
# Create 5 unmuted findings
finding_uids = []
for i in range(5):
finding = Finding.objects.create(
tenant_id=tenant_id,
uid=f"test_finding_uid_mute_{i}",
scan=scan,
status=Status.FAIL,
status_extended=f"Test status {i}",
impact=Severity.high,
severity=Severity.high,
raw_result={
"status": Status.FAIL,
"impact": Severity.high,
"severity": Severity.high,
},
check_id=f"test_check_id_{i}",
check_metadata={
"CheckId": f"test_check_id_{i}",
"Description": f"Test description {i}",
},
muted=False,
)
finding_uids.append(finding.uid)
# Create mute rule targeting all findings
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test Multiple Findings Mute Rule",
reason="Testing batch muting",
enabled=True,
created_by=test_user,
finding_uids=finding_uids,
)
return mute_rule, finding_uids
@pytest.fixture(scope="function")
def mute_rule_already_muted(self, findings_fixture, test_user):
"""
Create a mute rule that targets an already-muted finding.
"""
tenant_id = findings_fixture[1].tenant_id
already_muted_finding = findings_fixture[1]
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test Already Muted Rule",
reason="Testing already muted findings",
enabled=True,
created_by=test_user,
finding_uids=[already_muted_finding.uid],
)
return mute_rule
@pytest.fixture(scope="function")
def mute_rule_mixed_findings(self, scans_fixture, test_user):
"""
Create a mute rule with a mix of muted and unmuted findings.
"""
scan = scans_fixture[0]
tenant_id = scan.tenant_id
# Create 3 unmuted findings
unmuted_uids = []
for i in range(3):
finding = Finding.objects.create(
tenant_id=tenant_id,
uid=f"unmuted_finding_{i}",
scan=scan,
status=Status.FAIL,
status_extended=f"Unmuted status {i}",
impact=Severity.medium,
severity=Severity.medium,
raw_result={
"status": Status.FAIL,
"impact": Severity.medium,
"severity": Severity.medium,
},
check_id=f"unmuted_check_{i}",
check_metadata={
"CheckId": f"unmuted_check_{i}",
"Description": f"Unmuted description {i}",
},
muted=False,
)
unmuted_uids.append(finding.uid)
# Create 2 already muted findings
muted_uids = []
for i in range(2):
finding = Finding.objects.create(
tenant_id=tenant_id,
uid=f"muted_finding_{i}",
scan=scan,
status=Status.FAIL,
status_extended=f"Muted status {i}",
impact=Severity.low,
severity=Severity.low,
raw_result={
"status": Status.FAIL,
"impact": Severity.low,
"severity": Severity.low,
},
check_id=f"muted_check_{i}",
check_metadata={
"CheckId": f"muted_check_{i}",
"Description": f"Muted description {i}",
},
muted=True,
muted_at=datetime.now(timezone.utc),
muted_reason="Already muted",
)
muted_uids.append(finding.uid)
# Create mute rule targeting all findings
all_uids = unmuted_uids + muted_uids
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test Mixed Findings Rule",
reason="Testing mixed muted/unmuted findings",
enabled=True,
created_by=test_user,
finding_uids=all_uids,
)
return mute_rule, unmuted_uids, muted_uids
@pytest.fixture(scope="function")
def mute_rule_batch_test(self, scans_fixture, test_user):
"""
Create enough findings to test batch processing (>1000 for default batch size).
"""
scan = scans_fixture[0]
tenant_id = scan.tenant_id
# Create 1500 findings to exceed default batch size of 1000
finding_uids = []
for i in range(1500):
finding = Finding.objects.create(
tenant_id=tenant_id,
uid=f"batch_test_finding_{i}",
scan=scan,
status=Status.FAIL,
status_extended=f"Batch test status {i}",
impact=Severity.critical,
severity=Severity.critical,
raw_result={
"status": Status.FAIL,
"impact": Severity.critical,
"severity": Severity.critical,
},
check_id=f"batch_test_check_{i}",
check_metadata={
"CheckId": f"batch_test_check_{i}",
"Description": f"Batch test description {i}",
},
muted=False,
)
finding_uids.append(finding.uid)
# Create mute rule targeting all findings
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test Batch Processing Rule",
reason="Testing batch processing functionality",
enabled=True,
created_by=test_user,
finding_uids=finding_uids,
)
return mute_rule, finding_uids
def test_mute_historical_findings_single_finding(
self, mute_rule_with_findings, findings_fixture
):
"""
Test muting a single historical finding.
"""
mute_rule = mute_rule_with_findings
tenant_id = str(mute_rule.tenant_id)
finding = findings_fixture[0]
# Ensure the finding is not muted before execution
finding.refresh_from_db()
assert finding.muted is False
assert finding.muted_at is None
assert finding.muted_reason is None
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify return value
assert result["findings_muted"] == 1
assert result["rule_id"] == str(mute_rule.id)
# Verify the finding was muted
finding.refresh_from_db()
assert finding.muted is True
assert finding.muted_at == mute_rule.inserted_at
assert finding.muted_reason == mute_rule.reason
def test_mute_historical_findings_multiple_findings(
self, mute_rule_multiple_findings
):
"""
Test muting multiple historical findings.
"""
mute_rule, finding_uids = mute_rule_multiple_findings
tenant_id = str(mute_rule.tenant_id)
# Verify all findings are unmuted
findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
assert findings.count() == 5
for finding in findings:
assert finding.muted is False
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify return value
assert result["findings_muted"] == 5
assert result["rule_id"] == str(mute_rule.id)
# Verify all findings were muted
findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
for finding in findings:
assert finding.muted is True
assert finding.muted_at == mute_rule.inserted_at
assert finding.muted_reason == mute_rule.reason
def test_mute_historical_findings_already_muted(
self, mute_rule_already_muted, findings_fixture
):
"""
Test that already-muted findings are not counted or updated.
"""
mute_rule = mute_rule_already_muted
tenant_id = str(mute_rule.tenant_id)
finding = findings_fixture[1]
# Verify the finding is already muted
finding.refresh_from_db()
assert finding.muted is True
original_muted_at = finding.muted_at
original_muted_reason = finding.muted_reason
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify no findings were muted
assert result["findings_muted"] == 0
assert result["rule_id"] == str(mute_rule.id)
# Verify the finding's mute status did not change
finding.refresh_from_db()
assert finding.muted is True
assert finding.muted_at == original_muted_at
assert finding.muted_reason == original_muted_reason
def test_mute_historical_findings_mixed_status(self, mute_rule_mixed_findings):
"""
Test muting when some findings are already muted and others are not.
"""
mute_rule, unmuted_uids, muted_uids = mute_rule_mixed_findings
tenant_id = str(mute_rule.tenant_id)
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify only unmuted findings were counted
assert result["findings_muted"] == 3
assert result["rule_id"] == str(mute_rule.id)
# Verify unmuted findings are now muted
unmuted_findings = Finding.objects.filter(
tenant_id=tenant_id, uid__in=unmuted_uids
)
for finding in unmuted_findings:
assert finding.muted is True
assert finding.muted_at == mute_rule.inserted_at
assert finding.muted_reason == mute_rule.reason
# Verify already-muted findings remained unchanged
already_muted_findings = Finding.objects.filter(
tenant_id=tenant_id, uid__in=muted_uids
)
for finding in already_muted_findings:
assert finding.muted is True
assert finding.muted_reason == "Already muted"
def test_mute_historical_findings_nonexistent_rule(self, tenants_fixture):
"""
Test that a nonexistent mute rule raises ObjectDoesNotExist.
"""
tenant_id = str(tenants_fixture[0].id)
nonexistent_rule_id = str(uuid4())
with pytest.raises(ObjectDoesNotExist):
mute_historical_findings(tenant_id, nonexistent_rule_id)
def test_mute_historical_findings_no_matching_findings(
self, tenants_fixture, test_user
):
"""
Test muting when no findings match the rule's UIDs.
"""
tenant_id = str(tenants_fixture[0].id)
# Create a mute rule with non-existent finding UIDs
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test No Match Rule",
reason="Testing no matching findings",
enabled=True,
created_by=test_user,
finding_uids=[
"nonexistent_uid_1",
"nonexistent_uid_2",
"nonexistent_uid_3",
],
)
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify no findings were muted
assert result["findings_muted"] == 0
assert result["rule_id"] == str(mute_rule.id)
def test_mute_historical_findings_batch_processing(self, mute_rule_batch_test):
"""
Test that large numbers of findings are processed in batches correctly.
"""
mute_rule, finding_uids = mute_rule_batch_test
tenant_id = str(mute_rule.tenant_id)
# Verify all findings exist and are unmuted
findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
assert findings.count() == 1500
for finding in findings:
assert finding.muted is False
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify return value
assert result["findings_muted"] == 1500
assert result["rule_id"] == str(mute_rule.id)
# Verify all findings were muted
findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=finding_uids)
for finding in findings:
assert finding.muted is True
assert finding.muted_at == mute_rule.inserted_at
assert finding.muted_reason == mute_rule.reason
def test_mute_historical_findings_preserves_muted_at_timestamp(
self, mute_rule_with_findings, findings_fixture
):
"""
Test that muted_at is set to the rule's inserted_at, not the current time.
"""
mute_rule = mute_rule_with_findings
tenant_id = str(mute_rule.tenant_id)
finding = findings_fixture[0]
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify the finding was muted
assert result["findings_muted"] == 1
# Verify muted_at matches the rule's inserted_at timestamp
finding.refresh_from_db()
assert finding.muted_at == mute_rule.inserted_at
assert finding.muted_at is not None
def test_mute_historical_findings_partial_match(self, scans_fixture, test_user):
"""
Test muting when only some of the rule's UIDs exist as findings.
"""
scan = scans_fixture[0]
tenant_id = str(scan.tenant_id)
# Create 3 findings
existing_uids = []
for i in range(3):
finding = Finding.objects.create(
tenant_id=tenant_id,
uid=f"partial_match_finding_{i}",
scan=scan,
status=Status.FAIL,
status_extended=f"Partial match status {i}",
impact=Severity.high,
severity=Severity.high,
raw_result={
"status": Status.FAIL,
"impact": Severity.high,
"severity": Severity.high,
},
check_id=f"partial_match_check_{i}",
check_metadata={
"CheckId": f"partial_match_check_{i}",
"Description": f"Partial match description {i}",
},
muted=False,
)
existing_uids.append(finding.uid)
# Create a mute rule with both existing and non-existing UIDs
all_uids = existing_uids + [
"nonexistent_uid_1",
"nonexistent_uid_2",
]
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test Partial Match Rule",
reason="Testing partial matching",
enabled=True,
created_by=test_user,
finding_uids=all_uids,
)
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify only existing findings were muted
assert result["findings_muted"] == 3
assert result["rule_id"] == str(mute_rule.id)
# Verify the existing findings were muted
findings = Finding.objects.filter(tenant_id=tenant_id, uid__in=existing_uids)
assert findings.count() == 3
for finding in findings:
assert finding.muted is True
assert finding.muted_at == mute_rule.inserted_at
assert finding.muted_reason == mute_rule.reason
def test_mute_historical_findings_empty_uids(self, tenants_fixture, test_user):
"""
Test muting when the rule has an empty finding_uids array.
"""
tenant_id = str(tenants_fixture[0].id)
# Create a mute rule with empty finding_uids
mute_rule = MuteRule.objects.create(
tenant_id=tenant_id,
name="Test Empty UIDs Rule",
reason="Testing empty UIDs",
enabled=True,
created_by=test_user,
finding_uids=[],
)
# Execute the muting function
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify no findings were muted
assert result["findings_muted"] == 0
assert result["rule_id"] == str(mute_rule.id)
def test_mute_historical_findings_return_format(self, mute_rule_with_findings):
"""
Test that the return value has the correct format and fields.
"""
mute_rule = mute_rule_with_findings
tenant_id = str(mute_rule.tenant_id)
result = mute_historical_findings(tenant_id, str(mute_rule.id))
# Verify return value structure
assert isinstance(result, dict)
assert "findings_muted" in result
assert "rule_id" in result
assert isinstance(result["findings_muted"], int)
assert isinstance(result["rule_id"], str)
assert result["rule_id"] == str(mute_rule.id)
api/src/backend/tasks/tests/test_scan.py
+564 -1
View File
@@ -18,7 +18,15 @@ from tasks.utils import CustomEncoder
from api.db_router import MainRouter
from api.exceptions import ProviderConnectionError
from api.models import Finding, Provider, Resource, Scan, StateChoices, StatusChoices
from api.models import (
Finding,
MuteRule,
Provider,
Resource,
Scan,
StateChoices,
StatusChoices,
)
from prowler.lib.check.models import Severity
@@ -739,6 +747,561 @@ class TestPerformScan:
# Assert that failed_findings_count was reset to 0 during the scan
assert resource.failed_findings_count == 0
def test_perform_prowler_scan_with_active_mute_rules(
self,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Test active MuteRule mutes findings with correct reason"""
with (
patch("api.db_utils.rls_transaction"),
patch(
"tasks.jobs.scan.initialize_prowler_provider"
) as mock_initialize_prowler_provider,
patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
patch(
"tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
new_callable=dict,
),
patch("api.compliance.PROWLER_CHECKS", new_callable=dict),
):
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
provider.provider = Provider.ProviderChoices.AWS
provider.save()
tenant_id = str(tenant.id)
scan_id = str(scan.id)
provider_id = str(provider.id)
# Create active MuteRule with specific finding UIDs
mute_rule_reason = "Accepted risk - production exception"
finding_uid_1 = "finding_to_mute_1"
finding_uid_2 = "finding_to_mute_2"
MuteRule.objects.create(
tenant_id=tenant_id,
name="Production Exception Rule",
reason=mute_rule_reason,
enabled=True,
finding_uids=[finding_uid_1, finding_uid_2],
)
# Mock findings: one FAIL and one PASS, both should be muted
muted_fail_finding = MagicMock()
muted_fail_finding.uid = finding_uid_1
muted_fail_finding.status = StatusChoices.FAIL
muted_fail_finding.status_extended = "muted fail"
muted_fail_finding.severity = Severity.high
muted_fail_finding.check_id = "muted_fail_check"
muted_fail_finding.get_metadata.return_value = {"key": "value"}
muted_fail_finding.resource_uid = "resource_uid_1"
muted_fail_finding.resource_name = "resource_1"
muted_fail_finding.region = "us-east-1"
muted_fail_finding.service_name = "ec2"
muted_fail_finding.resource_type = "instance"
muted_fail_finding.resource_tags = {}
muted_fail_finding.muted = False
muted_fail_finding.raw = {}
muted_fail_finding.resource_metadata = {}
muted_fail_finding.resource_details = {}
muted_fail_finding.partition = "aws"
muted_fail_finding.compliance = {}
muted_pass_finding = MagicMock()
muted_pass_finding.uid = finding_uid_2
muted_pass_finding.status = StatusChoices.PASS
muted_pass_finding.status_extended = "muted pass"
muted_pass_finding.severity = Severity.medium
muted_pass_finding.check_id = "muted_pass_check"
muted_pass_finding.get_metadata.return_value = {"key": "value"}
muted_pass_finding.resource_uid = "resource_uid_2"
muted_pass_finding.resource_name = "resource_2"
muted_pass_finding.region = "us-east-1"
muted_pass_finding.service_name = "s3"
muted_pass_finding.resource_type = "bucket"
muted_pass_finding.resource_tags = {}
muted_pass_finding.muted = False
muted_pass_finding.raw = {}
muted_pass_finding.resource_metadata = {}
muted_pass_finding.resource_details = {}
muted_pass_finding.partition = "aws"
muted_pass_finding.compliance = {}
# Mock the ProwlerScan instance
mock_prowler_scan_instance = MagicMock()
mock_prowler_scan_instance.scan.return_value = [
(100, [muted_fail_finding, muted_pass_finding])
]
mock_prowler_scan_class.return_value = mock_prowler_scan_instance
# Mock prowler_provider
mock_prowler_provider_instance = MagicMock()
mock_prowler_provider_instance.get_regions.return_value = ["us-east-1"]
mock_initialize_prowler_provider.return_value = (
mock_prowler_provider_instance
)
# Call the function under test
perform_prowler_scan(tenant_id, scan_id, provider_id, [])
# Verify findings are muted with correct reason
fail_finding_db = Finding.objects.get(uid=finding_uid_1)
pass_finding_db = Finding.objects.get(uid=finding_uid_2)
assert fail_finding_db.muted
assert fail_finding_db.muted_reason == mute_rule_reason
assert fail_finding_db.muted_at is not None
assert pass_finding_db.muted
assert pass_finding_db.muted_reason == mute_rule_reason
assert pass_finding_db.muted_at is not None
# Verify failed_findings_count is 0 for muted FAIL finding
resource_1 = Resource.objects.get(uid="resource_uid_1")
assert resource_1.failed_findings_count == 0
def test_perform_prowler_scan_with_inactive_mute_rules(
self,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Test inactive MuteRule does not mute findings"""
with (
patch("api.db_utils.rls_transaction"),
patch(
"tasks.jobs.scan.initialize_prowler_provider"
) as mock_initialize_prowler_provider,
patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
patch(
"tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
new_callable=dict,
),
patch("api.compliance.PROWLER_CHECKS", new_callable=dict),
):
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
provider.provider = Provider.ProviderChoices.AWS
provider.save()
tenant_id = str(tenant.id)
scan_id = str(scan.id)
provider_id = str(provider.id)
# Create inactive MuteRule
finding_uid = "finding_inactive_rule"
MuteRule.objects.create(
tenant_id=tenant_id,
name="Inactive Rule",
reason="Should not apply",
enabled=False,
finding_uids=[finding_uid],
)
# Mock FAIL finding
fail_finding = MagicMock()
fail_finding.uid = finding_uid
fail_finding.status = StatusChoices.FAIL
fail_finding.status_extended = "test fail"
fail_finding.severity = Severity.high
fail_finding.check_id = "fail_check"
fail_finding.get_metadata.return_value = {"key": "value"}
fail_finding.resource_uid = "resource_uid_inactive"
fail_finding.resource_name = "resource_inactive"
fail_finding.region = "us-east-1"
fail_finding.service_name = "ec2"
fail_finding.resource_type = "instance"
fail_finding.resource_tags = {}
fail_finding.muted = False
fail_finding.raw = {}
fail_finding.resource_metadata = {}
fail_finding.resource_details = {}
fail_finding.partition = "aws"
fail_finding.compliance = {}
# Mock the ProwlerScan instance
mock_prowler_scan_instance = MagicMock()
mock_prowler_scan_instance.scan.return_value = [(100, [fail_finding])]
mock_prowler_scan_class.return_value = mock_prowler_scan_instance
# Mock prowler_provider
mock_prowler_provider_instance = MagicMock()
mock_prowler_provider_instance.get_regions.return_value = ["us-east-1"]
mock_initialize_prowler_provider.return_value = (
mock_prowler_provider_instance
)
# Call the function under test
perform_prowler_scan(tenant_id, scan_id, provider_id, [])
# Verify finding is NOT muted
finding_db = Finding.objects.get(uid=finding_uid)
assert not finding_db.muted
assert finding_db.muted_reason is None
assert finding_db.muted_at is None
# Verify failed_findings_count increments for FAIL finding
resource = Resource.objects.get(uid="resource_uid_inactive")
assert resource.failed_findings_count == 1
def test_perform_prowler_scan_mutelist_overrides_mute_rules(
self,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Test mutelist processor takes precedence over MuteRule"""
with (
patch("api.db_utils.rls_transaction"),
patch(
"tasks.jobs.scan.initialize_prowler_provider"
) as mock_initialize_prowler_provider,
patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
patch(
"tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
new_callable=dict,
),
patch("api.compliance.PROWLER_CHECKS", new_callable=dict),
):
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
provider.provider = Provider.ProviderChoices.AWS
provider.save()
tenant_id = str(tenant.id)
scan_id = str(scan.id)
provider_id = str(provider.id)
# Create active MuteRule
finding_uid = "finding_both_rules"
MuteRule.objects.create(
tenant_id=tenant_id,
name="Manual Mute Rule",
reason="Muted by manual rule",
enabled=True,
finding_uids=[finding_uid],
)
# Mock finding with mutelist processor muted=True
muted_finding = MagicMock()
muted_finding.uid = finding_uid
muted_finding.status = StatusChoices.FAIL
muted_finding.status_extended = "test"
muted_finding.severity = Severity.high
muted_finding.check_id = "test_check"
muted_finding.get_metadata.return_value = {"key": "value"}
muted_finding.resource_uid = "resource_both"
muted_finding.resource_name = "resource_both"
muted_finding.region = "us-east-1"
muted_finding.service_name = "ec2"
muted_finding.resource_type = "instance"
muted_finding.resource_tags = {}
muted_finding.muted = True
muted_finding.raw = {}
muted_finding.resource_metadata = {}
muted_finding.resource_details = {}
muted_finding.partition = "aws"
muted_finding.compliance = {}
# Mock the ProwlerScan instance
mock_prowler_scan_instance = MagicMock()
mock_prowler_scan_instance.scan.return_value = [(100, [muted_finding])]
mock_prowler_scan_class.return_value = mock_prowler_scan_instance
# Mock prowler_provider
mock_prowler_provider_instance = MagicMock()
mock_prowler_provider_instance.get_regions.return_value = ["us-east-1"]
mock_initialize_prowler_provider.return_value = (
mock_prowler_provider_instance
)
# Call the function under test
perform_prowler_scan(tenant_id, scan_id, provider_id, [])
# Verify mutelist reason takes precedence
finding_db = Finding.objects.get(uid=finding_uid)
assert finding_db.muted
assert finding_db.muted_reason == "Muted by mutelist"
assert finding_db.muted_at is not None
# Verify failed_findings_count is 0
resource = Resource.objects.get(uid="resource_both")
assert resource.failed_findings_count == 0
def test_perform_prowler_scan_mute_rules_multiple_findings(
self,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Test MuteRule with multiple finding UIDs mutes all findings"""
with (
patch("api.db_utils.rls_transaction"),
patch(
"tasks.jobs.scan.initialize_prowler_provider"
) as mock_initialize_prowler_provider,
patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
patch(
"tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
new_callable=dict,
),
patch("api.compliance.PROWLER_CHECKS", new_callable=dict),
):
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
provider.provider = Provider.ProviderChoices.AWS
provider.save()
tenant_id = str(tenant.id)
scan_id = str(scan.id)
provider_id = str(provider.id)
# Create MuteRule with multiple finding UIDs
mute_rule_reason = "Bulk exception for dev environment"
finding_uids = [
"bulk_finding_1",
"bulk_finding_2",
"bulk_finding_3",
"bulk_finding_4",
]
MuteRule.objects.create(
tenant_id=tenant_id,
name="Bulk Mute Rule",
reason=mute_rule_reason,
enabled=True,
finding_uids=finding_uids,
)
# Mock multiple findings with mixed statuses
findings = []
for i, uid in enumerate(finding_uids):
finding = MagicMock()
finding.uid = uid
finding.status = (
StatusChoices.FAIL if i % 2 == 0 else StatusChoices.PASS
)
finding.status_extended = f"test {i}"
finding.severity = Severity.medium
finding.check_id = f"check_{i}"
finding.get_metadata.return_value = {"key": f"value_{i}"}
finding.resource_uid = f"resource_bulk_{i}"
finding.resource_name = f"resource_{i}"
finding.region = "us-west-2"
finding.service_name = "lambda"
finding.resource_type = "function"
finding.resource_tags = {}
finding.muted = False
finding.raw = {}
finding.resource_metadata = {}
finding.resource_details = {}
finding.partition = "aws"
finding.compliance = {}
findings.append(finding)
# Mock the ProwlerScan instance
mock_prowler_scan_instance = MagicMock()
mock_prowler_scan_instance.scan.return_value = [(100, findings)]
mock_prowler_scan_class.return_value = mock_prowler_scan_instance
# Mock prowler_provider
mock_prowler_provider_instance = MagicMock()
mock_prowler_provider_instance.get_regions.return_value = ["us-west-2"]
mock_initialize_prowler_provider.return_value = (
mock_prowler_provider_instance
)
# Call the function under test
perform_prowler_scan(tenant_id, scan_id, provider_id, [])
# Verify all findings are muted with same reason
for uid in finding_uids:
finding_db = Finding.objects.get(uid=uid)
assert finding_db.muted
assert finding_db.muted_reason == mute_rule_reason
assert finding_db.muted_at is not None
# Verify all resources have failed_findings_count = 0
for i in range(len(finding_uids)):
resource = Resource.objects.get(uid=f"resource_bulk_{i}")
assert resource.failed_findings_count == 0
def test_perform_prowler_scan_mute_rules_error_handling(
self,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Test scan continues when MuteRule loading fails"""
with (
patch("api.db_utils.rls_transaction"),
patch(
"tasks.jobs.scan.initialize_prowler_provider"
) as mock_initialize_prowler_provider,
patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
patch(
"tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
new_callable=dict,
),
patch("api.compliance.PROWLER_CHECKS", new_callable=dict),
patch("api.models.MuteRule.objects.filter") as mock_mute_rule_filter,
):
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
provider.provider = Provider.ProviderChoices.AWS
provider.save()
tenant_id = str(tenant.id)
scan_id = str(scan.id)
provider_id = str(provider.id)
# Mock MuteRule.objects.filter to raise exception
mock_mute_rule_filter.side_effect = Exception("Database error")
# Mock finding
finding = MagicMock()
finding.uid = "finding_error_handling"
finding.status = StatusChoices.FAIL
finding.status_extended = "test"
finding.severity = Severity.high
finding.check_id = "test_check"
finding.get_metadata.return_value = {"key": "value"}
finding.resource_uid = "resource_error"
finding.resource_name = "resource_error"
finding.region = "us-east-1"
finding.service_name = "ec2"
finding.resource_type = "instance"
finding.resource_tags = {}
finding.muted = False
finding.raw = {}
finding.resource_metadata = {}
finding.resource_details = {}
finding.partition = "aws"
finding.compliance = {}
# Mock the ProwlerScan instance
mock_prowler_scan_instance = MagicMock()
mock_prowler_scan_instance.scan.return_value = [(100, [finding])]
mock_prowler_scan_class.return_value = mock_prowler_scan_instance
# Mock prowler_provider
mock_prowler_provider_instance = MagicMock()
mock_prowler_provider_instance.get_regions.return_value = ["us-east-1"]
mock_initialize_prowler_provider.return_value = (
mock_prowler_provider_instance
)
# Call the function under test - should not raise
perform_prowler_scan(tenant_id, scan_id, provider_id, [])
# Verify scan completed successfully
scan.refresh_from_db()
assert scan.state == StateChoices.COMPLETED
# Verify finding is not muted (mute_rules_cache was empty dict)
finding_db = Finding.objects.get(uid="finding_error_handling")
assert not finding_db.muted
assert finding_db.muted_reason is None
# Verify failed_findings_count increments
resource = Resource.objects.get(uid="resource_error")
assert resource.failed_findings_count == 1
def test_perform_prowler_scan_muted_at_timestamp(
self,
tenants_fixture,
scans_fixture,
providers_fixture,
):
"""Test muted_at timestamp is set correctly for muted findings"""
with (
patch("api.db_utils.rls_transaction"),
patch(
"tasks.jobs.scan.initialize_prowler_provider"
) as mock_initialize_prowler_provider,
patch("tasks.jobs.scan.ProwlerScan") as mock_prowler_scan_class,
patch(
"tasks.jobs.scan.PROWLER_COMPLIANCE_OVERVIEW_TEMPLATE",
new_callable=dict,
),
patch("api.compliance.PROWLER_CHECKS", new_callable=dict),
):
tenant = tenants_fixture[0]
scan = scans_fixture[0]
provider = providers_fixture[0]
provider.provider = Provider.ProviderChoices.AWS
provider.save()
tenant_id = str(tenant.id)
scan_id = str(scan.id)
provider_id = str(provider.id)
# Create active MuteRule
finding_uid = "finding_timestamp_test"
MuteRule.objects.create(
tenant_id=tenant_id,
name="Timestamp Test Rule",
reason="Testing timestamp",
enabled=True,
finding_uids=[finding_uid],
)
# Mock finding
finding = MagicMock()
finding.uid = finding_uid
finding.status = StatusChoices.FAIL
finding.status_extended = "test"
finding.severity = Severity.high
finding.check_id = "test_check"
finding.get_metadata.return_value = {"key": "value"}
finding.resource_uid = "resource_timestamp"
finding.resource_name = "resource_timestamp"
finding.region = "us-east-1"
finding.service_name = "ec2"
finding.resource_type = "instance"
finding.resource_tags = {}
finding.muted = False
finding.raw = {}
finding.resource_metadata = {}
finding.resource_details = {}
finding.partition = "aws"
finding.compliance = {}
# Mock the ProwlerScan instance
mock_prowler_scan_instance = MagicMock()
mock_prowler_scan_instance.scan.return_value = [(100, [finding])]
mock_prowler_scan_class.return_value = mock_prowler_scan_instance
# Mock prowler_provider
mock_prowler_provider_instance = MagicMock()
mock_prowler_provider_instance.get_regions.return_value = ["us-east-1"]
mock_initialize_prowler_provider.return_value = (
mock_prowler_provider_instance
)
# Capture time before and after scan
before_scan = datetime.now(timezone.utc)
perform_prowler_scan(tenant_id, scan_id, provider_id, [])
after_scan = datetime.now(timezone.utc)
# Verify muted_at is within the scan time window
finding_db = Finding.objects.get(uid=finding_uid)
assert finding_db.muted
assert finding_db.muted_at is not None
assert before_scan <= finding_db.muted_at <= after_scan
# TODO Add tests for aggregations