mirror of
https://github.com/prowler-cloud/prowler.git
synced 2026-07-23 12:31:54 +00:00
fix(mutelist): Split code for AWS and the rest of providers (#4143)
This commit is contained in:
Pepe Fagoaga
@@ -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,
|
||||
},
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
@@ -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(
|
||||
|
||||
@@ -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 {
|
||||
|
||||
@@ -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 {}
|
||||
@@ -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 = {}
|
||||
|
||||
|
||||
@@ -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):
|
||||
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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
|
||||
)
|
||||
== {}
|
||||
)
|
||||
Reference in New Issue
Block a user