diff --git a/prowler/lib/cli/parser.py b/prowler/lib/cli/parser.py index 741b5ef6fa..868de4d51c 100644 --- a/prowler/lib/cli/parser.py +++ b/prowler/lib/cli/parser.py @@ -12,7 +12,7 @@ from prowler.config.config import ( default_output_directory, ) from prowler.lib.check.models import Severity -from prowler.lib.outputs.finding import Status +from prowler.lib.outputs.common import Status from prowler.providers.common.arguments import ( init_providers_parser, validate_provider_arguments, diff --git a/prowler/lib/mutelist/mutelist.py b/prowler/lib/mutelist/mutelist.py index 56414ea425..c1314a6460 100644 --- a/prowler/lib/mutelist/mutelist.py +++ b/prowler/lib/mutelist/mutelist.py @@ -5,6 +5,8 @@ import yaml from prowler.lib.logger import logger from prowler.lib.mutelist.models import mutelist_schema +from prowler.lib.outputs.common import Status +from prowler.lib.outputs.utils import unroll_dict, unroll_tags class Mutelist(ABC): @@ -237,6 +239,35 @@ class Mutelist(ABC): ) return False + def mute_finding(self, finding): + """ + Check if the provided finding is muted + + Args: + finding (Finding): The finding to be evaluated for muting. + + Returns: + Finding: The finding with the status updated if it is muted, otherwise the finding is returned + + """ + try: + if self.is_muted( + finding.account_uid, + finding.metadata.CheckID, + finding.region, + finding.resource_uid, + unroll_dict(unroll_tags(finding.resource_tags)), + ): + finding.raw["status"] = finding.status + finding.status = Status.MUTED + finding.muted = True + return finding + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" + ) + return finding + def is_excepted( self, exceptions, diff --git a/prowler/lib/outputs/common.py b/prowler/lib/outputs/common.py index 484d14ad71..f77988a21e 100644 --- a/prowler/lib/outputs/common.py +++ b/prowler/lib/outputs/common.py @@ -1,3 +1,5 @@ +from enum import Enum + from prowler.config.config import timestamp from prowler.lib.outputs.utils import unroll_tags from prowler.lib.utils.utils import outputs_unix_timestamp @@ -15,3 +17,10 @@ def fill_common_finding_data(finding: dict, unix_timestamp: bool) -> dict: "resource_tags": unroll_tags(finding.resource_tags), } return finding_data + + +class Status(str, Enum): + PASS = "PASS" + FAIL = "FAIL" + MANUAL = "MANUAL" + MUTED = "MUTED" diff --git a/prowler/lib/outputs/finding.py b/prowler/lib/outputs/finding.py index daa24c0f75..9867111765 100644 --- a/prowler/lib/outputs/finding.py +++ b/prowler/lib/outputs/finding.py @@ -1,5 +1,4 @@ from datetime import datetime -from enum import Enum from typing import Optional, Union from pydantic import BaseModel, Field @@ -7,18 +6,12 @@ from pydantic import BaseModel, Field from prowler.config.config import prowler_version from prowler.lib.check.models import Check_Report, CheckMetadata from prowler.lib.logger import logger -from prowler.lib.outputs.common import fill_common_finding_data +from prowler.lib.outputs.common import Status, fill_common_finding_data from prowler.lib.outputs.compliance.compliance import get_check_compliance from prowler.lib.utils.utils import dict_to_lowercase, get_nested_attribute from prowler.providers.common.provider import Provider -class Status(str, Enum): - PASS = "PASS" - FAIL = "FAIL" - MANUAL = "MANUAL" - - class Finding(BaseModel): """ Represents the output model for a finding across different providers. @@ -49,6 +42,7 @@ class Finding(BaseModel): region: str compliance: dict prowler_version: str = prowler_version + raw: dict = Field(default_factory=dict) @property def provider(self) -> str: @@ -85,13 +79,6 @@ class Finding(BaseModel): """ return self.metadata.ServiceName - @property - def raw(self) -> dict: - """ - Returns the raw (dict) finding without any post-processing. - """ - return {} - def get_metadata(self) -> dict: """ Retrieves the metadata of the object and returns it as a dictionary with all keys in lowercase. diff --git a/prowler/lib/scan/scan.py b/prowler/lib/scan/scan.py index 94fdd00855..213161b64b 100644 --- a/prowler/lib/scan/scan.py +++ b/prowler/lib/scan/scan.py @@ -12,7 +12,8 @@ from prowler.lib.check.compliance import update_checks_metadata_with_compliance from prowler.lib.check.compliance_models import Compliance from prowler.lib.check.models import CheckMetadata, Severity from prowler.lib.logger import logger -from prowler.lib.outputs.finding import Finding, Status +from prowler.lib.outputs.common import Status +from prowler.lib.outputs.finding import Finding from prowler.lib.scan.exceptions.exceptions import ( ScanInvalidCategoryError, ScanInvalidCheckError, diff --git a/tests/lib/outputs/finding_test.py b/tests/lib/outputs/finding_test.py index 77cc371c1e..627d7df905 100644 --- a/tests/lib/outputs/finding_test.py +++ b/tests/lib/outputs/finding_test.py @@ -8,7 +8,8 @@ from prowler.lib.check.models import ( Remediation, Severity, ) -from prowler.lib.outputs.finding import Finding, Status +from prowler.lib.outputs.common import Status +from prowler.lib.outputs.finding import Finding from tests.lib.outputs.fixtures.fixtures import generate_finding_output diff --git a/tests/providers/aws/lib/mutelist/aws_mutelist_test.py b/tests/providers/aws/lib/mutelist/aws_mutelist_test.py index 848f59bf91..9df192a016 100644 --- a/tests/providers/aws/lib/mutelist/aws_mutelist_test.py +++ b/tests/providers/aws/lib/mutelist/aws_mutelist_test.py @@ -10,6 +10,7 @@ from moto import mock_aws from prowler.config.config import encoding_format_utf_8 from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist +from tests.lib.outputs.fixtures.fixtures import generate_finding_output from tests.providers.aws.services.awslambda.awslambda_service_test import ( create_zip_file, ) @@ -1844,3 +1845,35 @@ class TestAWSMutelist: allowlist_resources = ["*.es"] assert AWSMutelist.is_item_matched(allowlist_resources, "google.es") + + def test_mute_finding(self): + # Mutelist + mutelist_content = { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["prowler", "^test", "prowler-pro"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + # Finding + finding_1 = generate_finding_output( + check_id="check_test", + status="FAIL", + region=AWS_REGION_US_EAST_1, + resource_uid="prowler", + resource_tags=[], + muted=False, + ) + + muted_finding = mutelist.mute_finding(finding_1) + + assert muted_finding.status == "MUTED" + assert muted_finding.muted + assert muted_finding.raw["status"] == "FAIL" diff --git a/tests/providers/azure/lib/mutelist/azure_mutelist_test.py b/tests/providers/azure/lib/mutelist/azure_mutelist_test.py index d946443732..3488beb54f 100644 --- a/tests/providers/azure/lib/mutelist/azure_mutelist_test.py +++ b/tests/providers/azure/lib/mutelist/azure_mutelist_test.py @@ -2,6 +2,7 @@ import yaml from mock import MagicMock from prowler.providers.azure.lib.mutelist.mutelist import AzureMutelist +from tests.lib.outputs.fixtures.fixtures import generate_finding_output MUTELIST_FIXTURE_PATH = ( "tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml" @@ -66,3 +67,36 @@ class TestAzureMutelist: finding.subscription = "subscription_1" assert mutelist.is_finding_muted(finding) + + def test_mute_finding(self): + # Mutelist + mutelist_content = { + "Accounts": { + "subscription_1": { + "Checks": { + "check_test": { + "Regions": ["*"], + "Resources": ["test_resource"], + } + } + } + } + } + + mutelist = AzureMutelist(mutelist_content=mutelist_content) + + finding_1 = generate_finding_output( + check_id="check_test", + status="FAIL", + account_uid="subscription_1", + region="subscription_1", + resource_uid="test_resource", + resource_tags=[], + muted=False, + ) + + muted_finding = mutelist.mute_finding(finding=finding_1) + + assert muted_finding.status == "MUTED" + assert muted_finding.muted is True + assert muted_finding.raw["status"] == "FAIL" diff --git a/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py b/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py index 831ff154c1..0471c5f0f1 100644 --- a/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py +++ b/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py @@ -2,6 +2,7 @@ import yaml from mock import MagicMock from prowler.providers.gcp.lib.mutelist.mutelist import GCPMutelist +from tests.lib.outputs.fixtures.fixtures import generate_finding_output MUTELIST_FIXTURE_PATH = "tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml" @@ -64,3 +65,36 @@ class TestGCPMutelist: finding.project_id = "project_1" assert mutelist.is_finding_muted(finding) + + def test_mute_finding(self): + # Mutelist + mutelist_content = { + "Accounts": { + "project_1": { + "Checks": { + "check_test": { + "Regions": ["*"], + "Resources": ["test_resource"], + } + } + } + } + } + + mutelist = GCPMutelist(mutelist_content=mutelist_content) + + finding_1 = generate_finding_output( + check_id="check_test", + status="FAIL", + account_uid="project_1", + region="test-region", + resource_uid="test_resource", + resource_tags=[], + muted=False, + ) + + muted_finding = mutelist.mute_finding(finding=finding_1) + + assert muted_finding.status == "MUTED" + assert muted_finding.muted + assert muted_finding.raw["status"] == "FAIL" diff --git a/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py b/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py index 5f6c8197dc..7cc1ae394d 100644 --- a/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py +++ b/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py @@ -2,6 +2,7 @@ import yaml from mock import MagicMock from prowler.providers.kubernetes.lib.mutelist.mutelist import KubernetesMutelist +from tests.lib.outputs.fixtures.fixtures import generate_finding_output MUTELIST_FIXTURE_PATH = ( "tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml" @@ -128,3 +129,36 @@ class TestKubernetesMutelist: finding.resource_tags = [] assert mutelist.is_finding_muted(finding, "cluster_1") + + def test_mute_finding(self): + # Mutelist + mutelist_content = { + "Accounts": { + "cluster_1": { + "Checks": { + "check_test": { + "Regions": ["*"], + "Resources": ["test_resource"], + } + } + } + } + } + + mutelist = KubernetesMutelist(mutelist_content=mutelist_content) + + finding_1 = generate_finding_output( + check_id="check_test", + status="FAIL", + account_uid="cluster_1", + region="test-region", + resource_uid="test_resource", + resource_tags=[], + muted=False, + ) + + muted_finding = mutelist.mute_finding(finding_1) + + assert muted_finding.status == "MUTED" + assert muted_finding.muted is True + assert muted_finding.raw["status"] == "FAIL"