From 577afbd5219868ea1a0e16942d6818b566a8083f Mon Sep 17 00:00:00 2001 From: Pepe Fagoaga Date: Tue, 16 Jul 2024 19:44:43 +0200 Subject: [PATCH] chore(mutelist): create new class to encapsulate the logic (#4413) --- prowler/lib/check/check.py | 25 +- prowler/lib/mutelist/mutelist.py | 593 ++++---- prowler/providers/aws/aws_provider.py | 48 +- .../providers/aws/lib/mutelist/mutelist.py | 197 ++- prowler/providers/azure/azure_provider.py | 26 +- .../providers/azure/lib/mutelist/__init__.py | 0 .../providers/azure/lib/mutelist/mutelist.py | 18 + prowler/providers/common/provider.py | 40 +- prowler/providers/gcp/gcp_provider.py | 26 +- .../providers/gcp/lib/mutelist/__init__.py | 0 .../providers/gcp/lib/mutelist/mutelist.py | 18 + .../kubernetes/kubernetes_provider.py | 24 +- .../kubernetes/lib/mutelist/__init__.py | 0 .../kubernetes/lib/mutelist/mutelist.py | 19 + tests/lib/mutelist/mutelist_test.py | 1244 ---------------- tests/providers/aws/aws_provider_test.py | 33 +- .../aws/lib/mutelist/aws_mutelist_test.py | 1302 ++++++++++++++++- .../lib/mutelist/fixtures/aws_mutelist.yaml | 0 .../azure/lib/mutelist/azure_mutelist_test.py | 68 + .../lib/mutelist/fixtures/azure_mutelist.yaml | 16 + .../lib/mutelist/fixtures/gcp_mutelist.yaml | 16 + .../gcp/lib/mutelist/gcp_mutelist_test.py | 66 + .../fixtures/kubernetes_mutelist.yaml | 16 + .../lib/mutelist/kubernetes_mutelist_test.py | 99 ++ 24 files changed, 2084 insertions(+), 1810 deletions(-) create mode 100644 prowler/providers/azure/lib/mutelist/__init__.py create mode 100644 prowler/providers/azure/lib/mutelist/mutelist.py create mode 100644 prowler/providers/gcp/lib/mutelist/__init__.py create mode 100644 prowler/providers/gcp/lib/mutelist/mutelist.py create mode 100644 prowler/providers/kubernetes/lib/mutelist/__init__.py create mode 100644 prowler/providers/kubernetes/lib/mutelist/mutelist.py delete mode 100644 tests/lib/mutelist/mutelist_test.py rename tests/{ => providers/aws}/lib/mutelist/fixtures/aws_mutelist.yaml (100%) create mode 100644 tests/providers/azure/lib/mutelist/azure_mutelist_test.py create mode 100644 tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml create mode 100644 tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml create mode 100644 tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py create mode 100644 tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml create mode 100644 tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py diff --git a/prowler/lib/check/check.py b/prowler/lib/check/check.py index a023701f15..8bfc65f7ec 100644 --- a/prowler/lib/check/check.py +++ b/prowler/lib/check/check.py @@ -19,7 +19,6 @@ from prowler.lib.check.compliance_models import load_compliance_framework from prowler.lib.check.custom_checks_metadata import update_check_metadata from prowler.lib.check.models import Check, load_check_metadata from prowler.lib.logger import logger -from prowler.lib.mutelist.mutelist import mutelist_findings from prowler.lib.outputs.outputs import report from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes from prowler.providers.common.models import Audit_Metadata @@ -612,9 +611,9 @@ def execute_checks( else: # Prepare your messages messages = [f"Config File: {Fore.YELLOW}{config_file}{Style.RESET_ALL}"] - if global_provider.mutelist_file_path: + if global_provider.mutelist.mutelist_file_path: messages.append( - f"Mutelist File: {Fore.YELLOW}{global_provider.mutelist_file_path}{Style.RESET_ALL}" + f"Mutelist File: {Fore.YELLOW}{global_provider.mutelist.mutelist_file_path}{Style.RESET_ALL}" ) if global_provider.type == "aws": messages.append( @@ -715,11 +714,21 @@ def execute( ) # Mutelist findings - if hasattr(global_provider, "mutelist") and global_provider.mutelist: - check_findings = mutelist_findings( - global_provider, - check_findings, - ) + if hasattr(global_provider, "mutelist") and global_provider.mutelist.mutelist: + # TODO: make this prettier + is_finding_muted_args = {} + if global_provider.type == "aws": + is_finding_muted_args["aws_account_id"] = ( + global_provider.identity.account + ) + elif global_provider.type == "kubernetes": + is_finding_muted_args["cluster"] = global_provider.identity.cluster + + for finding in check_findings: + is_finding_muted_args["finding"] = finding + finding.muted = global_provider.mutelist.is_finding_muted( + **is_finding_muted_args + ) # Refactor(Outputs) # Report the check's findings diff --git a/prowler/lib/mutelist/mutelist.py b/prowler/lib/mutelist/mutelist.py index e61b6ee48e..19cf1cbfec 100644 --- a/prowler/lib/mutelist/mutelist.py +++ b/prowler/lib/mutelist/mutelist.py @@ -1,373 +1,292 @@ import re -from typing import Any +from abc import ABC, abstractmethod import yaml from prowler.lib.logger import logger from prowler.lib.mutelist.models import mutelist_schema -from prowler.lib.outputs.utils import unroll_tags -def get_mutelist_file_from_local_file(mutelist_path: str): - try: - with open(mutelist_path) as f: - mutelist = yaml.safe_load(f)["Mutelist"] - return mutelist - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return {} +class Mutelist(ABC): + _mutelist: dict = {} + _mutelist_file_path: str = None + MUTELIST_KEY = "Mutelist" -def validate_mutelist(mutelist: dict) -> dict: - try: - mutelist = mutelist_schema.validate(mutelist) - return mutelist - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- Mutelist YAML is malformed - {error}[{error.__traceback__.tb_lineno}]" - ) - return {} + def __init__( + self, mutelist_path: str = "", mutelist_content: dict = {} + ) -> "Mutelist": + if mutelist_path: + self._mutelist_file_path = mutelist_path + self.get_mutelist_file_from_local_file(mutelist_path) + else: + self._mutelist = mutelist_content + if self._mutelist: + self.validate_mutelist() -def mutelist_findings( - global_provider: Any, - check_findings: list[Any], -) -> list[Any]: - # Check if finding is muted - for finding in check_findings: - # TODO: Move this mapping to the execute_check function and pass that output to the mutelist and the report - if global_provider.type == "aws": - finding.muted = is_muted( - global_provider.mutelist, - global_provider.identity.account, - finding.check_metadata.CheckID, - finding.region, - finding.resource_id, - unroll_tags(finding.resource_tags), + @property + def mutelist(self) -> dict: + return self._mutelist + + @property + def mutelist_file_path(self) -> dict: + return self._mutelist_file_path + + @abstractmethod + def is_finding_muted(self) -> bool: + raise NotImplementedError + + def get_mutelist_file_from_local_file(self, mutelist_path: str): + try: + with open(mutelist_path) as f: + self._mutelist = yaml.safe_load(f)[self.MUTELIST_KEY] + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) - elif global_provider.type == "azure": - finding.muted = is_muted( - global_provider.mutelist, - finding.subscription, - finding.check_metadata.CheckID, - # TODO: add region to the findings when we add Azure Locations - # finding.region, - "", - finding.resource_name, - unroll_tags(finding.resource_tags), + + def validate_mutelist(self) -> bool: + try: + self._mutelist = mutelist_schema.validate(self._mutelist) + return True + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- Mutelist YAML is malformed - {error}[{error.__traceback__.tb_lineno}]" ) - elif global_provider.type == "gcp": - finding.muted = is_muted( - global_provider.mutelist, - finding.project_id, - finding.check_metadata.CheckID, - finding.location, - finding.resource_name, - unroll_tags(finding.resource_tags), + self._mutelist = {} + return False + + def is_muted( + self, + audited_account: str, + check: str, + finding_region: str, + finding_resource: str, + finding_tags, + ) -> bool: + """ + Check if the provided finding is muted for the audited account, check, region, resource and tags. + + Args: + mutelist (dict): Dictionary containing information about muted checks for different accounts. + audited_account (str): The account being audited. + check (str): The check to be evaluated for muting. + finding_region (str): The region where the finding occurred. + finding_resource (str): The resource related to the finding. + finding_tags: The tags associated with the finding. + + Returns: + bool: True if the finding is muted for the audited account, check, region, resource and tags., otherwise False. + """ + try: + # By default is not muted + is_finding_muted = False + + # We always check all the accounts present in the mutelist + # if one mutes the finding we set the finding as muted + for account in self._mutelist.get("Accounts", []): + if account == audited_account or account == "*": + if self.is_muted_in_check( + self._mutelist["Accounts"][account]["Checks"], + audited_account, + check, + finding_region, + finding_resource, + finding_tags, + ): + is_finding_muted = True + break + + return is_finding_muted + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) - elif global_provider.type == "kubernetes": - finding.muted = is_muted( - global_provider.mutelist, - global_provider.identity.cluster, - finding.check_metadata.CheckID, - finding.namespace, - finding.resource_name, - unroll_tags(finding.resource_tags), - ) - return check_findings + return False + def is_muted_in_check( + self, + muted_checks, + audited_account, + check, + finding_region, + finding_resource, + finding_tags, + ) -> bool: + """ + Check if the provided check is muted. -def is_muted( - mutelist: dict, - audited_account: str, - check: str, - finding_region: str, - finding_resource: str, - finding_tags, -) -> bool: - """ - Check if the provided finding is muted for the audited account, check, region, resource and tags. + Args: + muted_checks (dict): Dictionary containing information about muted checks. + audited_account (str): The account to be audited. + check (str): The check to be evaluated for muting. + finding_region (str): The region where the finding occurred. + finding_resource (str): The resource related to the finding. + finding_tags (str): The tags associated with the finding. - Args: - mutelist (dict): Dictionary containing information about muted checks for different accounts. - audited_account (str): The account being audited. - check (str): The check to be evaluated for muting. - finding_region (str): The region where the finding occurred. - finding_resource (str): The resource related to the finding. - finding_tags: The tags associated with the finding. + Returns: + bool: True if the check is muted, otherwise False. + """ + try: + # Default value is not muted + is_check_muted = False - Returns: - bool: True if the finding is muted for the audited account, check, region, resource and tags., otherwise False. - """ - try: - # By default is not muted - is_finding_muted = False + for muted_check, muted_check_info in muted_checks.items(): + # map lambda to awslambda + muted_check = re.sub("^lambda", "awslambda", muted_check) - # We always check all the accounts present in the mutelist - # if one mutes the finding we set the finding as muted - for account in mutelist["Accounts"]: - if account == audited_account or account == "*": - if is_muted_in_check( - mutelist["Accounts"][account]["Checks"], - audited_account, - check, - finding_region, - finding_resource, - finding_tags, + check_match = ( + "*" == muted_check + or check == muted_check + or self.is_item_matched([muted_check], check) + ) + + # Check if the finding is excepted + exceptions = muted_check_info.get("Exceptions") + if ( + self.is_excepted( + exceptions, + audited_account, + finding_region, + finding_resource, + finding_tags, + ) + and check_match ): - is_finding_muted = True + # Break loop and return default value since is excepted break - return is_finding_muted - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False + muted_regions = muted_check_info.get("Regions") + muted_resources = muted_check_info.get("Resources") + muted_tags = muted_check_info.get("Tags", "*") + # We need to set the muted_tags if None, "" or [], so the falsy helps + if not muted_tags: + muted_tags = "*" + # If there is a *, it affects to all checks + if check_match: + muted_in_check = True + muted_in_region = self.is_item_matched( + muted_regions, finding_region + ) + muted_in_resource = self.is_item_matched( + muted_resources, finding_resource + ) + muted_in_tags = self.is_item_matched(muted_tags, finding_tags) + # For a finding to be muted requires the following set to True: + # - muted_in_check -> True + # - muted_in_region -> True + # - muted_in_tags -> True + # - muted_in_resource -> True + # - excepted -> False -def is_muted_in_check( - muted_checks, - audited_account, - check, - finding_region, - finding_resource, - finding_tags, -) -> bool: - """ - Check if the provided check is muted. + if ( + muted_in_check + and muted_in_region + and muted_in_tags + and muted_in_resource + ): + is_check_muted = True - Args: - muted_checks (dict): Dictionary containing information about muted checks. - audited_account (str): The account to be audited. - check (str): The check to be evaluated for muting. - finding_region (str): The region where the finding occurred. - finding_resource (str): The resource related to the finding. - finding_tags (str): The tags associated with the finding. - - Returns: - bool: True if the check is muted, otherwise False. - """ - try: - # Default value is not muted - is_check_muted = False - - for muted_check, muted_check_info in muted_checks.items(): - # map lambda to awslambda - muted_check = re.sub("^lambda", "awslambda", muted_check) - - check_match = ( - "*" == muted_check - or check == muted_check - or re.search(muted_check, check) + return is_check_muted + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) - # Check if the finding is excepted - exceptions = muted_check_info.get("Exceptions") - if ( - is_excepted( - exceptions, - audited_account, - finding_region, - finding_resource, - finding_tags, - ) - and check_match - ): - # Break loop and return default value since is excepted - break + return False - muted_regions = muted_check_info.get("Regions") - muted_resources = muted_check_info.get("Resources") - muted_tags = muted_check_info.get("Tags", "*") - # We need to set the muted_tags if None, "" or [], so the falsy helps - if not muted_tags: - muted_tags = "*" - # If there is a *, it affects to all checks - if check_match: - muted_in_check = True - muted_in_region = is_muted_in_region(muted_regions, finding_region) - muted_in_resource = is_muted_in_resource( - muted_resources, finding_resource - ) - muted_in_tags = is_muted_in_tags(muted_tags, finding_tags) + def is_excepted( + self, + exceptions, + audited_account, + finding_region, + finding_resource, + finding_tags, + ) -> bool: + """ + Check if the provided account, region, resource, and tags are excepted based on the exceptions dictionary. - # For a finding to be muted requires the following set to True: - # - muted_in_check -> True - # - muted_in_region -> True - # - muted_in_tags -> True - # - muted_in_resource -> True - # - excepted -> False + Args: + exceptions (dict): Dictionary containing exceptions for different attributes like Accounts, Regions, Resources, and Tags. + audited_account (str): The account to be audited. + finding_region (str): The region where the finding occurred. + finding_resource (str): The resource related to the finding. + finding_tags (str): The tags associated with the finding. + + Returns: + bool: True if the account, region, resource, and tags are excepted based on the exceptions, otherwise False. + """ + try: + excepted = False + is_account_excepted = False + is_region_excepted = False + is_resource_excepted = False + is_tag_excepted = False + if exceptions: + excepted_accounts = exceptions.get("Accounts", []) + is_account_excepted = self.is_item_matched( + excepted_accounts, audited_account + ) + + excepted_regions = exceptions.get("Regions", []) + is_region_excepted = self.is_item_matched( + excepted_regions, finding_region + ) + + excepted_resources = exceptions.get("Resources", []) + is_resource_excepted = self.is_item_matched( + excepted_resources, finding_resource + ) + + excepted_tags = exceptions.get("Tags", []) + is_tag_excepted = self.is_item_matched(excepted_tags, finding_tags) if ( - muted_in_check - and muted_in_region - and muted_in_tags - and muted_in_resource + not is_account_excepted + and not is_region_excepted + and not is_resource_excepted + and not is_tag_excepted ): - is_check_muted = True - - return is_check_muted - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False - - -def is_muted_in_region( - mutelist_regions, - finding_region, -) -> bool: - """ - Check if the finding_region is present in the mutelist_regions. - - Args: - mutelist_regions (list): List of regions in the mute list. - finding_region (str): Region to check if it is muted. - - Returns: - bool: True if the finding_region is muted in any of the mutelist_regions, otherwise False. - """ - try: - return __is_item_matched__(mutelist_regions, finding_region) - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False - - -def is_muted_in_tags(muted_tags, finding_tags) -> bool: - """ - Check if any of the muted tags are present in the finding tags. - - Args: - muted_tags (list): List of muted tags to be checked. - finding_tags (str): String containing tags to search for muted tags. - - Returns: - bool: True if any of the muted tags are present in the finding tags, otherwise False. - """ - try: - return __is_item_matched__(muted_tags, finding_tags) - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False - - -def is_muted_in_resource(muted_resources, finding_resource) -> bool: - """ - Check if any of the muted_resources are present in the finding_resource. - - Args: - muted_resources (list): List of muted resources to be checked. - finding_resource (str): Resource to search for muted resources. - - Returns: - bool: True if any of the muted_resources are present in the finding_resource, otherwise False. - """ - try: - return __is_item_matched__(muted_resources, finding_resource) - - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False - - -def is_excepted( - exceptions, - audited_account, - finding_region, - finding_resource, - finding_tags, -) -> bool: - """ - Check if the provided account, region, resource, and tags are excepted based on the exceptions dictionary. - - Args: - exceptions (dict): Dictionary containing exceptions for different attributes like Accounts, Regions, Resources, and Tags. - audited_account (str): The account to be audited. - finding_region (str): The region where the finding occurred. - finding_resource (str): The resource related to the finding. - finding_tags (str): The tags associated with the finding. - - Returns: - bool: True if the account, region, resource, and tags are excepted based on the exceptions, otherwise False. - """ - try: - excepted = False - is_account_excepted = False - is_region_excepted = False - is_resource_excepted = False - is_tag_excepted = False - if exceptions: - excepted_accounts = exceptions.get("Accounts", []) - is_account_excepted = __is_item_matched__( - excepted_accounts, audited_account + excepted = False + elif ( + (is_account_excepted or not excepted_accounts) + and (is_region_excepted or not excepted_regions) + and (is_resource_excepted or not excepted_resources) + and (is_tag_excepted or not excepted_tags) + ): + excepted = True + return excepted + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) + return False - excepted_regions = exceptions.get("Regions", []) - is_region_excepted = __is_item_matched__(excepted_regions, finding_region) + @staticmethod + def is_item_matched(matched_items, finding_items): + """ + Check if any of the items in matched_items are present in finding_items. - excepted_resources = exceptions.get("Resources", []) - is_resource_excepted = __is_item_matched__( - excepted_resources, finding_resource + Args: + matched_items (list): List of items to be matched. + finding_items (str): String to search for matched items. + + Returns: + bool: True if any of the matched_items are present in finding_items, otherwise False. + """ + try: + is_item_matched = False + if matched_items and (finding_items or finding_items == ""): + for item in matched_items: + if item.startswith("*"): + item = ".*" + item[1:] + if re.match(item, finding_items): + is_item_matched = True + break + return is_item_matched + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) - - excepted_tags = exceptions.get("Tags", []) - is_tag_excepted = __is_item_matched__(excepted_tags, finding_tags) - - if ( - not is_account_excepted - and not is_region_excepted - and not is_resource_excepted - and not is_tag_excepted - ): - excepted = False - elif ( - (is_account_excepted or not excepted_accounts) - and (is_region_excepted or not excepted_regions) - and (is_resource_excepted or not excepted_resources) - and (is_tag_excepted or not excepted_tags) - ): - excepted = True - return excepted - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False - - -def __is_item_matched__(matched_items, finding_items): - """ - Check if any of the items in matched_items are present in finding_items. - - Args: - matched_items (list): List of items to be matched. - finding_items (str): String to search for matched items. - - Returns: - bool: True if any of the matched_items are present in finding_items, otherwise False. - """ - try: - is_item_matched = False - if matched_items and (finding_items or finding_items == ""): - for item in matched_items: - if item.startswith("*"): - item = ".*" + item[1:] - if re.search(item, finding_items): - is_item_matched = True - break - return is_item_matched - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return False + return False diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py index 43331de4b7..e2e6f6de0e 100644 --- a/prowler/providers/aws/aws_provider.py +++ b/prowler/providers/aws/aws_provider.py @@ -1,6 +1,5 @@ import os import pathlib -import re import sys from argparse import Namespace from datetime import datetime @@ -22,10 +21,6 @@ from prowler.config.config import ( ) from prowler.lib.check.check import list_modules, recover_checks_from_service from prowler.lib.logger import logger -from prowler.lib.mutelist.mutelist import ( - get_mutelist_file_from_local_file, - validate_mutelist, -) from prowler.lib.utils.utils import open_file, parse_json_file, print_boxes from prowler.providers.aws.config import ( AWS_REGION_US_EAST_1, @@ -35,11 +30,7 @@ from prowler.providers.aws.config import ( ) from prowler.providers.aws.lib.arn.arn import parse_iam_credentials_arn from prowler.providers.aws.lib.arn.models import ARN -from prowler.providers.aws.lib.mutelist.mutelist import ( - get_mutelist_file_from_dynamodb, - get_mutelist_file_from_lambda, - get_mutelist_file_from_s3, -) +from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist from prowler.providers.aws.lib.organizations.organizations import ( get_organizations_metadata, parse_organizations_metadata, @@ -295,12 +286,14 @@ class AwsProvider(Provider): ) @property - def mutelist(self): + def mutelist(self) -> AWSMutelist: """ mutelist method returns the provider's mutelist. """ return self._mutelist + # TODO: this is going to be called from another place soon, since the provider + # shouldn't hold the mutelist @mutelist.setter def mutelist(self, mutelist_path): """ @@ -309,35 +302,12 @@ class AwsProvider(Provider): # Set default mutelist path if none is set if not mutelist_path: mutelist_path = get_default_mute_file_path(self.type) - if mutelist_path: - # Mutelist from S3 URI - if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_path): - mutelist = get_mutelist_file_from_s3( - mutelist_path, self._session.current_session - ) - # Mutelist from Lambda Function ARN - elif re.search(r"^arn:(\w+):lambda:", mutelist_path): - mutelist = get_mutelist_file_from_lambda( - mutelist_path, - self._session.current_session, - ) - # Mutelist from DynamoDB ARN - elif re.search( - r"^arn:aws(-cn|-us-gov)?:dynamodb:[a-z]{2}-[a-z-]+-[1-9]{1}:[0-9]{12}:table\/[a-zA-Z0-9._-]+$", - mutelist_path, - ): - mutelist = get_mutelist_file_from_dynamodb( - mutelist_path, self._session.current_session, self._identity.account - ) - else: - mutelist = get_mutelist_file_from_local_file(mutelist_path) - mutelist = validate_mutelist(mutelist) - else: - mutelist = {} - - self._mutelist = mutelist - self._mutelist_file_path = mutelist_path + self._mutelist = AWSMutelist( + mutelist_path=mutelist_path, + session=self._session.current_session, + aws_account_id=self._identity.account, + ) @property def get_output_mapping(self): diff --git a/prowler/providers/aws/lib/mutelist/mutelist.py b/prowler/providers/aws/lib/mutelist/mutelist.py index 0750843f02..d86e374bcd 100644 --- a/prowler/providers/aws/lib/mutelist/mutelist.py +++ b/prowler/providers/aws/lib/mutelist/mutelist.py @@ -1,84 +1,137 @@ +import re +from typing import Any + import yaml from boto3 import Session from boto3.dynamodb.conditions import Attr from prowler.lib.logger import logger +from prowler.lib.mutelist.mutelist import Mutelist +from prowler.lib.outputs.utils import unroll_tags -def get_mutelist_file_from_s3(mutelist_path: str, aws_session: Session = None): - try: - bucket = mutelist_path.split("/")[2] - key = ("/").join(mutelist_path.split("/")[3:]) - s3_client = aws_session.client("s3") - mutelist = yaml.safe_load(s3_client.get_object(Bucket=bucket, Key=key)["Body"])[ - "Mutelist" - ] - return mutelist - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" +class AWSMutelist(Mutelist): + def __init__( + self, + mutelist_content: dict = {}, + mutelist_path: str = None, + session: Session = None, + aws_account_id: str = "", + ) -> "AWSMutelist": + self._mutelist = mutelist_content + self._mutelist_file_path = mutelist_path + if mutelist_path: + # Mutelist from S3 URI + if re.search("^s3://([^/]+)/(.*?([^/]+))$", self._mutelist_file_path): + self._mutelist = self.get_mutelist_file_from_s3(session) + # Mutelist from Lambda Function ARN + elif re.search(r"^arn:(\w+):lambda:", self._mutelist_file_path): + self._mutelist = self.get_mutelist_file_from_lambda( + session, + ) + # Mutelist from DynamoDB ARN + elif re.search( + r"^arn:aws(-cn|-us-gov)?:dynamodb:[a-z]{2}-[a-z-]+-[1-9]{1}:[0-9]{12}:table\/[a-zA-Z0-9._-]+$", + self._mutelist_file_path, + ): + self._mutelist = self.get_mutelist_file_from_dynamodb( + session, + aws_account_id, + ) + else: + self.get_mutelist_file_from_local_file(mutelist_path) + if self._mutelist: + self.validate_mutelist() + + def is_finding_muted( + self, + finding: Any, + aws_account_id: str, + ) -> bool: + return self.is_muted( + aws_account_id, + finding.check_metadata.CheckID, + finding.region, + finding.resource_id, + unroll_tags(finding.resource_tags), ) - return {} - -def get_mutelist_file_from_lambda(mutelist_path: str, aws_session: Session = None): - try: - lambda_region = mutelist_path.split(":")[3] - lambda_client = aws_session.client("lambda", region_name=lambda_region) - lambda_response = lambda_client.invoke( - FunctionName=mutelist_path, InvocationType="RequestResponse" - ) - lambda_payload = lambda_response["Payload"].read() - mutelist = yaml.safe_load(lambda_payload)["Mutelist"] - - return mutelist - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return {} - - -def get_mutelist_file_from_dynamodb( - mutelist_path: str, aws_session: Session = None, aws_account: str = None -): - try: - mutelist = {"Accounts": {}} - table_region = mutelist_path.split(":")[3] - dynamodb_resource = aws_session.resource("dynamodb", region_name=table_region) - dynamo_table = dynamodb_resource.Table(mutelist_path.split("/")[1]) - response = dynamo_table.scan( - FilterExpression=Attr("Accounts").is_in([aws_account, "*"]) - ) - dynamodb_items = response["Items"] - # Paginate through all results - while "LastEvaluatedKey" in dynamodb_items: - response = dynamo_table.scan( - ExclusiveStartKey=response["LastEvaluatedKey"], - FilterExpression=Attr("Accounts").is_in([aws_account, "*"]), + def get_mutelist_file_from_s3(self, aws_session: Session = None): + try: + bucket = self._mutelist_file_path.split("/")[2] + key = ("/").join(self._mutelist_file_path.split("/")[3:]) + s3_client = aws_session.client("s3") + mutelist = yaml.safe_load( + s3_client.get_object(Bucket=bucket, Key=key)["Body"] + )["Mutelist"] + return mutelist + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) - dynamodb_items.update(response["Items"]) - for item in dynamodb_items: - # Create mutelist for every item - mutelist["Accounts"][item["Accounts"]] = { - "Checks": { - item["Checks"]: { - "Regions": item["Regions"], - "Resources": item["Resources"], + return {} + + def get_mutelist_file_from_lambda(self, aws_session: Session = None): + try: + lambda_region = self._mutelist_file_path.split(":")[3] + lambda_client = aws_session.client("lambda", region_name=lambda_region) + lambda_response = lambda_client.invoke( + FunctionName=self._mutelist_file_path, InvocationType="RequestResponse" + ) + lambda_payload = lambda_response["Payload"].read() + mutelist = yaml.safe_load(lambda_payload)["Mutelist"] + + return mutelist + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" + ) + return {} + + def get_mutelist_file_from_dynamodb( + self, aws_session: Session = None, aws_account: str = None + ): + try: + mutelist = {"Accounts": {}} + table_region = self._mutelist_file_path.split(":")[3] + dynamodb_resource = aws_session.resource( + "dynamodb", region_name=table_region + ) + dynamo_table = dynamodb_resource.Table( + self._mutelist_file_path.split("/")[1] + ) + response = dynamo_table.scan( + FilterExpression=Attr("Accounts").is_in([aws_account, "*"]) + ) + dynamodb_items = response["Items"] + # Paginate through all results + while "LastEvaluatedKey" in dynamodb_items: + response = dynamo_table.scan( + ExclusiveStartKey=response["LastEvaluatedKey"], + FilterExpression=Attr("Accounts").is_in([aws_account, "*"]), + ) + dynamodb_items.update(response["Items"]) + for item in dynamodb_items: + # Create mutelist for every item + mutelist["Accounts"][item["Accounts"]] = { + "Checks": { + item["Checks"]: { + "Regions": item["Regions"], + "Resources": item["Resources"], + } } } - } - if "Tags" in item: - mutelist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][ - "Tags" - ] = item["Tags"] - if "Exceptions" in item: - mutelist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][ - "Exceptions" - ] = item["Exceptions"] - return mutelist - except Exception as error: - logger.error( - f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" - ) - return {} + if "Tags" in item: + mutelist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][ + "Tags" + ] = item["Tags"] + if "Exceptions" in item: + mutelist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][ + "Exceptions" + ] = item["Exceptions"] + return mutelist + except Exception as error: + logger.error( + f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" + ) + return {} diff --git a/prowler/providers/azure/azure_provider.py b/prowler/providers/azure/azure_provider.py index 321bd8e587..8c2e029ecc 100644 --- a/prowler/providers/azure/azure_provider.py +++ b/prowler/providers/azure/azure_provider.py @@ -8,9 +8,13 @@ from azure.mgmt.subscription import SubscriptionClient from colorama import Fore, Style from msgraph import GraphServiceClient -from prowler.config.config import load_and_validate_config_file +from prowler.config.config import ( + get_default_mute_file_path, + load_and_validate_config_file, +) from prowler.lib.logger import logger from prowler.lib.utils.utils import print_boxes +from prowler.providers.azure.lib.mutelist.mutelist import AzureMutelist from prowler.providers.azure.lib.regions.regions import get_regions_config from prowler.providers.azure.models import ( AzureIdentityInfo, @@ -29,7 +33,7 @@ class AzureProvider(Provider): _region_config: AzureRegionConfig _locations: dict _output_options: AzureOutputOptions - _mutelist: dict + _mutelist: AzureMutelist # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -112,6 +116,24 @@ class AzureProvider(Provider): arguments, bulk_checks_metadata, self._identity ) + @property + def mutelist(self) -> AzureMutelist: + """ + mutelist method returns the provider's mutelist. + """ + return self._mutelist + + @mutelist.setter + def mutelist(self, mutelist_path): + """ + mutelist.setter sets the provider's mutelist. + """ + # Set default mutelist path if none is set + if not mutelist_path: + mutelist_path = get_default_mute_file_path(self.type) + + self._mutelist = AzureMutelist(mutelist_path) + @property def get_output_mapping(self): return { diff --git a/prowler/providers/azure/lib/mutelist/__init__.py b/prowler/providers/azure/lib/mutelist/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/azure/lib/mutelist/mutelist.py b/prowler/providers/azure/lib/mutelist/mutelist.py new file mode 100644 index 0000000000..bac563cc8b --- /dev/null +++ b/prowler/providers/azure/lib/mutelist/mutelist.py @@ -0,0 +1,18 @@ +from typing import Any + +from prowler.lib.mutelist.mutelist import Mutelist +from prowler.lib.outputs.utils import unroll_tags + + +class AzureMutelist(Mutelist): + def is_finding_muted( + self, + finding: Any, + ) -> bool: + return self.is_muted( + finding.subscription, + finding.check_metadata.CheckID, + finding.location, + finding.resource_name, + unroll_tags(finding.resource_tags), + ) diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 11f64f7c9f..628af105ec 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -5,12 +5,8 @@ from abc import ABC, abstractmethod from importlib import import_module from typing import Any, Optional -from prowler.config.config import get_default_mute_file_path from prowler.lib.logger import logger -from prowler.lib.mutelist.mutelist import ( - get_mutelist_file_from_local_file, - validate_mutelist, -) +from prowler.lib.mutelist.mutelist import Mutelist providers_path = "prowler.providers" @@ -30,8 +26,7 @@ providers_path = "prowler.providers" # TODO: enforce audit_metadata for all the providers class Provider(ABC): _global: Optional["Provider"] = None - _mutelist: dict - _mutelist_file_path: str + mutelist: Mutelist """ The Provider class is an abstract base class that defines the interface for all provider classes in the auditing system. @@ -158,37 +153,6 @@ class Provider(ABC): """ return set() - @property - def mutelist(self): - """ - mutelist method returns the provider's mutelist. - """ - return self._mutelist - - @property - def mutelist_file_path(self): - """ - mutelist method returns the provider's mutelist file path. - """ - return self._mutelist_file_path - - @mutelist.setter - def mutelist(self, mutelist_path): - """ - mutelist.setter sets the provider's mutelist. - """ - # Set default mutelist path if none is set - if not mutelist_path: - mutelist_path = get_default_mute_file_path(self.type) - if mutelist_path: - mutelist = get_mutelist_file_from_local_file(mutelist_path) - mutelist = validate_mutelist(mutelist) - else: - mutelist = {} - - self._mutelist = mutelist - self._mutelist_file_path = mutelist_path - @staticmethod def get_global_provider() -> "Provider": return Provider._global diff --git a/prowler/providers/gcp/gcp_provider.py b/prowler/providers/gcp/gcp_provider.py index d28c0903fa..e7071c48d7 100644 --- a/prowler/providers/gcp/gcp_provider.py +++ b/prowler/providers/gcp/gcp_provider.py @@ -9,11 +9,15 @@ from google.oauth2.credentials import Credentials from googleapiclient import discovery from googleapiclient.errors import HttpError -from prowler.config.config import load_and_validate_config_file +from prowler.config.config import ( + get_default_mute_file_path, + load_and_validate_config_file, +) from prowler.lib.logger import logger from prowler.lib.utils.utils import print_boxes from prowler.providers.common.models import Audit_Metadata from prowler.providers.common.provider import Provider +from prowler.providers.gcp.lib.mutelist.mutelist import GCPMutelist from prowler.providers.gcp.models import ( GCPIdentityInfo, GCPOrganization, @@ -30,7 +34,7 @@ class GcpProvider(Provider): _identity: GCPIdentityInfo _audit_config: dict _output_options: GCPOutputOptions - _mutelist: dict + _mutelist: GCPMutelist # TODO: this is not optional, enforce for all providers audit_metadata: Audit_Metadata @@ -155,6 +159,24 @@ class GcpProvider(Provider): arguments, bulk_checks_metadata, self._identity ) + @property + def mutelist(self) -> GCPMutelist: + """ + mutelist method returns the provider's mutelist. + """ + return self._mutelist + + @mutelist.setter + def mutelist(self, mutelist_path): + """ + mutelist.setter sets the provider's mutelist. + """ + # Set default mutelist path if none is set + if not mutelist_path: + mutelist_path = get_default_mute_file_path(self.type) + + self._mutelist = GCPMutelist(mutelist_path) + @property def get_output_mapping(self): return { diff --git a/prowler/providers/gcp/lib/mutelist/__init__.py b/prowler/providers/gcp/lib/mutelist/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/gcp/lib/mutelist/mutelist.py b/prowler/providers/gcp/lib/mutelist/mutelist.py new file mode 100644 index 0000000000..513e33c297 --- /dev/null +++ b/prowler/providers/gcp/lib/mutelist/mutelist.py @@ -0,0 +1,18 @@ +from typing import Any + +from prowler.lib.mutelist.mutelist import Mutelist +from prowler.lib.outputs.utils import unroll_tags + + +class GCPMutelist(Mutelist): + def is_finding_muted( + self, + finding: Any, + ) -> bool: + return self.is_muted( + finding.project_id, + finding.check_metadata.CheckID, + finding.location, + finding.resource_name, + unroll_tags(finding.resource_tags), + ) diff --git a/prowler/providers/kubernetes/kubernetes_provider.py b/prowler/providers/kubernetes/kubernetes_provider.py index c00ba3fadf..65625e7b84 100644 --- a/prowler/providers/kubernetes/kubernetes_provider.py +++ b/prowler/providers/kubernetes/kubernetes_provider.py @@ -6,11 +6,15 @@ from colorama import Fore, Style from kubernetes.config.config_exception import ConfigException from kubernetes import client, config -from prowler.config.config import load_and_validate_config_file +from prowler.config.config import ( + get_default_mute_file_path, + load_and_validate_config_file, +) from prowler.lib.logger import logger from prowler.lib.utils.utils import print_boxes from prowler.providers.common.models import Audit_Metadata from prowler.providers.common.provider import Provider +from prowler.providers.kubernetes.lib.mutelist.mutelist import KubernetesMutelist from prowler.providers.kubernetes.models import ( KubernetesIdentityInfo, KubernetesOutputOptions, @@ -97,6 +101,24 @@ class KubernetesProvider(Provider): arguments, bulk_checks_metadata, self._identity ) + @property + def mutelist(self) -> KubernetesMutelist: + """ + mutelist method returns the provider's mutelist. + """ + return self._mutelist + + @mutelist.setter + def mutelist(self, mutelist_path): + """ + mutelist.setter sets the provider's mutelist. + """ + # Set default mutelist path if none is set + if not mutelist_path: + mutelist_path = get_default_mute_file_path(self.type) + + self._mutelist = KubernetesMutelist(mutelist_path) + @property def get_output_mapping(self): return { diff --git a/prowler/providers/kubernetes/lib/mutelist/__init__.py b/prowler/providers/kubernetes/lib/mutelist/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/kubernetes/lib/mutelist/mutelist.py b/prowler/providers/kubernetes/lib/mutelist/mutelist.py new file mode 100644 index 0000000000..0fc50f7332 --- /dev/null +++ b/prowler/providers/kubernetes/lib/mutelist/mutelist.py @@ -0,0 +1,19 @@ +from typing import Any + +from prowler.lib.mutelist.mutelist import Mutelist +from prowler.lib.outputs.utils import unroll_tags + + +class KubernetesMutelist(Mutelist): + def is_finding_muted( + self, + finding: Any, + cluster: str, + ) -> bool: + return self.is_muted( + cluster, + finding.check_metadata.CheckID, + finding.namespace, + finding.resource_name, + unroll_tags(finding.resource_tags), + ) diff --git a/tests/lib/mutelist/mutelist_test.py b/tests/lib/mutelist/mutelist_test.py deleted file mode 100644 index dcb78aed61..0000000000 --- a/tests/lib/mutelist/mutelist_test.py +++ /dev/null @@ -1,1244 +0,0 @@ -import yaml -from mock import MagicMock - -from prowler.lib.mutelist.mutelist import ( - get_mutelist_file_from_local_file, - is_excepted, - is_muted, - is_muted_in_check, - is_muted_in_region, - is_muted_in_resource, - is_muted_in_tags, - mutelist_findings, - validate_mutelist, -) -from tests.providers.aws.utils import ( - AWS_ACCOUNT_NUMBER, - AWS_REGION_EU_CENTRAL_1, - AWS_REGION_EU_SOUTH_3, - AWS_REGION_EU_WEST_1, - AWS_REGION_US_EAST_1, - set_mocked_aws_provider, -) - - -class TestMutelist: - def test_get_mutelist_file_from_local_file(self): - mutelist_path = "tests/lib/mutelist/fixtures/aws_mutelist.yaml" - with open(mutelist_path) as f: - mutelist_fixture = yaml.safe_load(f)["Mutelist"] - - assert get_mutelist_file_from_local_file(mutelist_path) == mutelist_fixture - - def test_get_mutelist_file_from_local_file_non_existent(self): - mutelist_path = "tests/lib/mutelist/fixtures/not_present" - - assert get_mutelist_file_from_local_file(mutelist_path) == {} - - def test_validate_mutelist(self): - mutelist_path = "tests/lib/mutelist/fixtures/aws_mutelist.yaml" - with open(mutelist_path) as f: - mutelist_fixture = yaml.safe_load(f)["Mutelist"] - - assert validate_mutelist(mutelist_fixture) == mutelist_fixture - - def test_validate_mutelist_not_valid_key(self): - mutelist_path = "tests/lib/mutelist/fixtures/aws_mutelist.yaml" - with open(mutelist_path) as f: - mutelist_fixture = yaml.safe_load(f)["Mutelist"] - - mutelist_fixture["Accounts1"] = mutelist_fixture["Accounts"] - del mutelist_fixture["Accounts"] - assert validate_mutelist(mutelist_fixture) == {} - - def test_mutelist_findings_only_wildcard(self): - - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["prowler", "^test", "prowler-pro"], - } - } - } - } - } - - # Check Findings - check_findings = [] - finding_1 = MagicMock - finding_1.check_metadata = MagicMock - finding_1.check_metadata.CheckID = "check_test" - finding_1.status = "FAIL" - finding_1.region = AWS_REGION_US_EAST_1 - finding_1.resource_id = "prowler" - finding_1.resource_tags = [] - aws_provider = set_mocked_aws_provider() - aws_provider._mutelist = mutelist - - check_findings.append(finding_1) - - muted_findings = mutelist_findings(aws_provider, check_findings) - assert len(muted_findings) == 1 - assert muted_findings[0].status == "FAIL" - assert muted_findings[0].muted - - def test_mutelist_all_exceptions_empty(self): - - mutelist = { - "Accounts": { - "*": { - "Checks": { - "*": { - "Tags": ["*"], - "Regions": [AWS_REGION_US_EAST_1], - "Resources": ["*"], - "Exceptions": { - "Tags": [], - "Regions": [], - "Accounts": [], - "Resources": [], - }, - } - } - } - } - } - - # Check Findings - check_findings = [] - finding_1 = MagicMock - finding_1.check_metadata = MagicMock - finding_1.check_metadata.CheckID = "check_test" - finding_1.status = "FAIL" - finding_1.region = AWS_REGION_US_EAST_1 - finding_1.resource_id = "prowler" - finding_1.resource_tags = [] - aws_provider = set_mocked_aws_provider() - aws_provider._mutelist = mutelist - - check_findings.append(finding_1) - - muted_findings = mutelist_findings(aws_provider, check_findings) - assert len(muted_findings) == 1 - assert muted_findings[0].status == "FAIL" - assert muted_findings[0].muted - - def test_is_muted_with_everything_excepted(self): - mutelist = { - "Accounts": { - "*": { - "Checks": { - "athena_*": { - "Regions": "*", - "Resources": "*", - "Tags": "*", - "Exceptions": { - "Accounts": ["*"], - "Regions": ["*"], - "Resources": ["*"], - "Tags": ["*"], - }, - } - } - } - } - } - - assert not is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "athena_1", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - def test_is_muted_with_default_mutelist(self): - mutelist = { - "Accounts": { - "*": { - "Checks": { - "*": { - "Tags": ["*"], - "Regions": ["*"], - "Resources": ["*"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "athena_1", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - def test_is_muted_with_default_mutelist_with_tags(self): - mutelist = { - "Accounts": { - "*": { - "Checks": { - "*": { - "Regions": ["*"], - "Resources": ["*"], - "Tags": ["Compliance=allow"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "athena_1", - AWS_REGION_US_EAST_1, - "prowler", - "Compliance=allow", - ) - - assert not is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "athena_1", - AWS_REGION_US_EAST_1, - "prowler", - "Compliance=deny", - ) - - def test_is_muted(self): - - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["prowler", "^test", "prowler-pro"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-test", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "test-prowler", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-pro-test", - "", - ) - - assert not ( - is_muted( - mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "" - ) - ) - - def test_is_muted_wildcard(self): - # Mutelist example - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": [".*"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-test", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "test-prowler", - "", - ) - - assert not ( - is_muted( - mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "" - ) - ) - - def test_is_muted_asterisk(self): - # Mutelist example - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-test", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "test-prowler", - "", - ) - - assert not ( - is_muted( - mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "" - ) - ) - - def test_is_muted_exceptions_before_match(self): - # Mutelist example - mutelist = { - "Accounts": { - "*": { - "Checks": { - "accessanalyzer_enabled": { - "Exceptions": { - "Accounts": [], - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": [], - "Tags": [], - }, - "Regions": ["*"], - "Resources": ["*"], - "Tags": ["*"], - }, - "sns_*": { - "Regions": ["*"], - "Resources": ["aws-controltower-*"], - }, - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "sns_topics_not_publicly_accessible", - AWS_REGION_EU_WEST_1, - "aws-controltower-AggregateSecurityNotifications", - "", - ) - - def test_is_muted_all_and_single_account(self): - # Mutelist example - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test_2": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - } - } - }, - AWS_ACCOUNT_NUMBER: { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1], - "Resources": ["*"], - } - } - }, - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test_2", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-test", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "test-prowler", - "", - ) - - assert not ( - is_muted( - mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "" - ) - ) - - def test_is_muted_all_and_single_account_with_different_resources(self): - - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test_1": { - "Regions": ["*"], - "Resources": ["resource_1", "resource_2"], - }, - } - }, - AWS_ACCOUNT_NUMBER: { - "Checks": { - "check_test_1": { - "Regions": ["*"], - "Resources": ["resource_3"], - } - } - }, - } - } - - assert is_muted( - mutelist, - "111122223333", - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_1", - "", - ) - - assert is_muted( - mutelist, - "111122223333", - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_2", - "", - ) - - assert not is_muted( - mutelist, - "111122223333", - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_3", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_3", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_2", - "", - ) - - def test_is_muted_all_and_single_account_with_different_resources_and_exceptions( - self, - ): - - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test_1": { - "Regions": ["*"], - "Resources": ["resource_1", "resource_2"], - "Exceptions": {"Regions": [AWS_REGION_US_EAST_1]}, - }, - } - }, - AWS_ACCOUNT_NUMBER: { - "Checks": { - "check_test_1": { - "Regions": ["*"], - "Resources": ["resource_3"], - "Exceptions": {"Regions": [AWS_REGION_EU_WEST_1]}, - } - } - }, - } - } - - assert not is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_2", - "", - ) - - assert not is_muted( - mutelist, - "111122223333", - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_1", - "", - ) - - assert is_muted( - mutelist, - "111122223333", - "check_test_1", - AWS_REGION_EU_WEST_1, - "resource_2", - "", - ) - - assert not is_muted( - mutelist, - "111122223333", - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_3", - "", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test_1", - AWS_REGION_US_EAST_1, - "resource_3", - "", - ) - - assert not is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test_1", - AWS_REGION_EU_WEST_1, - "resource_3", - "", - ) - - def test_is_muted_single_account(self): - mutelist = { - "Accounts": { - AWS_ACCOUNT_NUMBER: { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1], - "Resources": ["prowler"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert not ( - is_muted( - mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "" - ) - ) - - def test_is_muted_in_region(self): - muted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1] - finding_region = AWS_REGION_US_EAST_1 - - assert is_muted_in_region(muted_regions, finding_region) - - def test_is_muted_in_region_wildcard(self): - muted_regions = ["*"] - finding_region = AWS_REGION_US_EAST_1 - - assert is_muted_in_region(muted_regions, finding_region) - - def test_is_not_muted_in_region(self): - muted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1] - finding_region = "eu-west-2" - - assert not is_muted_in_region(muted_regions, finding_region) - - def test_is_muted_in_check(self): - muted_checks = { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - } - } - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-test", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "test-prowler", - "", - ) - - assert not ( - is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "check_test", - "us-east-2", - "test", - "", - ) - ) - - def test_is_muted_in_check_regex(self): - # Mutelist example - muted_checks = { - "s3_*": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - } - } - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "s3_bucket_public_access", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "s3_bucket_no_mfa_delete", - AWS_REGION_US_EAST_1, - "prowler-test", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "s3_bucket_policy_public_write_access", - AWS_REGION_US_EAST_1, - "test-prowler", - "", - ) - - assert not ( - is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "iam_user_hardware_mfa_enabled", - AWS_REGION_US_EAST_1, - "test", - "", - ) - ) - - def test_is_muted_lambda_generic_check(self): - muted_checks = { - "lambda_*": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - } - } - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_no_secrets_in_code", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_no_secrets_in_variables", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_not_publicly_accessible", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_url_cors_policy", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_url_public", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_using_supported_runtimes", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - def test_is_muted_lambda_concrete_check(self): - muted_checks = { - "lambda_function_no_secrets_in_variables": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - } - } - - assert is_muted_in_check( - muted_checks, - AWS_ACCOUNT_NUMBER, - "awslambda_function_no_secrets_in_variables", - AWS_REGION_US_EAST_1, - "prowler", - "", - ) - - def test_is_muted_tags(self): - # Mutelist example - mutelist = { - "Accounts": { - "*": { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], - "Resources": ["*"], - "Tags": ["environment=dev", "project=.*"], - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler", - "environment=dev", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_US_EAST_1, - "prowler-test", - "environment=dev | project=prowler", - ) - - assert not ( - is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - "us-east-2", - "test", - "environment=pro", - ) - ) - - def test_is_muted_specific_account_with_other_account_excepted(self): - - mutelist = { - "Accounts": { - AWS_ACCOUNT_NUMBER: { - "Checks": { - "check_test": { - "Regions": [AWS_REGION_EU_WEST_1], - "Resources": ["*"], - "Tags": [], - "Exceptions": {"Accounts": ["111122223333"]}, - } - } - } - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "check_test", - AWS_REGION_EU_WEST_1, - "prowler", - "environment=dev", - ) - - assert not is_muted( - mutelist, - "111122223333", - "check_test", - AWS_REGION_EU_WEST_1, - "prowler", - "environment=dev", - ) - - def test_is_muted_complex_mutelist(self): - - mutelist = { - "Accounts": { - "*": { - "Checks": { - "s3_bucket_object_versioning": { - "Regions": [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1], - "Resources": ["ci-logs", "logs", ".+-logs"], - }, - "ecs_task_definitions_no_environment_secrets": { - "Regions": ["*"], - "Resources": ["*"], - "Exceptions": { - "Accounts": [AWS_ACCOUNT_NUMBER], - "Regions": [ - AWS_REGION_EU_WEST_1, - AWS_REGION_EU_SOUTH_3, - ], - }, - }, - "*": { - "Regions": ["*"], - "Resources": ["*"], - "Tags": ["environment=dev"], - }, - } - }, - AWS_ACCOUNT_NUMBER: { - "Checks": { - "*": { - "Regions": ["*"], - "Resources": ["*"], - "Exceptions": { - "Resources": ["test"], - "Tags": ["environment=prod"], - }, - } - } - }, - } - } - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "test_check", - AWS_REGION_EU_WEST_1, - "prowler-logs", - "environment=dev", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "ecs_task_definitions_no_environment_secrets", - AWS_REGION_EU_WEST_1, - "prowler", - "environment=dev", - ) - - assert is_muted( - mutelist, - AWS_ACCOUNT_NUMBER, - "s3_bucket_object_versioning", - AWS_REGION_EU_WEST_1, - "prowler-logs", - "environment=dev", - ) - - def test_is_muted_in_tags(self): - mutelist_tags = ["environment=dev", "project=prowler"] - - assert is_muted_in_tags(mutelist_tags, "environment=dev") - - assert is_muted_in_tags( - mutelist_tags, - "environment=dev | project=prowler", - ) - - assert not ( - is_muted_in_tags( - mutelist_tags, - "environment=pro", - ) - ) - - def test_is_muted_in_tags_regex(self): - mutelist_tags = ["environment=(dev|test)", ".*=prowler"] - - assert is_muted_in_tags( - mutelist_tags, - "environment=test | proj=prowler", - ) - - assert is_muted_in_tags( - mutelist_tags, - "env=prod | project=prowler", - ) - - assert not is_muted_in_tags( - mutelist_tags, - "environment=prod | project=myproj", - ) - - def test_is_muted_in_tags_with_no_tags_in_finding(self): - mutelist_tags = ["environment=(dev|test)", ".*=prowler"] - finding_tags = "" - - assert not is_muted_in_tags(mutelist_tags, finding_tags) - - def test_is_excepted(self): - # Mutelist example - exceptions = { - "Accounts": [AWS_ACCOUNT_NUMBER], - "Regions": ["eu-central-1", "eu-south-3"], - "Resources": ["test"], - "Tags": ["environment=test", "project=.*"], - } - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-central-1", - "test", - "environment=test", - ) - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-south-3", - "test", - "environment=test", - ) - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-south-3", - "test123", - "environment=test", - ) - - def test_is_excepted_only_in_account(self): - - exceptions = { - "Accounts": [AWS_ACCOUNT_NUMBER], - "Regions": [], - "Resources": [], - "Tags": [], - } - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-central-1", - "test", - "environment=test", - ) - - def test_is_excepted_only_in_region(self): - - exceptions = { - "Accounts": [], - "Regions": [AWS_REGION_EU_CENTRAL_1, AWS_REGION_EU_SOUTH_3], - "Resources": [], - "Tags": [], - } - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - AWS_REGION_EU_CENTRAL_1, - "test", - "environment=test", - ) - - def test_is_excepted_only_in_resources(self): - - exceptions = { - "Accounts": [], - "Regions": [], - "Resources": ["resource_1"], - "Tags": [], - } - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - AWS_REGION_EU_CENTRAL_1, - "resource_1", - "environment=test", - ) - - def test_is_excepted_only_in_tags(self): - - exceptions = { - "Accounts": [], - "Regions": [], - "Resources": [], - "Tags": ["environment=test"], - } - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - AWS_REGION_EU_CENTRAL_1, - "resource_1", - "environment=test", - ) - - def test_is_excepted_in_account_and_tags(self): - - exceptions = { - "Accounts": [AWS_ACCOUNT_NUMBER], - "Regions": [], - "Resources": [], - "Tags": ["environment=test"], - } - - assert is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - AWS_REGION_EU_CENTRAL_1, - "resource_1", - "environment=test", - ) - - assert not is_excepted( - exceptions, - "111122223333", - AWS_REGION_EU_CENTRAL_1, - "resource_1", - "environment=test", - ) - - assert not is_excepted( - exceptions, - "111122223333", - AWS_REGION_EU_CENTRAL_1, - "resource_1", - "environment=dev", - ) - - def test_is_excepted_all_wildcard(self): - exceptions = { - "Accounts": ["*"], - "Regions": ["*"], - "Resources": ["*"], - "Tags": ["*"], - } - assert is_excepted( - exceptions, AWS_ACCOUNT_NUMBER, "eu-south-2", "test", "environment=test" - ) - assert not is_excepted( - exceptions, AWS_ACCOUNT_NUMBER, "eu-south-2", "test", None - ) - - def test_is_not_excepted(self): - exceptions = { - "Accounts": [AWS_ACCOUNT_NUMBER], - "Regions": ["eu-central-1", "eu-south-3"], - "Resources": ["test"], - "Tags": ["environment=test", "project=.*"], - } - - assert not is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-south-2", - "test", - "environment=test", - ) - - assert not is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-south-3", - "prowler", - "environment=test", - ) - - assert not is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-south-3", - "test", - "environment=pro", - ) - - def test_is_excepted_all_empty(self): - exceptions = { - "Accounts": [], - "Regions": [], - "Resources": [], - "Tags": [], - } - - assert not is_excepted( - exceptions, - AWS_ACCOUNT_NUMBER, - "eu-south-2", - "test", - "environment=test", - ) - - def test_is_muted_in_resource(self): - mutelist_resources = ["prowler", "^test", "prowler-pro"] - - assert is_muted_in_resource(mutelist_resources, "prowler") - assert is_muted_in_resource(mutelist_resources, "prowler-test") - assert is_muted_in_resource(mutelist_resources, "test-prowler") - assert not is_muted_in_resource(mutelist_resources, "random") - - def test_is_muted_in_resource_starting_by_star(self): - allowlist_resources = ["*.es"] - - assert is_muted_in_resource(allowlist_resources, "google.es") diff --git a/tests/providers/aws/aws_provider_test.py b/tests/providers/aws/aws_provider_test.py index 1977b78232..829f1cd458 100644 --- a/tests/providers/aws/aws_provider_test.py +++ b/tests/providers/aws/aws_provider_test.py @@ -27,6 +27,7 @@ from prowler.providers.aws.config import ( BOTO3_USER_AGENT_EXTRA, ) from prowler.providers.aws.lib.arn.models import ARN +from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist from prowler.providers.aws.models import ( AWSAssumeRoleInfo, AWSCallerIdentity, @@ -607,7 +608,9 @@ aws: os.remove(mutelist_file.name) - assert aws_provider.mutelist == mutelist["Mutelist"] + assert isinstance(aws_provider.mutelist, AWSMutelist) + assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] + assert aws_provider.mutelist.mutelist_file_path == mutelist_file.name @mock_aws def test_aws_provider_mutelist_none(self): @@ -620,7 +623,9 @@ aws: ): aws_provider.mutelist = None - assert aws_provider.mutelist == {} + assert isinstance(aws_provider.mutelist, AWSMutelist) + assert aws_provider.mutelist.mutelist == {} + assert aws_provider.mutelist.mutelist_file_path is None @mock_aws def test_aws_provider_mutelist_s3(self): @@ -670,7 +675,9 @@ aws: aws_provider.mutelist = mutelist_bucket_object_uri os.remove(mutelist_file.name) - assert aws_provider.mutelist == mutelist["Mutelist"] + assert isinstance(aws_provider.mutelist, AWSMutelist) + assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] + assert aws_provider.mutelist.mutelist_file_path == mutelist_bucket_object_uri @mock_aws def test_aws_provider_mutelist_lambda(self): @@ -696,17 +703,19 @@ aws: } } } - + lambda_mutelist_path = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function:lambda-mutelist" arguments = Namespace() aws_provider = AwsProvider(arguments) with patch( - "prowler.providers.aws.aws_provider.get_mutelist_file_from_lambda", + "prowler.providers.aws.lib.mutelist.mutelist.AWSMutelist.get_mutelist_file_from_lambda", return_value=mutelist["Mutelist"], ): - aws_provider.mutelist = f"arn:aws:lambda:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:function:lambda-mutelist" + aws_provider.mutelist = lambda_mutelist_path - assert aws_provider.mutelist == mutelist["Mutelist"] + assert isinstance(aws_provider.mutelist, AWSMutelist) + assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] + assert aws_provider.mutelist.mutelist_file_path == lambda_mutelist_path @mock_aws def test_aws_provider_mutelist_dynamodb(self): @@ -732,17 +741,19 @@ aws: } } } - + dynamodb_mutelist_path = f"arn:aws:dynamodb:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:table/mutelist-dynamo" arguments = Namespace() aws_provider = AwsProvider(arguments) with patch( - "prowler.providers.aws.aws_provider.get_mutelist_file_from_dynamodb", + "prowler.providers.aws.lib.mutelist.mutelist.AWSMutelist.get_mutelist_file_from_dynamodb", return_value=mutelist["Mutelist"], ): - aws_provider.mutelist = f"arn:aws:dynamodb:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:table/mutelist-dynamo" + aws_provider.mutelist = dynamodb_mutelist_path - assert aws_provider.mutelist == mutelist["Mutelist"] + assert isinstance(aws_provider.mutelist, AWSMutelist) + assert aws_provider.mutelist.mutelist == mutelist["Mutelist"] + assert aws_provider.mutelist.mutelist_file_path == dynamodb_mutelist_path @mock_aws def test_generate_regional_clients_all_enabled_regions(self): diff --git a/tests/providers/aws/lib/mutelist/aws_mutelist_test.py b/tests/providers/aws/lib/mutelist/aws_mutelist_test.py index 62fd8e49a5..d5ecea0efb 100644 --- a/tests/providers/aws/lib/mutelist/aws_mutelist_test.py +++ b/tests/providers/aws/lib/mutelist/aws_mutelist_test.py @@ -4,20 +4,18 @@ from json import dumps import botocore import yaml from boto3 import client, resource -from mock import patch +from mock import MagicMock, patch from moto import mock_aws from prowler.config.config import enconding_format_utf_8 -from prowler.providers.aws.lib.mutelist.mutelist import ( - get_mutelist_file_from_dynamodb, - get_mutelist_file_from_lambda, - get_mutelist_file_from_s3, -) +from prowler.providers.aws.lib.mutelist.mutelist import AWSMutelist from tests.providers.aws.services.awslambda.awslambda_service_test import ( create_zip_file, ) from tests.providers.aws.utils import ( AWS_ACCOUNT_NUMBER, + AWS_REGION_EU_CENTRAL_1, + AWS_REGION_EU_SOUTH_3, AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1, set_mocked_aws_provider, @@ -25,6 +23,8 @@ from tests.providers.aws.utils import ( make_api_call = botocore.client.BaseClient._make_api_call +MUTELIST_FIXTURE_PATH = "tests/providers/aws/lib/mutelist/fixtures/aws_mutelist.yaml" + def mock_make_api_call(self, operation_name, kwarg): if operation_name == "Invoke": @@ -53,7 +53,7 @@ def mock_make_api_call(self, operation_name, kwarg): return make_api_call(self, operation_name, kwarg) -class TestMutelistAWS: +class TestAWSMutelist: @mock_aws def test_get_mutelist_file_from_s3(self): aws_provider = set_mocked_aws_provider() @@ -62,33 +62,31 @@ class TestMutelistAWS: s3_resource.create_bucket(Bucket="test-mutelist") s3_resource.Object("test-mutelist", "mutelist.yaml").put( Body=open( - "tests/lib/mutelist/fixtures/aws_mutelist.yaml", + MUTELIST_FIXTURE_PATH, "rb", ) ) - with open("tests/lib/mutelist/fixtures/aws_mutelist.yaml") as f: + with open(MUTELIST_FIXTURE_PATH) as f: fixture_mutelist = yaml.safe_load(f)["Mutelist"] - - assert ( - get_mutelist_file_from_s3( - "s3://test-mutelist/mutelist.yaml", - aws_provider.session.current_session, - ) - == fixture_mutelist + mutelist_path = "s3://test-mutelist/mutelist.yaml" + mutelist = AWSMutelist( + mutelist_path=mutelist_path, session=aws_provider.session.current_session ) + assert mutelist.mutelist == fixture_mutelist + assert mutelist.mutelist_file_path == mutelist_path + @mock_aws def test_get_mutelist_file_from_s3_not_present(self): aws_provider = set_mocked_aws_provider() + mutelist_path = "s3://test-mutelist/mutelist.yaml" - assert ( - get_mutelist_file_from_s3( - "s3://test-mutelist/mutelist.yaml", - aws_provider.session.current_session, - ) - == {} + mutelist = AWSMutelist( + mutelist_path=mutelist_path, session=aws_provider.session.current_session ) + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path == mutelist_path @mock_aws def test_get_mutelist_file_from_dynamodb(self): @@ -120,7 +118,7 @@ class TestMutelistAWS: "Resources": ["keyword"], "Exceptions": {}, } - mutelist = { + mutelist_content = { "Accounts": { "*": { "Checks": { @@ -135,14 +133,13 @@ class TestMutelistAWS: } table.put_item(Item=dynamo_db_mutelist) - assert ( - get_mutelist_file_from_dynamodb( - table_arn, - aws_provider.session.current_session, - aws_provider.identity.account, - ) - == mutelist + mutelist = AWSMutelist( + mutelist_path=table_arn, + session=aws_provider.session.current_session, + aws_account_id=aws_provider.identity.account, ) + assert mutelist.mutelist == mutelist_content + assert mutelist.mutelist_file_path == table_arn @mock_aws def test_get_mutelist_file_from_dynamodb_with_tags(self): @@ -174,7 +171,7 @@ class TestMutelistAWS: "Resources": ["*"], "Tags": ["environment=dev"], } - mutelist = { + mutelist_content = { "Accounts": { "*": { "Checks": { @@ -189,28 +186,26 @@ class TestMutelistAWS: } table.put_item(Item=dynamo_db_mutelist) - assert ( - get_mutelist_file_from_dynamodb( - table_arn, - aws_provider.session.current_session, - aws_provider.identity.account, - ) - == mutelist + mutelist = AWSMutelist( + mutelist_path=table_arn, + session=aws_provider.session.current_session, + aws_account_id=aws_provider.identity.account, ) + assert mutelist.mutelist == mutelist_content + assert mutelist.mutelist_file_path == table_arn @mock_aws def test_get_mutelist_file_from_dynamodb_not_present(self): aws_provider = set_mocked_aws_provider() table_name = "non-existent" table_arn = f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/{table_name}" - assert ( - get_mutelist_file_from_dynamodb( - table_arn, - aws_provider.session.current_session, - aws_provider.identity.account, - ) - == {} + mutelist = AWSMutelist( + mutelist_path=table_arn, + session=aws_provider.session.current_session, + aws_account_id=aws_provider.identity.account, ) + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path == table_arn @mock_aws(config={"lambda": {"use_docker": False}}) @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) @@ -249,7 +244,7 @@ class TestMutelistAWS: Description="test lambda function", ) lambda_function_arn = lambda_function["FunctionArn"] - mutelist = { + mutelist_content = { "Accounts": { "*": { "Checks": { @@ -263,21 +258,1216 @@ class TestMutelistAWS: } } - assert ( - get_mutelist_file_from_lambda( - lambda_function_arn, aws_provider.session.current_session - ) - == mutelist + mutelist = AWSMutelist( + mutelist_path=lambda_function_arn, + session=aws_provider.session.current_session, + aws_account_id=aws_provider.identity.account, ) + assert mutelist.mutelist == mutelist_content + assert mutelist.mutelist_file_path == lambda_function_arn @mock_aws def test_get_mutelist_file_from_lambda_invalid_arn(self): aws_provider = set_mocked_aws_provider() lambda_function_arn = "invalid_arn" - assert ( - get_mutelist_file_from_lambda( - lambda_function_arn, aws_provider.session.current_session - ) - == {} + mutelist = AWSMutelist( + mutelist_path=lambda_function_arn, + session=aws_provider.session.current_session, + aws_account_id=aws_provider.identity.account, ) + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path == lambda_function_arn + + def test_get_mutelist_file_from_local_file(self): + mutelist_path = MUTELIST_FIXTURE_PATH + mutelist = AWSMutelist(mutelist_path=mutelist_path) + + with open(mutelist_path) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + assert mutelist.mutelist == mutelist_fixture + + def test_get_mutelist_file_from_local_file_non_existent(self): + mutelist_path = "tests/lib/mutelist/fixtures/not_present" + mutelist = AWSMutelist(mutelist_path=mutelist_path) + + assert mutelist.mutelist == {} + + def test_validate_mutelist(self): + mutelist_path = MUTELIST_FIXTURE_PATH + + with open(mutelist_path) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + mutelist = AWSMutelist(mutelist_content=mutelist_fixture) + + assert mutelist.validate_mutelist() + assert mutelist.mutelist == mutelist_fixture + + def test_validate_mutelist_not_valid_key(self): + mutelist_path = MUTELIST_FIXTURE_PATH + with open(mutelist_path) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + mutelist_fixture["Accounts1"] = mutelist_fixture["Accounts"] + del mutelist_fixture["Accounts"] + + mutelist = AWSMutelist(mutelist_content=mutelist_fixture) + + assert not mutelist.validate_mutelist() + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path is None + + def test_mutelist_findings_only_wildcard(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "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 = MagicMock + finding_1.check_metadata = MagicMock + finding_1.check_metadata.CheckID = "check_test" + finding_1.status = "FAIL" + finding_1.region = AWS_REGION_US_EAST_1 + finding_1.resource_id = "prowler" + finding_1.resource_tags = [] + + assert mutelist.is_finding_muted(finding_1, AWS_ACCOUNT_NUMBER) + + def test_mutelist_all_exceptions_empty(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "*": { + "Tags": ["*"], + "Regions": [AWS_REGION_US_EAST_1], + "Resources": ["*"], + "Exceptions": { + "Tags": [], + "Regions": [], + "Accounts": [], + "Resources": [], + }, + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + # Check Findings + finding_1 = MagicMock + finding_1.check_metadata = MagicMock + finding_1.check_metadata.CheckID = "check_test" + finding_1.status = "FAIL" + finding_1.region = AWS_REGION_US_EAST_1 + finding_1.resource_id = "prowler" + finding_1.resource_tags = [] + + assert mutelist.is_finding_muted(finding_1, AWS_ACCOUNT_NUMBER) + + def test_is_muted_with_everything_excepted(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "athena_*": { + "Regions": "*", + "Resources": "*", + "Tags": "*", + "Exceptions": { + "Accounts": ["*"], + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["*"], + }, + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert not mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "athena_1", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + def test_is_muted_with_default_mutelist(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "*": { + "Tags": ["*"], + "Regions": ["*"], + "Resources": ["*"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "athena_1", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + def test_is_muted_with_default_mutelist_with_tags(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "*": { + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["Compliance=allow"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "athena_1", + AWS_REGION_US_EAST_1, + "prowler", + "Compliance=allow", + ) + + assert not mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "athena_1", + AWS_REGION_US_EAST_1, + "prowler", + "Compliance=deny", + ) + + def test_is_muted(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "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) + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-test", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "test-prowler", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-pro-test", + "", + ) + + assert not ( + mutelist.is_muted(AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "") + ) + + def test_is_muted_wildcard(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": [".*"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-test", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "test-prowler", + "", + ) + + assert not ( + mutelist.is_muted(AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "") + ) + + def test_is_muted_asterisk(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-test", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "test-prowler", + "", + ) + + assert not ( + mutelist.is_muted(AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "") + ) + + def test_is_muted_exceptions_before_match(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "accessanalyzer_enabled": { + "Exceptions": { + "Accounts": [], + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": [], + "Tags": [], + }, + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["*"], + }, + "sns_*": { + "Regions": ["*"], + "Resources": ["aws-controltower-*"], + }, + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "sns_topics_not_publicly_accessible", + AWS_REGION_EU_WEST_1, + "aws-controltower-AggregateSecurityNotifications", + "", + ) + + def test_is_muted_all_and_single_account(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "check_test_2": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + } + } + }, + AWS_ACCOUNT_NUMBER: { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1], + "Resources": ["*"], + } + } + }, + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test_2", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-test", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "test-prowler", + "", + ) + + assert not ( + mutelist.is_muted(AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "") + ) + + def test_is_muted_all_and_single_account_with_different_resources(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "check_test_1": { + "Regions": ["*"], + "Resources": ["resource_1", "resource_2"], + }, + } + }, + AWS_ACCOUNT_NUMBER: { + "Checks": { + "check_test_1": { + "Regions": ["*"], + "Resources": ["resource_3"], + } + } + }, + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + "111122223333", + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_1", + "", + ) + + assert mutelist.is_muted( + "111122223333", + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_2", + "", + ) + + assert not mutelist.is_muted( + "111122223333", + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_3", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_3", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_2", + "", + ) + + def test_is_muted_all_and_single_account_with_different_resources_and_exceptions( + self, + ): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "check_test_1": { + "Regions": ["*"], + "Resources": ["resource_1", "resource_2"], + "Exceptions": {"Regions": [AWS_REGION_US_EAST_1]}, + }, + } + }, + AWS_ACCOUNT_NUMBER: { + "Checks": { + "check_test_1": { + "Regions": ["*"], + "Resources": ["resource_3"], + "Exceptions": {"Regions": [AWS_REGION_EU_WEST_1]}, + } + } + }, + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert not mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_2", + "", + ) + + assert not mutelist.is_muted( + "111122223333", + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_1", + "", + ) + + assert mutelist.is_muted( + "111122223333", + "check_test_1", + AWS_REGION_EU_WEST_1, + "resource_2", + "", + ) + + assert not mutelist.is_muted( + "111122223333", + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_3", + "", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test_1", + AWS_REGION_US_EAST_1, + "resource_3", + "", + ) + + assert not mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test_1", + AWS_REGION_EU_WEST_1, + "resource_3", + "", + ) + + def test_is_muted_single_account(self): + # Mutelist + mutelist_content = { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1], + "Resources": ["prowler"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert not ( + mutelist.is_muted(AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", "") + ) + + def test_is_muted_in_region(self): + muted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1] + finding_region = AWS_REGION_US_EAST_1 + + assert AWSMutelist.is_item_matched(muted_regions, finding_region) + + def test_is_muted_in_region_wildcard(self): + muted_regions = ["*"] + finding_region = AWS_REGION_US_EAST_1 + + assert AWSMutelist.is_item_matched(muted_regions, finding_region) + + def test_is_not_muted_in_region(self): + muted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1] + finding_region = "eu-west-2" + + assert not AWSMutelist.is_item_matched(muted_regions, finding_region) + + def test_is_muted_in_check(self): + muted_checks = { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + } + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-test", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "test-prowler", + "", + ) + + assert not ( + mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "check_test", + "us-east-2", + "test", + "", + ) + ) + + def test_is_muted_in_check_regex(self): + # Mutelist example + muted_checks = { + "s3_*": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + } + } + + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "s3_bucket_public_access", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "s3_bucket_no_mfa_delete", + AWS_REGION_US_EAST_1, + "prowler-test", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "s3_bucket_policy_public_write_access", + AWS_REGION_US_EAST_1, + "test-prowler", + "", + ) + + assert not ( + mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "iam_user_hardware_mfa_enabled", + AWS_REGION_US_EAST_1, + "test", + "", + ) + ) + + def test_is_muted_lambda_generic_check(self): + muted_checks = { + "lambda_*": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + } + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_invoke_api_operations_cloudtrail_logging_enabled", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_no_secrets_in_code", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_no_secrets_in_variables", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_not_publicly_accessible", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_url_cors_policy", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_url_public", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_using_supported_runtimes", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + def test_is_muted_lambda_concrete_check(self): + muted_checks = { + "lambda_function_no_secrets_in_variables": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + } + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_muted_in_check( + muted_checks, + AWS_ACCOUNT_NUMBER, + "awslambda_function_no_secrets_in_variables", + AWS_REGION_US_EAST_1, + "prowler", + "", + ) + + def test_is_muted_tags(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1], + "Resources": ["*"], + "Tags": ["environment=dev", "project=.*"], + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler", + "environment=dev", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_US_EAST_1, + "prowler-test", + "environment=dev | project=prowler", + ) + + assert not ( + mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + "us-east-2", + "test", + "environment=pro", + ) + ) + + def test_is_muted_specific_account_with_other_account_excepted(self): + # Mutelist + mutelist_content = { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "check_test": { + "Regions": [AWS_REGION_EU_WEST_1], + "Resources": ["*"], + "Tags": [], + "Exceptions": {"Accounts": ["111122223333"]}, + } + } + } + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "check_test", + AWS_REGION_EU_WEST_1, + "prowler", + "environment=dev", + ) + + assert not mutelist.is_muted( + "111122223333", + "check_test", + AWS_REGION_EU_WEST_1, + "prowler", + "environment=dev", + ) + + def test_is_muted_complex_mutelist(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "s3_bucket_object_versioning": { + "Regions": [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1], + "Resources": ["ci-logs", "logs", ".+-logs"], + }, + "ecs_task_definitions_no_environment_secrets": { + "Regions": ["*"], + "Resources": ["*"], + "Exceptions": { + "Accounts": [AWS_ACCOUNT_NUMBER], + "Regions": [ + AWS_REGION_EU_WEST_1, + AWS_REGION_EU_SOUTH_3, + ], + }, + }, + "*": { + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["environment=dev"], + }, + } + }, + AWS_ACCOUNT_NUMBER: { + "Checks": { + "*": { + "Regions": ["*"], + "Resources": ["*"], + "Exceptions": { + "Resources": ["test"], + "Tags": ["environment=prod"], + }, + } + } + }, + } + } + mutelist = AWSMutelist(mutelist_content=mutelist_content) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "test_check", + AWS_REGION_EU_WEST_1, + "prowler-logs", + "environment=dev", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "ecs_task_definitions_no_environment_secrets", + AWS_REGION_EU_WEST_1, + "prowler", + "environment=dev", + ) + + assert mutelist.is_muted( + AWS_ACCOUNT_NUMBER, + "s3_bucket_object_versioning", + AWS_REGION_EU_WEST_1, + "prowler-logs", + "environment=dev", + ) + + def test_is_muted_in_tags(self): + mutelist_tags = ["environment=dev", "project=prowler"] + + assert AWSMutelist.is_item_matched(mutelist_tags, "environment=dev") + + assert AWSMutelist.is_item_matched( + mutelist_tags, + "environment=dev | project=prowler", + ) + + assert not ( + AWSMutelist.is_item_matched( + mutelist_tags, + "environment=pro", + ) + ) + + def test_is_muted_in_tags_regex(self): + mutelist_tags = ["environment=(dev|test)", ".*=prowler"] + assert AWSMutelist.is_item_matched( + mutelist_tags, + "environment=test | proj=prowler", + ) + + assert AWSMutelist.is_item_matched( + mutelist_tags, + "env=prod | project=prowler", + ) + + assert not AWSMutelist.is_item_matched( + mutelist_tags, + "environment=prod | project=myproj", + ) + + def test_is_muted_in_tags_with_no_tags_in_finding(self): + mutelist_tags = ["environment=(dev|test)", ".*=prowler"] + finding_tags = "" + assert not AWSMutelist.is_item_matched(mutelist_tags, finding_tags) + + def test_is_excepted(self): + exceptions = { + "Accounts": [AWS_ACCOUNT_NUMBER], + "Regions": ["eu-central-1", "eu-south-3"], + "Resources": ["test"], + "Tags": ["environment=test", "project=.*"], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-central-1", + "test", + "environment=test", + ) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-south-3", + "test", + "environment=test", + ) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-south-3", + "test123", + "environment=test", + ) + + def test_is_excepted_only_in_account(self): + exceptions = { + "Accounts": [AWS_ACCOUNT_NUMBER], + "Regions": [], + "Resources": [], + "Tags": [], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-central-1", + "test", + "environment=test", + ) + + def test_is_excepted_only_in_region(self): + exceptions = { + "Accounts": [], + "Regions": [AWS_REGION_EU_CENTRAL_1, AWS_REGION_EU_SOUTH_3], + "Resources": [], + "Tags": [], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + AWS_REGION_EU_CENTRAL_1, + "test", + "environment=test", + ) + + def test_is_excepted_only_in_resources(self): + exceptions = { + "Accounts": [], + "Regions": [], + "Resources": ["resource_1"], + "Tags": [], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + AWS_REGION_EU_CENTRAL_1, + "resource_1", + "environment=test", + ) + + def test_is_excepted_only_in_tags(self): + exceptions = { + "Accounts": [], + "Regions": [], + "Resources": [], + "Tags": ["environment=test"], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + AWS_REGION_EU_CENTRAL_1, + "resource_1", + "environment=test", + ) + + def test_is_excepted_in_account_and_tags(self): + exceptions = { + "Accounts": [AWS_ACCOUNT_NUMBER], + "Regions": [], + "Resources": [], + "Tags": ["environment=test"], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + AWS_REGION_EU_CENTRAL_1, + "resource_1", + "environment=test", + ) + + assert not mutelist.is_excepted( + exceptions, + "111122223333", + AWS_REGION_EU_CENTRAL_1, + "resource_1", + "environment=test", + ) + + assert not mutelist.is_excepted( + exceptions, + "111122223333", + AWS_REGION_EU_CENTRAL_1, + "resource_1", + "environment=dev", + ) + + def test_is_excepted_all_wildcard(self): + exceptions = { + "Accounts": ["*"], + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["*"], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert mutelist.is_excepted( + exceptions, AWS_ACCOUNT_NUMBER, "eu-south-2", "test", "environment=test" + ) + assert not mutelist.is_excepted( + exceptions, AWS_ACCOUNT_NUMBER, "eu-south-2", "test", None + ) + + def test_is_not_excepted(self): + exceptions = { + "Accounts": [AWS_ACCOUNT_NUMBER], + "Regions": ["eu-central-1", "eu-south-3"], + "Resources": ["test"], + "Tags": ["environment=test", "project=.*"], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert not mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-south-2", + "test", + "environment=test", + ) + + assert not mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-south-3", + "prowler", + "environment=test", + ) + + assert not mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-south-3", + "test", + "environment=pro", + ) + + def test_is_excepted_all_empty(self): + exceptions = { + "Accounts": [], + "Regions": [], + "Resources": [], + "Tags": [], + } + mutelist = AWSMutelist(mutelist_content={}) + + assert not mutelist.is_excepted( + exceptions, + AWS_ACCOUNT_NUMBER, + "eu-south-2", + "test", + "environment=test", + ) + + def test_is_muted_in_resource(self): + mutelist_resources = ["prowler", "^test", "prowler-pro"] + + assert AWSMutelist.is_item_matched(mutelist_resources, "prowler") + assert AWSMutelist.is_item_matched(mutelist_resources, "prowler-test") + assert AWSMutelist.is_item_matched(mutelist_resources, "test-prowler") + assert not AWSMutelist.is_item_matched(mutelist_resources, "random") + + def test_is_muted_in_resource_starting_by_star(self): + allowlist_resources = ["*.es"] + + assert AWSMutelist.is_item_matched(allowlist_resources, "google.es") diff --git a/tests/lib/mutelist/fixtures/aws_mutelist.yaml b/tests/providers/aws/lib/mutelist/fixtures/aws_mutelist.yaml similarity index 100% rename from tests/lib/mutelist/fixtures/aws_mutelist.yaml rename to tests/providers/aws/lib/mutelist/fixtures/aws_mutelist.yaml diff --git a/tests/providers/azure/lib/mutelist/azure_mutelist_test.py b/tests/providers/azure/lib/mutelist/azure_mutelist_test.py new file mode 100644 index 0000000000..d946443732 --- /dev/null +++ b/tests/providers/azure/lib/mutelist/azure_mutelist_test.py @@ -0,0 +1,68 @@ +import yaml +from mock import MagicMock + +from prowler.providers.azure.lib.mutelist.mutelist import AzureMutelist + +MUTELIST_FIXTURE_PATH = ( + "tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml" +) + + +class TestAzureMutelist: + def test_get_mutelist_file_from_local_file(self): + mutelist = AzureMutelist(mutelist_path=MUTELIST_FIXTURE_PATH) + + with open(MUTELIST_FIXTURE_PATH) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + assert mutelist.mutelist == mutelist_fixture + assert mutelist.mutelist_file_path == MUTELIST_FIXTURE_PATH + + def test_get_mutelist_file_from_local_file_non_existent(self): + mutelist_path = "tests/lib/mutelist/fixtures/not_present" + mutelist = AzureMutelist(mutelist_path=mutelist_path) + + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path == mutelist_path + + def test_validate_mutelist_not_valid_key(self): + mutelist_path = MUTELIST_FIXTURE_PATH + with open(mutelist_path) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + mutelist_fixture["Accounts1"] = mutelist_fixture["Accounts"] + del mutelist_fixture["Accounts"] + + mutelist = AzureMutelist(mutelist_content=mutelist_fixture) + + assert not mutelist.validate_mutelist() + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path is None + + def test_is_finding_muted(self): + # Mutelist + mutelist_content = { + "Accounts": { + "subscription_1": { + "Checks": { + "check_test": { + "Regions": ["*"], + "Resources": ["test_resource"], + } + } + } + } + } + + mutelist = AzureMutelist(mutelist_content=mutelist_content) + + finding = MagicMock + finding.check_metadata = MagicMock + finding.check_metadata.CheckID = "check_test" + finding.location = "West Europe" + finding.status = "FAIL" + finding.resource_name = "test_resource" + finding.resource_tags = [] + finding.subscription = "subscription_1" + + assert mutelist.is_finding_muted(finding) diff --git a/tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml b/tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml new file mode 100644 index 0000000000..69fb405006 --- /dev/null +++ b/tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml @@ -0,0 +1,16 @@ +### Account, Check and/or Region can be * to apply for all the cases. +### Resources and tags are lists that can have either Regex or Keywords. +### Tags is an optional list that matches on tuples of 'key=value' and are "ANDed" together. +### Use an alternation Regex to match one of multiple tags with "ORed" logic. +### For each check you can except Accounts, Regions, Resources and/or Tags. +########################### MUTELIST EXAMPLE ########################### +Mutelist: + Accounts: + "subscription_1": + Checks: + "aks_cluster_rbac_enabled": + Regions: + - "*" + Resources: + - "resource_1" + - "resource_2" diff --git a/tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml b/tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml new file mode 100644 index 0000000000..9d530a7a81 --- /dev/null +++ b/tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml @@ -0,0 +1,16 @@ +### Account, Check and/or Region can be * to apply for all the cases. +### Resources and tags are lists that can have either Regex or Keywords. +### Tags is an optional list that matches on tuples of 'key=value' and are "ANDed" together. +### Use an alternation Regex to match one of multiple tags with "ORed" logic. +### For each check you can except Accounts, Regions, Resources and/or Tags. +########################### MUTELIST EXAMPLE ########################### +Mutelist: + Accounts: + "project_1": + Checks: + "apikeys_api_restrictions_configured": + Regions: + - "*" + Resources: + - "resource_1" + - "resource_2" diff --git a/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py b/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py new file mode 100644 index 0000000000..831ff154c1 --- /dev/null +++ b/tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py @@ -0,0 +1,66 @@ +import yaml +from mock import MagicMock + +from prowler.providers.gcp.lib.mutelist.mutelist import GCPMutelist + +MUTELIST_FIXTURE_PATH = "tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml" + + +class TestGCPMutelist: + def test_get_mutelist_file_from_local_file(self): + mutelist = GCPMutelist(mutelist_path=MUTELIST_FIXTURE_PATH) + + with open(MUTELIST_FIXTURE_PATH) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + assert mutelist.mutelist == mutelist_fixture + assert mutelist.mutelist_file_path == MUTELIST_FIXTURE_PATH + + def test_get_mutelist_file_from_local_file_non_existent(self): + mutelist_path = "tests/lib/mutelist/fixtures/not_present" + mutelist = GCPMutelist(mutelist_path=mutelist_path) + + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path == mutelist_path + + def test_validate_mutelist_not_valid_key(self): + mutelist_path = MUTELIST_FIXTURE_PATH + with open(mutelist_path) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + mutelist_fixture["Accounts1"] = mutelist_fixture["Accounts"] + del mutelist_fixture["Accounts"] + + mutelist = GCPMutelist(mutelist_content=mutelist_fixture) + + assert not mutelist.validate_mutelist() + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path is None + + def test_is_finding_muted(self): + # Mutelist + mutelist_content = { + "Accounts": { + "project_1": { + "Checks": { + "check_test": { + "Regions": ["*"], + "Resources": ["test_resource"], + } + } + } + } + } + + mutelist = GCPMutelist(mutelist_content=mutelist_content) + + finding = MagicMock + finding.check_metadata = MagicMock + finding.check_metadata.CheckID = "check_test" + finding.status = "FAIL" + finding.resource_name = "test_resource" + finding.location = "test-location" + finding.resource_tags = [] + finding.project_id = "project_1" + + assert mutelist.is_finding_muted(finding) diff --git a/tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml b/tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml new file mode 100644 index 0000000000..5a5deb9b02 --- /dev/null +++ b/tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml @@ -0,0 +1,16 @@ +### Account, Check and/or Region can be * to apply for all the cases. +### Resources and tags are lists that can have either Regex or Keywords. +### Tags is an optional list that matches on tuples of 'key=value' and are "ANDed" together. +### Use an alternation Regex to match one of multiple tags with "ORed" logic. +### For each check you can except Accounts, Regions, Resources and/or Tags. +########################### MUTELIST EXAMPLE ########################### +Mutelist: + Accounts: + "cluster_1": + Checks: + "controllermanager_bind_address": + Regions: + - "*" + Resources: + - "resource_1" + - "resource_2" diff --git a/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py b/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py new file mode 100644 index 0000000000..e60516e2ce --- /dev/null +++ b/tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py @@ -0,0 +1,99 @@ +import yaml +from mock import MagicMock + +from prowler.providers.kubernetes.lib.mutelist.mutelist import KubernetesMutelist + +MUTELIST_FIXTURE_PATH = ( + "tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml" +) + + +class TestKubernetesMutelist: + def test_get_mutelist_file_from_local_file(self): + mutelist = KubernetesMutelist(mutelist_path=MUTELIST_FIXTURE_PATH) + + with open(MUTELIST_FIXTURE_PATH) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + assert mutelist.mutelist == mutelist_fixture + assert mutelist.mutelist_file_path == MUTELIST_FIXTURE_PATH + + def test_get_mutelist_file_from_local_file_non_existent(self): + mutelist_path = "tests/lib/mutelist/fixtures/not_present" + mutelist = KubernetesMutelist(mutelist_path=mutelist_path) + + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path == mutelist_path + + def test_validate_mutelist_not_valid_key(self): + mutelist_path = MUTELIST_FIXTURE_PATH + with open(mutelist_path) as f: + mutelist_fixture = yaml.safe_load(f)["Mutelist"] + + mutelist_fixture["Accounts1"] = mutelist_fixture["Accounts"] + del mutelist_fixture["Accounts"] + + mutelist = KubernetesMutelist(mutelist_content=mutelist_fixture) + + assert not mutelist.validate_mutelist() + assert mutelist.mutelist == {} + assert mutelist.mutelist_file_path is None + + def test_is_finding_muted(self): + # Mutelist + mutelist_content = { + "Accounts": { + "cluster_1": { + "Checks": { + "check_test": { + # TODO: review this with Sergio + "Regions": ["*"], + "Resources": ["test_resource"], + } + } + } + } + } + + mutelist = KubernetesMutelist(mutelist_content=mutelist_content) + + finding = MagicMock + finding.check_metadata = MagicMock + finding.check_metadata.CheckID = "check_test" + finding.status = "FAIL" + finding.resource_name = "test_resource" + finding.namespace = "test-location" + finding.resource_tags = [] + + assert mutelist.is_finding_muted(finding, "cluster_1") + + def test_is_finding_muted_etcd_star_within_check_name(self): + # Mutelist + mutelist_content = { + "Accounts": { + "*": { + "Checks": { + "etcd_*": { + "Regions": ["*"], + "Resources": ["*"], + "Exceptions": { + "Accounts": ["k8s-cluster-2"], + "Regions": ["namespace1", "namespace2"], + }, + } + } + } + } + } + + mutelist = KubernetesMutelist(mutelist_content=mutelist_content) + + finding = MagicMock + finding.check_metadata = MagicMock + finding.check_metadata.CheckID = "apiserver_etcd_cafile_set" + finding.status = "FAIL" + finding.resource_name = "test_resource" + finding.namespace = "namespace1" + finding.resource_tags = [] + + assert not mutelist.is_finding_muted(finding, "cluster_1")