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

chore(mutelist): create new class to encapsulate the logic (#4413)

Browse Source
This commit is contained in:
Pepe Fagoaga
2024-07-16 19:44:43 +02:00
committed by GitHub
parent d01cc51b6d
commit 577afbd521
24 changed files with 2084 additions and 1810 deletions
Download Patch File Download Diff File Expand all files Collapse all files
prowler/lib/check/check.py
+17 -8
View File
@@ -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
prowler/lib/mutelist/mutelist.py
+256 -337
View File
@@ -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
prowler/providers/aws/aws_provider.py
+9 -39
View File
@@ -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):
prowler/providers/aws/lib/mutelist/mutelist.py
+125 -72
View File
@@ -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 {}
prowler/providers/azure/azure_provider.py
+24 -2
View File
@@ -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 {
prowler/providers/azure/lib/mutelist/__init__.py
View File
prowler/providers/azure/lib/mutelist/mutelist.py
+18
View File
@@ -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),
)
prowler/providers/common/provider.py
+2 -38
View File
@@ -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
prowler/providers/gcp/gcp_provider.py
+24 -2
View File
@@ -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 {
prowler/providers/gcp/lib/mutelist/__init__.py
View File
prowler/providers/gcp/lib/mutelist/mutelist.py
+18
View File
@@ -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),
)
prowler/providers/kubernetes/kubernetes_provider.py
+23 -1
View File
@@ -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 {
prowler/providers/kubernetes/lib/mutelist/__init__.py
View File
prowler/providers/kubernetes/lib/mutelist/mutelist.py
+19
View File
@@ -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),
)
tests/lib/mutelist/mutelist_test.py
-1244
View File
File diff suppressed because it is too large Load Diff
tests/providers/aws/aws_provider_test.py
+22 -11
View File
@@ -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):
tests/providers/aws/lib/mutelist/aws_mutelist_test.py
+1246 -56
View File
File diff suppressed because it is too large Load Diff
tests/lib/mutelist/fixtures/aws_mutelist.yaml → tests/providers/aws/lib/mutelist/fixtures/aws_mutelist.yaml
View File
tests/providers/azure/lib/mutelist/azure_mutelist_test.py
+68
View File
@@ -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)
tests/providers/azure/lib/mutelist/fixtures/azure_mutelist.yaml
+16
View File
@@ -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"
tests/providers/gcp/lib/mutelist/fixtures/gcp_mutelist.yaml
+16
View File
@@ -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"
tests/providers/gcp/lib/mutelist/gcp_mutelist_test.py
+66
View File
@@ -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)
tests/providers/kubernetes/lib/mutelist/fixtures/kubernetes_mutelist.yaml
+16
View File
@@ -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"
tests/providers/kubernetes/lib/mutelist/kubernetes_mutelist_test.py
+99
View File
@@ -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")