diff --git a/prowler/lib/mutelist/models.py b/prowler/lib/mutelist/models.py new file mode 100644 index 0000000000..b77e3e04ba --- /dev/null +++ b/prowler/lib/mutelist/models.py @@ -0,0 +1,23 @@ +from schema import Optional, Schema + +mutelist_schema = Schema( + { + "Accounts": { + str: { + "Checks": { + str: { + "Regions": list, + "Resources": list, + Optional("Tags"): list, + Optional("Exceptions"): { + Optional("Accounts"): list, + Optional("Regions"): list, + Optional("Resources"): list, + Optional("Tags"): list, + }, + } + } + } + } + } +) diff --git a/prowler/lib/mutelist/mutelist.py b/prowler/lib/mutelist/mutelist.py index 23d3297af0..e61b6ee48e 100644 --- a/prowler/lib/mutelist/mutelist.py +++ b/prowler/lib/mutelist/mutelist.py @@ -1,115 +1,34 @@ import re -import sys from typing import Any import yaml -from boto3 import Session -from boto3.dynamodb.conditions import Attr -from schema import Optional, Schema from prowler.lib.logger import logger +from prowler.lib.mutelist.models import mutelist_schema from prowler.lib.outputs.utils import unroll_tags -mutelist_schema = Schema( - { - "Accounts": { - str: { - "Checks": { - str: { - "Regions": list, - "Resources": list, - Optional("Tags"): list, - Optional("Exceptions"): { - Optional("Accounts"): list, - Optional("Regions"): list, - Optional("Resources"): list, - Optional("Tags"): list, - }, - } - } - } - } - } -) - -def parse_mutelist_file( - mutelist_path: str, aws_session: Session = None, aws_account: str = None -): +def get_mutelist_file_from_local_file(mutelist_path: str): try: - # Check if file is a S3 URI - if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_path): - 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"] - # Check if file is a Lambda Function ARN - elif re.search(r"^arn:(\w+):lambda:", mutelist_path): - 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"] - # Check if file is a 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 = {"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, "*"]), - ) - 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"] - else: - with open(mutelist_path) as f: - mutelist = yaml.safe_load(f)["Mutelist"] - try: - mutelist_schema.validate(mutelist) - except Exception as error: - logger.critical( - f"{error.__class__.__name__} -- Mutelist YAML is malformed - {error}[{error.__traceback__.tb_lineno}]" - ) - sys.exit(1) - return mutelist + with open(mutelist_path) as f: + mutelist = yaml.safe_load(f)["Mutelist"] + return mutelist except Exception as error: - logger.critical( + logger.error( f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]" ) - sys.exit(1) + return {} + + +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 mutelist_findings( diff --git a/prowler/providers/aws/aws_provider.py b/prowler/providers/aws/aws_provider.py index 7cfb3b62ad..e84873fd4a 100644 --- a/prowler/providers/aws/aws_provider.py +++ b/prowler/providers/aws/aws_provider.py @@ -1,5 +1,6 @@ import os import pathlib +import re import sys from argparse import Namespace from datetime import datetime @@ -15,11 +16,16 @@ from tzlocal import get_localzone from prowler.config.config import ( aws_services_json_file, + get_default_mute_file_path, load_and_validate_config_file, load_and_validate_fixer_config_file, ) 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_STS_GLOBAL_ENDPOINT_REGION, @@ -28,6 +34,11 @@ 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.organizations.organizations import ( get_organizations_metadata, parse_organizations_metadata, @@ -285,6 +296,51 @@ class AwsProvider(Provider): arguments, bulk_checks_metadata, self._identity ) + @property + def mutelist(self): + """ + 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) + 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 + @property def get_output_mapping(self): return { diff --git a/prowler/providers/aws/lib/mutelist/__init__.py b/prowler/providers/aws/lib/mutelist/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/lib/mutelist/mutelist.py b/prowler/providers/aws/lib/mutelist/mutelist.py new file mode 100644 index 0000000000..0750843f02 --- /dev/null +++ b/prowler/providers/aws/lib/mutelist/mutelist.py @@ -0,0 +1,84 @@ +import yaml +from boto3 import Session +from boto3.dynamodb.conditions import Attr + +from prowler.lib.logger import logger + + +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}]" + ) + 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, "*"]), + ) + 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 {} diff --git a/prowler/providers/common/provider.py b/prowler/providers/common/provider.py index 577abb73ff..11f64f7c9f 100644 --- a/prowler/providers/common/provider.py +++ b/prowler/providers/common/provider.py @@ -7,7 +7,10 @@ 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 parse_mutelist_file +from prowler.lib.mutelist.mutelist import ( + get_mutelist_file_from_local_file, + validate_mutelist, +) providers_path = "prowler.providers" @@ -178,7 +181,8 @@ class Provider(ABC): if not mutelist_path: mutelist_path = get_default_mute_file_path(self.type) if mutelist_path: - mutelist = parse_mutelist_file(mutelist_path) + mutelist = get_mutelist_file_from_local_file(mutelist_path) + mutelist = validate_mutelist(mutelist) else: mutelist = {} diff --git a/tests/lib/mutelist/mutelist_test.py b/tests/lib/mutelist/mutelist_test.py index c55886d55a..dcb78aed61 100644 --- a/tests/lib/mutelist/mutelist_test.py +++ b/tests/lib/mutelist/mutelist_test.py @@ -1,9 +1,8 @@ import yaml -from boto3 import resource from mock import MagicMock -from moto import mock_aws from prowler.lib.mutelist.mutelist import ( + get_mutelist_file_from_local_file, is_excepted, is_muted, is_muted_in_check, @@ -11,7 +10,7 @@ from prowler.lib.mutelist.mutelist import ( is_muted_in_resource, is_muted_in_tags, mutelist_findings, - parse_mutelist_file, + validate_mutelist, ) from tests.providers.aws.utils import ( AWS_ACCOUNT_NUMBER, @@ -24,118 +23,33 @@ from tests.providers.aws.utils import ( class TestMutelist: - # Test S3 mutelist - @mock_aws - def test_s3_mutelist(self): - aws_provider = set_mocked_aws_provider() - # Create bucket and upload mutelist yaml - s3_resource = resource("s3", region_name=AWS_REGION_US_EAST_1) - s3_resource.create_bucket(Bucket="test-mutelist") - s3_resource.Object("test-mutelist", "mutelist.yaml").put( - Body=open( - "tests//lib/mutelist/fixtures/aws_mutelist.yaml", - "rb", - ) - ) + 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"] - with open("tests//lib/mutelist/fixtures/aws_mutelist.yaml") as f: - assert yaml.safe_load(f)["Mutelist"] == parse_mutelist_file( - "s3://test-mutelist/mutelist.yaml", - aws_provider.session.current_session, - aws_provider.identity.account, - ) + assert get_mutelist_file_from_local_file(mutelist_path) == mutelist_fixture - # Test DynamoDB mutelist - @mock_aws - def test_dynamo_mutelist(self): - aws_provider = set_mocked_aws_provider() - # Create table and put item - dynamodb_resource = resource("dynamodb", region_name=AWS_REGION_US_EAST_1) - table_name = "test-mutelist" - params = { - "TableName": table_name, - "KeySchema": [ - {"AttributeName": "Accounts", "KeyType": "HASH"}, - {"AttributeName": "Checks", "KeyType": "RANGE"}, - ], - "AttributeDefinitions": [ - {"AttributeName": "Accounts", "AttributeType": "S"}, - {"AttributeName": "Checks", "AttributeType": "S"}, - ], - "ProvisionedThroughput": { - "ReadCapacityUnits": 10, - "WriteCapacityUnits": 10, - }, - } - table = dynamodb_resource.create_table(**params) - table.put_item( - Item={ - "Accounts": "*", - "Checks": "iam_user_hardware_mfa_enabled", - "Regions": [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1], - "Resources": ["keyword"], - } - ) + def test_get_mutelist_file_from_local_file_non_existent(self): + mutelist_path = "tests/lib/mutelist/fixtures/not_present" - assert ( - "keyword" - in parse_mutelist_file( - "arn:aws:dynamodb:" - + AWS_REGION_US_EAST_1 - + ":" - + str(AWS_ACCOUNT_NUMBER) - + ":table/" - + table_name, - aws_provider.session.current_session, - aws_provider.identity.account, - )["Accounts"]["*"]["Checks"]["iam_user_hardware_mfa_enabled"]["Resources"] - ) + assert get_mutelist_file_from_local_file(mutelist_path) == {} - @mock_aws - def test_dynamo_mutelist_with_tags(self): - aws_provider = set_mocked_aws_provider() - # Create table and put item - dynamodb_resource = resource("dynamodb", region_name=AWS_REGION_US_EAST_1) - table_name = "test-mutelist" - params = { - "TableName": table_name, - "KeySchema": [ - {"AttributeName": "Accounts", "KeyType": "HASH"}, - {"AttributeName": "Checks", "KeyType": "RANGE"}, - ], - "AttributeDefinitions": [ - {"AttributeName": "Accounts", "AttributeType": "S"}, - {"AttributeName": "Checks", "AttributeType": "S"}, - ], - "ProvisionedThroughput": { - "ReadCapacityUnits": 10, - "WriteCapacityUnits": 10, - }, - } - table = dynamodb_resource.create_table(**params) - table.put_item( - Item={ - "Accounts": "*", - "Checks": "*", - "Regions": ["*"], - "Resources": ["*"], - "Tags": ["environment=dev"], - } - ) + 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 ( - "environment=dev" - in parse_mutelist_file( - "arn:aws:dynamodb:" - + AWS_REGION_US_EAST_1 - + ":" - + str(AWS_ACCOUNT_NUMBER) - + ":table/" - + table_name, - aws_provider.session.current_session, - aws_provider.identity.account, - )["Accounts"]["*"]["Checks"]["*"]["Tags"] - ) + 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): diff --git a/tests/providers/aws/aws_provider_test.py b/tests/providers/aws/aws_provider_test.py index 0c44c3642a..2b1d7a4f12 100644 --- a/tests/providers/aws/aws_provider_test.py +++ b/tests/providers/aws/aws_provider_test.py @@ -9,7 +9,7 @@ from os import rmdir from re import search import botocore -from boto3 import client, session +from boto3 import client, resource, session from freezegun import freeze_time from mock import patch from moto import mock_aws @@ -56,7 +56,6 @@ from tests.providers.aws.utils import ( set_mocked_aws_provider, ) -# Mocking GetCallerIdentity for China and GovCloud make_api_call = botocore.client.BaseClient._make_api_call @@ -528,38 +527,39 @@ aws: @mock_aws def test_aws_provider_mutelist(self): mutelist = { - "Accounts": { - AWS_ACCOUNT_NUMBER: { - "Checks": { - "test-check": { - "Regions": [], - "Resources": [], - "Tags": [], - "Exceptions": { - "Accounts": [], + "Mutelist": { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "test-check": { "Regions": [], "Resources": [], "Tags": [], - }, + "Exceptions": { + "Accounts": [], + "Regions": [], + "Resources": [], + "Tags": [], + }, + } } } } } } - mutelist_content = {"Mutelist": mutelist} - config_file = tempfile.NamedTemporaryFile(delete=False) - with open(config_file.name, "w") as allowlist_file: - allowlist_file.write(json.dumps(mutelist_content, indent=4)) + mutelist_file = tempfile.NamedTemporaryFile(delete=False) + with open(mutelist_file.name, "w") as mutelist_file: + mutelist_file.write(json.dumps(mutelist, indent=4)) arguments = Namespace() aws_provider = AwsProvider(arguments) - aws_provider.mutelist = config_file.name + aws_provider.mutelist = mutelist_file.name - os.remove(config_file.name) + os.remove(mutelist_file.name) - assert aws_provider.mutelist == mutelist + assert aws_provider.mutelist == mutelist["Mutelist"] @mock_aws def test_aws_provider_mutelist_none(self): @@ -567,13 +567,135 @@ aws: aws_provider = AwsProvider(arguments) with patch( - "prowler.providers.common.provider.get_default_mute_file_path", + "prowler.providers.aws.aws_provider.get_default_mute_file_path", return_value=None, ): aws_provider.mutelist = None assert aws_provider.mutelist == {} + @mock_aws + def test_aws_provider_mutelist_s3(self): + # Create mutelist temp file + mutelist = { + "Mutelist": { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "test-check": { + "Regions": [], + "Resources": [], + "Tags": [], + "Exceptions": { + "Accounts": [], + "Regions": [], + "Resources": [], + "Tags": [], + }, + } + } + } + } + } + } + + mutelist_file = tempfile.NamedTemporaryFile(delete=False) + with open(mutelist_file.name, "w") as mutelist_file: + mutelist_file.write(json.dumps(mutelist, indent=4)) + + # Create bucket and upload mutelist yaml + s3_resource = resource("s3", region_name=AWS_REGION_US_EAST_1) + bucket_name = "test-mutelist" + mutelist_file_name = "mutelist.yaml" + mutelist_bucket_object_uri = f"s3://{bucket_name}/{mutelist_file_name}" + s3_resource.create_bucket(Bucket=bucket_name) + s3_resource.Object(bucket_name, "mutelist.yaml").put( + Body=open( + mutelist_file.name, + "rb", + ) + ) + + arguments = Namespace() + aws_provider = AwsProvider(arguments) + + aws_provider.mutelist = mutelist_bucket_object_uri + os.remove(mutelist_file.name) + + assert aws_provider.mutelist == mutelist["Mutelist"] + + @mock_aws + def test_aws_provider_mutelist_lambda(self): + # Create mutelist temp file + mutelist = { + "Mutelist": { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "test-check": { + "Regions": [], + "Resources": [], + "Tags": [], + "Exceptions": { + "Accounts": [], + "Regions": [], + "Resources": [], + "Tags": [], + }, + } + } + } + } + } + } + + arguments = Namespace() + aws_provider = AwsProvider(arguments) + + with patch( + "prowler.providers.aws.aws_provider.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" + + assert aws_provider.mutelist == mutelist["Mutelist"] + + @mock_aws + def test_aws_provider_mutelist_dynamodb(self): + # Create mutelist temp file + mutelist = { + "Mutelist": { + "Accounts": { + AWS_ACCOUNT_NUMBER: { + "Checks": { + "test-check": { + "Regions": [], + "Resources": [], + "Tags": [], + "Exceptions": { + "Accounts": [], + "Regions": [], + "Resources": [], + "Tags": [], + }, + } + } + } + } + } + } + + arguments = Namespace() + aws_provider = AwsProvider(arguments) + + with patch( + "prowler.providers.aws.aws_provider.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" + + assert aws_provider.mutelist == mutelist["Mutelist"] + @mock_aws def test_generate_regional_clients_all_enabled_regions(self): arguments = Namespace() diff --git a/tests/providers/aws/lib/mutelist/aws_mutelist_test.py b/tests/providers/aws/lib/mutelist/aws_mutelist_test.py new file mode 100644 index 0000000000..919436e721 --- /dev/null +++ b/tests/providers/aws/lib/mutelist/aws_mutelist_test.py @@ -0,0 +1,282 @@ +import io +from json import dumps + +import botocore +import yaml +from boto3 import client, resource +from mock import patch +from moto import mock_aws + +from prowler.providers.aws.lib.mutelist.mutelist import ( + get_mutelist_file_from_dynamodb, + get_mutelist_file_from_lambda, + get_mutelist_file_from_s3, +) +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_WEST_1, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +make_api_call = botocore.client.BaseClient._make_api_call + + +def mock_make_api_call(self, operation_name, kwarg): + if operation_name == "Invoke": + return { + "Payload": io.BytesIO( + dumps( + { + "Mutelist": { + "Accounts": { + "*": { + "Checks": { + "*": { + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["key:value"], + }, + } + }, + } + } + } + ).encode("utf-8") + ) + } + + return make_api_call(self, operation_name, kwarg) + + +class TestMutelistAWS: + @mock_aws + def test_get_mutelist_file_from_s3(self): + aws_provider = set_mocked_aws_provider() + # Create bucket and upload mutelist yaml + s3_resource = resource("s3", region_name=AWS_REGION_US_EAST_1) + s3_resource.create_bucket(Bucket="test-mutelist") + s3_resource.Object("test-mutelist", "mutelist.yaml").put( + Body=open( + "tests/lib/mutelist/fixtures/aws_mutelist.yaml", + "rb", + ) + ) + + with open("tests/lib/mutelist/fixtures/aws_mutelist.yaml") 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 + ) + + @mock_aws + def test_get_mutelist_file_from_s3_not_present(self): + aws_provider = set_mocked_aws_provider() + + assert ( + get_mutelist_file_from_s3( + "s3://test-mutelist/mutelist.yaml", + aws_provider.session.current_session, + ) + == {} + ) + + @mock_aws + def test_get_mutelist_file_from_dynamodb(self): + aws_provider = set_mocked_aws_provider() + # Create table and put item + dynamodb_resource = resource("dynamodb", region_name=AWS_REGION_US_EAST_1) + table_name = "test-mutelist" + table_arn = f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/{table_name}" + params = { + "TableName": table_name, + "KeySchema": [ + {"AttributeName": "Accounts", "KeyType": "HASH"}, + {"AttributeName": "Checks", "KeyType": "RANGE"}, + ], + "AttributeDefinitions": [ + {"AttributeName": "Accounts", "AttributeType": "S"}, + {"AttributeName": "Checks", "AttributeType": "S"}, + ], + "ProvisionedThroughput": { + "ReadCapacityUnits": 10, + "WriteCapacityUnits": 10, + }, + } + table = dynamodb_resource.create_table(**params) + dynamo_db_mutelist = { + "Accounts": "*", + "Checks": "iam_user_hardware_mfa_enabled", + "Regions": [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1], + "Resources": ["keyword"], + "Exceptions": {}, + } + mutelist = { + "Accounts": { + "*": { + "Checks": { + "iam_user_hardware_mfa_enabled": { + "Regions": [AWS_REGION_EU_WEST_1, AWS_REGION_US_EAST_1], + "Resources": ["keyword"], + "Exceptions": {}, + }, + } + }, + } + } + 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 + ) + + @mock_aws + def test_get_mutelist_file_from_dynamodb_with_tags(self): + aws_provider = set_mocked_aws_provider() + # Create table and put item + dynamodb_resource = resource("dynamodb", region_name=AWS_REGION_US_EAST_1) + table_name = "test-mutelist" + table_arn = f"arn:aws:dynamodb:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:table/{table_name}" + params = { + "TableName": table_name, + "KeySchema": [ + {"AttributeName": "Accounts", "KeyType": "HASH"}, + {"AttributeName": "Checks", "KeyType": "RANGE"}, + ], + "AttributeDefinitions": [ + {"AttributeName": "Accounts", "AttributeType": "S"}, + {"AttributeName": "Checks", "AttributeType": "S"}, + ], + "ProvisionedThroughput": { + "ReadCapacityUnits": 10, + "WriteCapacityUnits": 10, + }, + } + table = dynamodb_resource.create_table(**params) + dynamo_db_mutelist = { + "Accounts": "*", + "Checks": "*", + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["environment=dev"], + } + mutelist = { + "Accounts": { + "*": { + "Checks": { + "*": { + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["environment=dev"], + }, + } + }, + } + } + 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 + ) + + @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, + ) + == {} + ) + + @mock_aws(config={"lambda": {"use_docker": False}}) + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + def test_get_mutelist_file_from_lambda(self): + aws_provider = set_mocked_aws_provider() + lambda_name = "mutelist" + lambda_role = "lambda_role" + lambda_client = client("lambda", region_name=AWS_REGION_US_EAST_1) + iam_client = client("iam", region_name=AWS_REGION_US_EAST_1) + lambda_role_assume_policy = { + "Version": "2012-10-17", + "Statement": { + "Sid": "test", + "Effect": "Allow", + "Principal": {"AWS": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"}, + "Action": "sts:AssumeRole", + }, + } + lambda_role_arn = iam_client.create_role( + RoleName=lambda_role, + AssumeRolePolicyDocument=dumps(lambda_role_assume_policy), + )["Role"]["Arn"] + lambda_code = """def handler(event, context): + checks = {} + checks["*"] = { "Regions": [ "*" ], "Resources": [ "" ], Optional("Tags"): [ "key:value" ] } + + al = { "Mutelist": { "Accounts": { "*": { "Checks": checks } } } } + return al""" + + lambda_function = lambda_client.create_function( + FunctionName=lambda_name, + Runtime="3.9", + Role=lambda_role_arn, + Handler="lambda_function.lambda_handler", + Code={"ZipFile": create_zip_file(code=lambda_code).read()}, + Description="test lambda function", + ) + lambda_function_arn = lambda_function["FunctionArn"] + mutelist = { + "Accounts": { + "*": { + "Checks": { + "*": { + "Regions": ["*"], + "Resources": ["*"], + "Tags": ["key:value"], + }, + } + }, + } + } + + assert ( + get_mutelist_file_from_lambda( + lambda_function_arn, aws_provider.session.current_session + ) + == mutelist + ) + + @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 + ) + == {} + )