feat(mute list): change allowlist to mute list (#3039)

Co-authored-by: Nacho Rivera <nachor1992@gmail.com>
This commit is contained in:
Sergio Garcia
2023-12-12 16:57:52 +01:00
committed by Sergio Garcia
parent c3d129a4b2
commit 445821eceb
36 changed files with 474 additions and 449 deletions
+6 -6
View File
@@ -29,10 +29,10 @@ The following list includes all the AWS checks with configurable variables that
| `organizations_delegated_administrators` | `organizations_trusted_delegated_administrators` | List of Strings |
| `ecr_repositories_scan_vulnerabilities_in_latest_image` | `ecr_repository_vulnerability_minimum_severity` | String |
| `trustedadvisor_premium_support_plan_subscribed` | `verify_premium_support_plans` | Boolean |
| `config_recorder_all_regions_enabled` | `allowlist_non_default_regions` | Boolean |
| `drs_job_exist` | `allowlist_non_default_regions` | Boolean |
| `guardduty_is_enabled` | `allowlist_non_default_regions` | Boolean |
| `securityhub_enabled` | `allowlist_non_default_regions` | Boolean |
| `config_recorder_all_regions_enabled` | `mute_non_default_regions` | Boolean |
| `drs_job_exist` | `mute_non_default_regions` | Boolean |
| `guardduty_is_enabled` | `mute_non_default_regions` | Boolean |
| `securityhub_enabled` | `mute_non_default_regions` | Boolean |
## Azure
@@ -50,8 +50,8 @@ The following list includes all the AWS checks with configurable variables that
aws:
# AWS Global Configuration
# aws.allowlist_non_default_regions --> Allowlist Failed Findings in non-default regions for GuardDuty, SecurityHub, DRS and Config
allowlist_non_default_regions: False
# aws.mute_non_default_regions --> Mute Failed Findings in non-default regions for GuardDuty, SecurityHub, DRS and Config
mute_non_default_regions: False
# AWS IAM Configuration
# aws.iam_user_accesskey_unused --> CIS recommends 45 days

Before

Width:  |  Height:  |  Size: 10 KiB

After

Width:  |  Height:  |  Size: 10 KiB

Before

Width:  |  Height:  |  Size: 94 KiB

After

Width:  |  Height:  |  Size: 94 KiB

+1 -1
View File
@@ -8,7 +8,7 @@ There are different log levels depending on the logging information that is desi
- **DEBUG**: It will show low-level logs from Python.
- **INFO**: It will show all the API calls that are being invoked by the provider.
- **WARNING**: It will show all resources that are being **allowlisted**.
- **WARNING**: It will show all resources that are being **muted**.
- **ERROR**: It will show any errors, e.g., not authorized actions.
- **CRITICAL**: The default log level. If a critical log appears, it will **exit** Prowlers execution.
@@ -1,19 +1,19 @@
# Allowlisting
# Mute Listing
Sometimes you may find resources that are intentionally configured in a certain way that may be a bad practice but it is all right with it, for example an AWS S3 Bucket open to the internet hosting a web site, or an AWS Security Group with an open port needed in your use case.
Allowlist option works along with other options and adds a `WARNING` instead of `INFO`, `PASS` or `FAIL` to any output format.
Mute List option works along with other options and adds a `MUTED` instead of `INFO`, `PASS` or `FAIL` to any output format.
You can use `-w`/`--allowlist-file` with the path of your allowlist yaml file, but first, let's review the syntax.
You can use `-w`/`--mutelist-file` with the path of your mutelist yaml file, but first, let's review the syntax.
## Allowlist Yaml File Syntax
## Mute List Yaml File Syntax
### 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.
########################### ALLOWLIST EXAMPLE ###########################
Allowlist:
########################### MUTE LIST EXAMPLE ###########################
Mute List:
Accounts:
"123456789012":
Checks:
@@ -79,10 +79,10 @@ You can use `-w`/`--allowlist-file` with the path of your allowlist yaml file, b
Tags:
- "environment=prod" # Will ignore every resource except in account 123456789012 except the ones containing the string "test" and tag environment=prod
## Allowlist specific regions
If you want to allowlist/mute failed findings only in specific regions, create a file with the following syntax and run it with `prowler aws -w allowlist.yaml`:
## Mute specific regions
If you want to mute failed findings only in specific regions, create a file with the following syntax and run it with `prowler aws -w mutelist.yaml`:
Allowlist:
Mute List:
Accounts:
"*":
Checks:
@@ -93,50 +93,50 @@ If you want to allowlist/mute failed findings only in specific regions, create a
Resources:
- "*"
## Default AWS Allowlist
Prowler provides you a Default AWS Allowlist with the AWS Resources that should be allowlisted such as all resources created by AWS Control Tower when setting up a landing zone.
You can execute Prowler with this allowlist using the following command:
## Default AWS Mute List
Prowler provides you a Default AWS Mute List with the AWS Resources that should be muted such as all resources created by AWS Control Tower when setting up a landing zone.
You can execute Prowler with this mutelist using the following command:
```sh
prowler aws --allowlist prowler/config/aws_allowlist.yaml
prowler aws --mutelist prowler/config/aws_mutelist.yaml
```
## Supported Allowlist Locations
## Supported Mute List Locations
The allowlisting flag supports the following locations:
The mutelisting flag supports the following locations:
### Local file
You will need to pass the local path where your Allowlist YAML file is located:
You will need to pass the local path where your Mute List YAML file is located:
```
prowler <provider> -w allowlist.yaml
prowler <provider> -w mutelist.yaml
```
### AWS S3 URI
You will need to pass the S3 URI where your Allowlist YAML file was uploaded to your bucket:
You will need to pass the S3 URI where your Mute List YAML file was uploaded to your bucket:
```
prowler aws -w s3://<bucket>/<prefix>/allowlist.yaml
prowler aws -w s3://<bucket>/<prefix>/mutelist.yaml
```
> Make sure that the used AWS credentials have s3:GetObject permissions in the S3 path where the allowlist file is located.
> Make sure that the used AWS credentials have s3:GetObject permissions in the S3 path where the mutelist file is located.
### AWS DynamoDB Table ARN
You will need to pass the DynamoDB Allowlist Table ARN:
You will need to pass the DynamoDB Mute List Table ARN:
```
prowler aws -w arn:aws:dynamodb:<region_name>:<account_id>:table/<table_name>
```
1. The DynamoDB Table must have the following String keys:
<img src="../img/allowlist-keys.png"/>
<img src="../img/mutelist-keys.png"/>
- The Allowlist Table must have the following columns:
- Accounts (String): This field can contain either an Account ID or an `*` (which applies to all the accounts that use this table as an allowlist).
- The Mute List Table must have the following columns:
- Accounts (String): This field can contain either an Account ID or an `*` (which applies to all the accounts that use this table as an mutelist).
- Checks (String): This field can contain either a Prowler Check Name or an `*` (which applies to all the scanned checks).
- Regions (List): This field contains a list of regions where this allowlist rule is applied (it can also contains an `*` to apply all scanned regions).
- Resources (List): This field contains a list of regex expressions that applies to the resources that are wanted to be allowlisted.
- Tags (List): -Optional- This field contains a list of tuples in the form of 'key=value' that applies to the resources tags that are wanted to be allowlisted.
- Exceptions (Map): -Optional- This field contains a map of lists of accounts/regions/resources/tags that are wanted to be excepted in the allowlist.
- Regions (List): This field contains a list of regions where this mutelist rule is applied (it can also contains an `*` to apply all scanned regions).
- Resources (List): This field contains a list of regex expressions that applies to the resources that are wanted to be muted.
- Tags (List): -Optional- This field contains a list of tuples in the form of 'key=value' that applies to the resources tags that are wanted to be muted.
- Exceptions (Map): -Optional- This field contains a map of lists of accounts/regions/resources/tags that are wanted to be excepted in the mutelist.
The following example will allowlist all resources in all accounts for the EC2 checks in the regions `eu-west-1` and `us-east-1` with the tags `environment=dev` and `environment=prod`, except the resources containing the string `test` in the account `012345678912` and region `eu-west-1` with the tag `environment=prod`:
The following example will mute all resources in all accounts for the EC2 checks in the regions `eu-west-1` and `us-east-1` with the tags `environment=dev` and `environment=prod`, except the resources containing the string `test` in the account `012345678912` and region `eu-west-1` with the tag `environment=prod`:
<img src="../img/allowlist-row.png"/>
<img src="../img/mutelist-row.png"/>
> Make sure that the used AWS credentials have `dynamodb:PartiQLSelect` permissions in the table.
@@ -151,7 +151,7 @@ prowler aws -w arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME
Make sure that the credentials that Prowler uses can invoke the Lambda Function:
```
- PolicyName: GetAllowList
- PolicyName: GetMuteList
PolicyDocument:
Version: '2012-10-17'
Statement:
@@ -160,14 +160,14 @@ Make sure that the credentials that Prowler uses can invoke the Lambda Function:
Resource: arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME
```
The Lambda Function can then generate an Allowlist dynamically. Here is the code an example Python Lambda Function that
generates an Allowlist:
The Lambda Function can then generate an Mute List dynamically. Here is the code an example Python Lambda Function that
generates an Mute List:
```
def handler(event, context):
checks = {}
checks["vpc_flow_logs_enabled"] = { "Regions": [ "*" ], "Resources": [ "" ], Optional("Tags"): [ "key:value" ] }
al = { "Allowlist": { "Accounts": { "*": { "Checks": checks } } } }
al = { "Mute List": { "Accounts": { "*": { "Checks": checks } } } }
return al
```
+1 -1
View File
@@ -36,7 +36,7 @@ nav:
- Slack Integration: tutorials/integrations.md
- Configuration File: tutorials/configuration_file.md
- Logging: tutorials/logging.md
- Allowlist: tutorials/allowlist.md
- Mute List: tutorials/mutelist.md
- Check Aliases: tutorials/check-aliases.md
- Custom Metadata: tutorials/custom-checks-metadata.md
- Ignore Unused Services: tutorials/ignore-unused-services.md
+4 -4
View File
@@ -49,12 +49,12 @@ from prowler.providers.aws.lib.security_hub.security_hub import (
resolve_security_hub_previous_findings,
verify_security_hub_integration_enabled_per_region,
)
from prowler.providers.common.allowlist import set_provider_allowlist
from prowler.providers.common.audit_info import (
set_provider_audit_info,
set_provider_execution_parameters,
)
from prowler.providers.common.clean import clean_provider_local_output_directories
from prowler.providers.common.mutelist import set_provider_mutelist
from prowler.providers.common.outputs import set_provider_output_options
from prowler.providers.common.quick_inventory import run_provider_quick_inventory
@@ -178,12 +178,12 @@ def prowler():
# Sort final check list
checks_to_execute = sorted(checks_to_execute)
# Parse Allowlist
allowlist_file = set_provider_allowlist(provider, audit_info, args)
# Parse Mute List
mutelist_file = set_provider_mutelist(provider, audit_info, args)
# Set output options based on the selected provider
audit_output_options = set_provider_output_options(
provider, args, audit_info, allowlist_file, bulk_checks_metadata
provider, args, audit_info, mutelist_file, bulk_checks_metadata
)
# Run the quick inventory for the provider if available
@@ -1,4 +1,4 @@
Allowlist:
Mute List:
Accounts:
"*":
########################### AWS CONTROL TOWER ###########################
@@ -3,8 +3,8 @@
### 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.
########################### ALLOWLIST EXAMPLE ###########################
Allowlist:
########################### MUTE LIST EXAMPLE ###########################
Mute List:
Accounts:
"123456789012":
Checks:
+4 -4
View File
@@ -2,10 +2,10 @@
aws:
# AWS Global Configuration
# aws.allowlist_non_default_regions --> Set to True to allowlist failed findings in non-default regions for AccessAnalyzer, GuardDuty, SecurityHub, DRS and Config
allowlist_non_default_regions: False
# If you want to allowlist/mute failed findings only in specific regions, create a file with the following syntax and run it with `prowler aws -w allowlist.yaml`:
# Allowlist:
# aws.mute_non_default_regions --> Set to True to mute failed findings in non-default regions for GuardDuty, SecurityHub, DRS and Config
mute_non_default_regions: False
# If you want to mute failed findings only in specific regions, create a file with the following syntax and run it with `prowler aws -w mutelist.yaml`:
# Mute List:
# Accounts:
# "*":
# Checks:
+1 -1
View File
@@ -21,7 +21,7 @@ def print_banner(args):
Color code for results:
- {Fore.YELLOW}INFO (Information){Style.RESET_ALL}
- {Fore.GREEN}PASS (Recommended value){Style.RESET_ALL}
- {orange_color}WARNING (Ignored by allowlist){Style.RESET_ALL}
- {orange_color}MUTED (Muted by muted list){Style.RESET_ALL}
- {Fore.RED}FAIL (Fix required){Style.RESET_ALL}
"""
)
+5 -5
View File
@@ -21,7 +21,7 @@ from prowler.lib.logger import logger
from prowler.lib.outputs.outputs import report
from prowler.lib.ui.live_display import live_display
from prowler.lib.utils.utils import open_file, parse_json_file
from prowler.providers.aws.lib.allowlist.allowlist import allowlist_findings
from prowler.providers.aws.lib.mutelist.mutelist import mutelist_findings
from prowler.providers.common.models import Audit_Metadata
from prowler.providers.common.outputs import Provider_Output_Options
@@ -576,10 +576,10 @@ def execute(
audit_info.audit_metadata, services_executed, checks_executed
)
# Allowlist findings
if audit_output_options.allowlist_file:
check_findings = allowlist_findings(
audit_output_options.allowlist_file,
# Mute List findings
if audit_output_options.mutelist_file:
check_findings = mutelist_findings(
audit_output_options.mutelist_file,
audit_info.audited_account,
check_findings,
)
+1 -1
View File
@@ -173,7 +173,7 @@ def fill_html(file_descriptor, finding, output_options):
row_class = "table-info"
elif finding.status == "FAIL":
row_class = "table-danger"
elif finding.status == "WARNING":
elif finding.status == "MUTED":
row_class = "table-warning"
file_descriptor.write(
f"""
+4 -4
View File
@@ -116,8 +116,8 @@ def generate_json_asff_status(status: str) -> str:
json_asff_status = "PASSED"
elif status == "FAIL":
json_asff_status = "FAILED"
elif status == "WARNING":
json_asff_status = "WARNING"
elif status == "MUTED":
json_asff_status = "MUTED"
else:
json_asff_status = "NOT_AVAILABLE"
@@ -293,7 +293,7 @@ def generate_json_ocsf_status(status: str):
json_ocsf_status = "Success"
elif status == "FAIL":
json_ocsf_status = "Failure"
elif status == "WARNING":
elif status == "MUTED":
json_ocsf_status = "Other"
else:
json_ocsf_status = "Unknown"
@@ -307,7 +307,7 @@ def generate_json_ocsf_status_id(status: str):
json_ocsf_status_id = 1
elif status == "FAIL":
json_ocsf_status_id = 2
elif status == "WARNING":
elif status == "MUTED":
json_ocsf_status_id = 99
else:
json_ocsf_status_id = 0
+2 -2
View File
@@ -165,12 +165,12 @@ def set_report_color(status: str) -> str:
color = Fore.RED
elif status == "ERROR":
color = Fore.BLACK
elif status == "WARNING":
elif status == "MUTED":
color = orange_color
elif status == "INFO":
color = Fore.YELLOW
else:
raise Exception("Invalid Report Status. Must be PASS, FAIL, ERROR or WARNING")
raise Exception("Invalid Report Status. Must be PASS, FAIL, ERROR or MUTED")
return color
@@ -125,14 +125,14 @@ def init_parser(self):
default=None,
help="Shodan API key used by check ec2_elastic_ip_shodan.",
)
# Allowlist
allowlist_subparser = aws_parser.add_argument_group("Allowlist")
allowlist_subparser.add_argument(
# Mute List
mutelist_subparser = aws_parser.add_argument_group("Mute List")
mutelist_subparser.add_argument(
"-w",
"--allowlist-file",
"--mutelist-file",
nargs="?",
default=None,
help="Path for allowlist yaml file. See example prowler/config/aws_allowlist.yaml for reference and format. It also accepts AWS DynamoDB Table or Lambda ARNs or S3 URIs, see more in https://docs.prowler.cloud/en/latest/tutorials/allowlist/",
help="Path for mutelist yaml file. See example prowler/config/aws_mutelist.yaml for reference and format. It also accepts AWS DynamoDB Table or Lambda ARNs or S3 URIs, see more in https://docs.prowler.cloud/en/latest/tutorials/mutelist/",
)
# Based Scans
@@ -9,7 +9,7 @@ from schema import Optional, Schema
from prowler.lib.logger import logger
from prowler.lib.outputs.models import unroll_tags
allowlist_schema = Schema(
mutelist_schema = Schema(
{
"Accounts": {
str: {
@@ -32,38 +32,38 @@ allowlist_schema = Schema(
)
def parse_allowlist_file(audit_info, allowlist_file):
def parse_mutelist_file(audit_info, mutelist_file):
try:
# Check if file is a S3 URI
if re.search("^s3://([^/]+)/(.*?([^/]+))$", allowlist_file):
bucket = allowlist_file.split("/")[2]
key = ("/").join(allowlist_file.split("/")[3:])
if re.search("^s3://([^/]+)/(.*?([^/]+))$", mutelist_file):
bucket = mutelist_file.split("/")[2]
key = ("/").join(mutelist_file.split("/")[3:])
s3_client = audit_info.audit_session.client("s3")
allowlist = yaml.safe_load(
mutelist = yaml.safe_load(
s3_client.get_object(Bucket=bucket, Key=key)["Body"]
)["Allowlist"]
)["Mute List"]
# Check if file is a Lambda Function ARN
elif re.search(r"^arn:(\w+):lambda:", allowlist_file):
lambda_region = allowlist_file.split(":")[3]
elif re.search(r"^arn:(\w+):lambda:", mutelist_file):
lambda_region = mutelist_file.split(":")[3]
lambda_client = audit_info.audit_session.client(
"lambda", region_name=lambda_region
)
lambda_response = lambda_client.invoke(
FunctionName=allowlist_file, InvocationType="RequestResponse"
FunctionName=mutelist_file, InvocationType="RequestResponse"
)
lambda_payload = lambda_response["Payload"].read()
allowlist = yaml.safe_load(lambda_payload)["Allowlist"]
mutelist = yaml.safe_load(lambda_payload)["Mute List"]
# 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._-]+$",
allowlist_file,
mutelist_file,
):
allowlist = {"Accounts": {}}
table_region = allowlist_file.split(":")[3]
mutelist = {"Accounts": {}}
table_region = mutelist_file.split(":")[3]
dynamodb_resource = audit_info.audit_session.resource(
"dynamodb", region_name=table_region
)
dynamo_table = dynamodb_resource.Table(allowlist_file.split("/")[1])
dynamo_table = dynamodb_resource.Table(mutelist_file.split("/")[1])
response = dynamo_table.scan(
FilterExpression=Attr("Accounts").is_in(
[audit_info.audited_account, "*"]
@@ -80,8 +80,8 @@ def parse_allowlist_file(audit_info, allowlist_file):
)
dynamodb_items.update(response["Items"])
for item in dynamodb_items:
# Create allowlist for every item
allowlist["Accounts"][item["Accounts"]] = {
# Create mutelist for every item
mutelist["Accounts"][item["Accounts"]] = {
"Checks": {
item["Checks"]: {
"Regions": item["Regions"],
@@ -90,24 +90,24 @@ def parse_allowlist_file(audit_info, allowlist_file):
}
}
if "Tags" in item:
allowlist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][
mutelist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][
"Tags"
] = item["Tags"]
if "Exceptions" in item:
allowlist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][
mutelist["Accounts"][item["Accounts"]]["Checks"][item["Checks"]][
"Exceptions"
] = item["Exceptions"]
else:
with open(allowlist_file) as f:
allowlist = yaml.safe_load(f)["Allowlist"]
with open(mutelist_file) as f:
mutelist = yaml.safe_load(f)["Mute List"]
try:
allowlist_schema.validate(allowlist)
mutelist_schema.validate(mutelist)
except Exception as error:
logger.critical(
f"{error.__class__.__name__} -- Allowlist YAML is malformed - {error}[{error.__traceback__.tb_lineno}]"
f"{error.__class__.__name__} -- Mute List YAML is malformed - {error}[{error.__traceback__.tb_lineno}]"
)
sys.exit(1)
return allowlist
return mutelist
except Exception as error:
logger.critical(
f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]"
@@ -115,27 +115,27 @@ def parse_allowlist_file(audit_info, allowlist_file):
sys.exit(1)
def allowlist_findings(
allowlist: dict,
def mutelist_findings(
mutelist: dict,
audited_account: str,
check_findings: [Any],
):
# Check if finding is allowlisted
# Check if finding is muted
for finding in check_findings:
if is_allowlisted(
allowlist,
if is_muted(
mutelist,
audited_account,
finding.check_metadata.CheckID,
finding.region,
finding.resource_id,
unroll_tags(finding.resource_tags),
):
finding.status = "WARNING"
finding.status = "MUTED"
return check_findings
def is_allowlisted(
allowlist: dict,
def is_muted(
mutelist: dict,
audited_account: str,
check: str,
finding_region: str,
@@ -143,31 +143,30 @@ def is_allowlisted(
finding_tags,
):
try:
allowlisted_checks = {}
# By default is not allowlisted
is_finding_allowlisted = False
# First set account key from allowlist dict
if audited_account in allowlist["Accounts"]:
allowlisted_checks = allowlist["Accounts"][audited_account]["Checks"]
muted_checks = {}
# By default is not muted
is_finding_muted = False
# First set account key from mutelist dict
if audited_account in mutelist["Accounts"]:
muted_checks = mutelist["Accounts"][audited_account]["Checks"]
# If there is a *, it affects to all accounts
# This cannot be elif since in the case of * and single accounts we
# want to merge allowlisted checks from * to the other accounts check list
if "*" in allowlist["Accounts"]:
checks_multi_account = allowlist["Accounts"]["*"]["Checks"]
allowlisted_checks.update(checks_multi_account)
# Test if it is allowlisted
if is_allowlisted_in_check(
allowlisted_checks,
# want to merge muted checks from * to the other accounts check list
if "*" in mutelist["Accounts"]:
checks_multi_account = mutelist["Accounts"]["*"]["Checks"]
muted_checks.update(checks_multi_account)
# Test if it is muted
if is_muted_in_check(
muted_checks,
audited_account,
check,
finding_region,
finding_resource,
finding_tags,
):
is_finding_allowlisted = True
is_finding_muted = True
return is_finding_allowlisted
return is_finding_muted
except Exception as error:
logger.critical(
f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]"
@@ -175,8 +174,8 @@ def is_allowlisted(
sys.exit(1)
def is_allowlisted_in_check(
allowlisted_checks,
def is_muted_in_check(
muted_checks,
audited_account,
check,
finding_region,
@@ -184,15 +183,15 @@ def is_allowlisted_in_check(
finding_tags,
):
try:
# Default value is not allowlisted
is_check_allowlisted = False
# Default value is not muted
is_check_muted = False
for allowlisted_check, allowlisted_check_info in allowlisted_checks.items():
for muted_check, muted_check_info in muted_checks.items():
# map lambda to awslambda
allowlisted_check = re.sub("^lambda", "awslambda", allowlisted_check)
muted_check = re.sub("^lambda", "awslambda", muted_check)
# Check if the finding is excepted
exceptions = allowlisted_check_info.get("Exceptions")
exceptions = muted_check_info.get("Exceptions")
if is_excepted(
exceptions,
audited_account,
@@ -203,40 +202,36 @@ def is_allowlisted_in_check(
# Break loop and return default value since is excepted
break
allowlisted_regions = allowlisted_check_info.get("Regions")
allowlisted_resources = allowlisted_check_info.get("Resources")
allowlisted_tags = allowlisted_check_info.get("Tags")
muted_regions = muted_check_info.get("Regions")
muted_resources = muted_check_info.get("Resources")
muted_tags = muted_check_info.get("Tags")
# If there is a *, it affects to all checks
if (
"*" == allowlisted_check
or check == allowlisted_check
or re.search(allowlisted_check, check)
"*" == muted_check
or check == muted_check
or re.search(muted_check, check)
):
allowlisted_in_check = True
allowlisted_in_region = is_allowlisted_in_region(
allowlisted_regions, finding_region
)
allowlisted_in_resource = is_allowlisted_in_resource(
allowlisted_resources, finding_resource
)
allowlisted_in_tags = is_allowlisted_in_tags(
allowlisted_tags, finding_tags
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)
# For a finding to be allowlisted requires the following set to True:
# - allowlisted_in_check -> True
# - allowlisted_in_region -> True
# - allowlisted_in_tags -> True or allowlisted_in_resource -> True
# For a finding to be muted requires the following set to True:
# - muted_in_check -> True
# - muted_in_region -> True
# - muted_in_tags -> True or muted_in_resource -> True
# - excepted -> False
if (
allowlisted_in_check
and allowlisted_in_region
and (allowlisted_in_tags or allowlisted_in_resource)
muted_in_check
and muted_in_region
and (muted_in_tags or muted_in_resource)
):
is_check_allowlisted = True
is_check_muted = True
return is_check_allowlisted
return is_check_muted
except Exception as error:
logger.critical(
f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]"
@@ -244,12 +239,12 @@ def is_allowlisted_in_check(
sys.exit(1)
def is_allowlisted_in_region(
allowlisted_regions,
def is_muted_in_region(
mutelist_regions,
finding_region,
):
try:
return __is_item_matched__(allowlisted_regions, finding_region)
return __is_item_matched__(mutelist_regions, finding_region)
except Exception as error:
logger.critical(
f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]"
@@ -257,9 +252,9 @@ def is_allowlisted_in_region(
sys.exit(1)
def is_allowlisted_in_tags(allowlisted_tags, finding_tags):
def is_muted_in_tags(muted_tags, finding_tags):
try:
return __is_item_matched__(allowlisted_tags, finding_tags)
return __is_item_matched__(muted_tags, finding_tags)
except Exception as error:
logger.critical(
f"{error.__class__.__name__} -- {error}[{error.__traceback__.tb_lineno}]"
@@ -267,9 +262,9 @@ def is_allowlisted_in_tags(allowlisted_tags, finding_tags):
sys.exit(1)
def is_allowlisted_in_resource(allowlisted_resources, finding_resource):
def is_muted_in_resource(muted_resources, finding_resource):
try:
return __is_item_matched__(allowlisted_resources, finding_resource)
return __is_item_matched__(muted_resources, finding_resource)
except Exception as error:
logger.critical(
@@ -31,11 +31,11 @@ class accessanalyzer_enabled(Check):
)
if (
accessanalyzer_client.audit_config.get(
"allowlist_non_default_regions", False
"mute_non_default_regions", False
)
and not analyzer.region == accessanalyzer_client.region
):
report.status = "WARNING"
report.status = "MUTED"
findings.append(report)
@@ -36,10 +36,10 @@ class config_recorder_all_regions_enabled(Check):
f"AWS Config recorder {recorder.name} is disabled."
)
if report.status == "FAIL" and (
config_client.audit_config.get("allowlist_non_default_regions", False)
config_client.audit_config.get("mute_non_default_regions", False)
and not recorder.region == config_client.region
):
report.status = "WARNING"
report.status = "MUTED"
findings.append(report)
@@ -20,10 +20,10 @@ class drs_job_exist(Check):
report.status_extended = "DRS is enabled for this region with jobs."
if report.status == "FAIL" and (
drs_client.audit_config.get("allowlist_non_default_regions", False)
drs_client.audit_config.get("mute_non_default_regions", False)
and not drs.region == drs_client.region
):
report.status = "WARNING"
report.status = "MUTED"
findings.append(report)
@@ -29,12 +29,10 @@ class guardduty_is_enabled(Check):
)
if report.status == "FAIL" and (
guardduty_client.audit_config.get(
"allowlist_non_default_regions", False
)
guardduty_client.audit_config.get("mute_non_default_regions", False)
and not detector.region == guardduty_client.region
):
report.status = "WARNING"
report.status = "MUTED"
findings.append(report)
@@ -26,12 +26,10 @@ class securityhub_enabled(Check):
report.status_extended = "Security Hub is not enabled."
if report.status == "FAIL" and (
securityhub_client.audit_config.get(
"allowlist_non_default_regions", False
)
securityhub_client.audit_config.get("mute_non_default_regions", False)
and not securityhub.region == securityhub_client.region
):
report.status = "WARNING"
report.status = "MUTED"
findings.append(report)
-35
View File
@@ -1,35 +0,0 @@
import importlib
import sys
from prowler.lib.logger import logger
from prowler.providers.aws.lib.allowlist.allowlist import parse_allowlist_file
def set_provider_allowlist(provider, audit_info, args):
"""
set_provider_allowlist configures the allowlist based on the selected provider.
"""
try:
# Check if the provider arguments has the allowlist_file
if hasattr(args, "allowlist_file"):
# Dynamically get the Provider allowlist handler
provider_allowlist_function = f"set_{provider}_allowlist"
allowlist_file = getattr(
importlib.import_module(__name__), provider_allowlist_function
)(audit_info, args.allowlist_file)
return allowlist_file
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
def set_aws_allowlist(audit_info, allowlist_file):
# Parse content from Allowlist file and get it, if necessary, from S3
if allowlist_file:
allowlist_file = parse_allowlist_file(audit_info, allowlist_file)
else:
allowlist_file = None
return allowlist_file
+35
View File
@@ -0,0 +1,35 @@
import importlib
import sys
from prowler.lib.logger import logger
from prowler.providers.aws.lib.mutelist.mutelist import parse_mutelist_file
def set_provider_mutelist(provider, audit_info, args):
"""
set_provider_mutelist configures the mutelist based on the selected provider.
"""
try:
# Check if the provider arguments has the mutelist_file
if hasattr(args, "mutelist_file"):
# Dynamically get the Provider mutelist handler
provider_mutelist_function = f"set_{provider}_mutelist"
mutelist_file = getattr(
importlib.import_module(__name__), provider_mutelist_function
)(audit_info, args.mutelist_file)
return mutelist_file
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
sys.exit(1)
def set_aws_mutelist(audit_info, mutelist_file):
# Parse content from Mute List file and get it, if necessary, from S3
if mutelist_file:
mutelist_file = parse_mutelist_file(audit_info, mutelist_file)
else:
mutelist_file = None
return mutelist_file
+11 -11
View File
@@ -9,7 +9,7 @@ from prowler.lib.logger import logger
def set_provider_output_options(
provider: str, arguments, audit_info, allowlist_file, bulk_checks_metadata
provider: str, arguments, audit_info, mutelist_file, bulk_checks_metadata
):
"""
set_provider_output_options configures automatically the outputs based on the selected provider and returns the Provider_Output_Options object.
@@ -19,7 +19,7 @@ def set_provider_output_options(
provider_output_class = f"{provider.capitalize()}_Output_Options"
provider_output_options = getattr(
importlib.import_module(__name__), provider_output_class
)(arguments, audit_info, allowlist_file, bulk_checks_metadata)
)(arguments, audit_info, mutelist_file, bulk_checks_metadata)
except Exception as error:
logger.critical(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
@@ -49,20 +49,20 @@ class Provider_Output_Options:
is_quiet: bool
output_modes: list
output_directory: str
allowlist_file: str
mutelist_file: str
bulk_checks_metadata: dict
verbose: str
output_filename: str
only_logs: bool
unix_timestamp: bool
def __init__(self, arguments, allowlist_file, bulk_checks_metadata):
def __init__(self, arguments, mutelist_file, bulk_checks_metadata):
self.is_quiet = arguments.quiet
self.output_modes = arguments.output_modes
self.output_directory = arguments.output_directory
self.verbose = arguments.verbose
self.bulk_checks_metadata = bulk_checks_metadata
self.allowlist_file = allowlist_file
self.mutelist_file = mutelist_file
self.only_logs = arguments.only_logs
self.unix_timestamp = arguments.unix_timestamp
# Check output directory, if it is not created -> create it
@@ -74,9 +74,9 @@ class Provider_Output_Options:
class Azure_Output_Options(Provider_Output_Options):
def __init__(self, arguments, audit_info, allowlist_file, bulk_checks_metadata):
def __init__(self, arguments, audit_info, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, allowlist_file, bulk_checks_metadata)
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# Check if custom output filename was input, if not, set the default
if (
@@ -95,9 +95,9 @@ class Azure_Output_Options(Provider_Output_Options):
class Gcp_Output_Options(Provider_Output_Options):
def __init__(self, arguments, audit_info, allowlist_file, bulk_checks_metadata):
def __init__(self, arguments, audit_info, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, allowlist_file, bulk_checks_metadata)
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# Check if custom output filename was input, if not, set the default
if (
@@ -112,9 +112,9 @@ class Gcp_Output_Options(Provider_Output_Options):
class Aws_Output_Options(Provider_Output_Options):
security_hub_enabled: bool
def __init__(self, arguments, audit_info, allowlist_file, bulk_checks_metadata):
def __init__(self, arguments, audit_info, mutelist_file, bulk_checks_metadata):
# First call Provider_Output_Options init
super().__init__(arguments, allowlist_file, bulk_checks_metadata)
super().__init__(arguments, mutelist_file, bulk_checks_metadata)
# Confire Shodan API
if arguments.shodan:
+10 -10
View File
@@ -76,7 +76,7 @@ class Test_Parser:
assert not parsed.output_bucket
assert not parsed.output_bucket_no_assume
assert not parsed.shodan
assert not parsed.allowlist_file
assert not parsed.mutelist_file
assert not parsed.resource_tags
assert not parsed.ignore_unused_services
@@ -943,19 +943,19 @@ class Test_Parser:
parsed = self.parser.parse(command)
assert parsed.shodan == shodan_api_key
def test_aws_parser_allowlist_short(self):
def test_aws_parser_mutelist_short(self):
argument = "-w"
allowlist_file = "allowlist.txt"
command = [prowler_command, argument, allowlist_file]
mutelist_file = "mutelist.txt"
command = [prowler_command, argument, mutelist_file]
parsed = self.parser.parse(command)
assert parsed.allowlist_file == allowlist_file
assert parsed.mutelist_file == mutelist_file
def test_aws_parser_allowlist_long(self):
argument = "--allowlist-file"
allowlist_file = "allowlist.txt"
command = [prowler_command, argument, allowlist_file]
def test_aws_parser_mutelist_long(self):
argument = "--mutelist-file"
mutelist_file = "mutelist.txt"
command = [prowler_command, argument, mutelist_file]
parsed = self.parser.parse(command)
assert parsed.allowlist_file == allowlist_file
assert parsed.mutelist_file == mutelist_file
def test_aws_parser_resource_tags(self):
argument = "--resource-tags"
+5 -5
View File
@@ -165,7 +165,7 @@ class Test_Outputs:
remove(expected[index][output_mode].name)
def test_set_report_color(self):
test_status = ["PASS", "FAIL", "ERROR", "WARNING"]
test_status = ["PASS", "FAIL", "ERROR", "MUTED"]
test_colors = [Fore.GREEN, Fore.RED, Fore.BLACK, orange_color]
for status in test_status:
@@ -177,7 +177,7 @@ class Test_Outputs:
with pytest.raises(Exception) as exc:
set_report_color(test_status)
assert "Invalid Report Status. Must be PASS, FAIL, ERROR or WARNING" in str(
assert "Invalid Report Status. Must be PASS, FAIL, ERROR or MUTED" in str(
exc.value
)
assert exc.type == Exception
@@ -1347,7 +1347,7 @@ class Test_Outputs:
def test_generate_json_asff_status(self):
assert generate_json_asff_status("PASS") == "PASSED"
assert generate_json_asff_status("FAIL") == "FAILED"
assert generate_json_asff_status("WARNING") == "WARNING"
assert generate_json_asff_status("MUTED") == "MUTED"
assert generate_json_asff_status("SOMETHING ELSE") == "NOT_AVAILABLE"
def test_generate_json_asff_resource_tags(self):
@@ -1364,13 +1364,13 @@ class Test_Outputs:
def test_generate_json_ocsf_status(self):
assert generate_json_ocsf_status("PASS") == "Success"
assert generate_json_ocsf_status("FAIL") == "Failure"
assert generate_json_ocsf_status("WARNING") == "Other"
assert generate_json_ocsf_status("MUTED") == "Other"
assert generate_json_ocsf_status("SOMETHING ELSE") == "Unknown"
def test_generate_json_ocsf_status_id(self):
assert generate_json_ocsf_status_id("PASS") == 1
assert generate_json_ocsf_status_id("FAIL") == 2
assert generate_json_ocsf_status_id("WARNING") == 99
assert generate_json_ocsf_status_id("MUTED") == 99
assert generate_json_ocsf_status_id("SOMETHING ELSE") == 0
def test_generate_json_ocsf_severity_id(self):
@@ -1,7 +1,10 @@
### Account, Check and/or Region can be * to apply for all the cases
### Resources is a list that can have either Regex or Keywords:
########################### ALLOWLIST EXAMPLE ###########################
Allowlist:
### 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.
########################### MUTE LIST EXAMPLE ###########################
Mute List:
Accounts:
"123456789012":
Checks:
@@ -29,7 +32,7 @@ Allowlist:
- ".+-logs" # Will ignore all buckets containing the terms ci-logs, qa-logs, etc. in specified check and regions
# EXAMPLE: CONTROL TOWER (to migrate)
# When using Control Tower, guardrails prevent access to certain protected resources. The allowlist
# When using Control Tower, guardrails prevent access to certain protected resources. The mutelist
# below ensures that warnings instead of errors are reported for the affected resources.
#extra734:aws-controltower-logs-[[:digit:]]+-[[:alpha:]\-]+
#extra734:aws-controltower-s3-access-logs-[[:digit:]]+-[[:alpha:]\-]+
@@ -3,16 +3,18 @@ from boto3 import resource
from mock import MagicMock
from moto import mock_dynamodb, mock_s3
from prowler.providers.aws.lib.allowlist.allowlist import (
allowlist_findings,
is_allowlisted,
is_allowlisted_in_check,
is_allowlisted_in_region,
is_allowlisted_in_resource,
is_allowlisted_in_tags,
from prowler.providers.aws.lib.audit_info.models import AWS_Audit_Info
from prowler.providers.aws.lib.mutelist.mutelist import (
is_excepted,
parse_allowlist_file,
is_muted,
is_muted_in_check,
is_muted_in_region,
is_muted_in_resource,
is_muted_in_tags,
mutelist_findings,
parse_mutelist_file,
)
from prowler.providers.common.models import Audit_Metadata
from tests.providers.aws.audit_info_utils import (
AWS_ACCOUNT_NUMBER,
AWS_REGION_EU_WEST_1,
@@ -21,33 +23,64 @@ from tests.providers.aws.audit_info_utils import (
)
class Test_Allowlist:
# Test S3 allowlist
class Test_Mutelist:
# Mocked Audit Info
def set_mocked_audit_info(self):
audit_info = AWS_Audit_Info(
session_config=None,
original_session=None,
audit_session=session.Session(
profile_name=None,
botocore_session=None,
),
audited_account=AWS_ACCOUNT_NUMBER,
audited_account_arn=f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root",
audited_user_id=None,
audited_partition="aws",
audited_identity_arn=None,
profile=None,
profile_region=None,
credentials=None,
assumed_role_info=None,
audited_regions=None,
organizations_metadata=None,
audit_resources=None,
mfa_enabled=False,
audit_metadata=Audit_Metadata(
services_scanned=0,
expected_checks=[],
completed_checks=0,
audit_progress=0,
),
)
return audit_info
# Test S3 mutelist
@mock_s3
def test_s3_allowlist(self):
audit_info = set_mocked_aws_audit_info()
# Create bucket and upload allowlist yaml
def test_s3_mutelist(self):
audit_info = self.set_mocked_audit_info()
# Create bucket and upload mutelist yaml
s3_resource = resource("s3", region_name=AWS_REGION_US_EAST_1)
s3_resource.create_bucket(Bucket="test-allowlist")
s3_resource.Object("test-allowlist", "allowlist.yaml").put(
s3_resource.create_bucket(Bucket="test-mutelist")
s3_resource.Object("test-mutelist", "mutelist.yaml").put(
Body=open(
"tests/providers/aws/lib/allowlist/fixtures/allowlist.yaml",
"tests/providers/aws/lib/mutelist/fixtures/mutelist.yaml",
"rb",
)
)
with open("tests/providers/aws/lib/allowlist/fixtures/allowlist.yaml") as f:
assert yaml.safe_load(f)["Allowlist"] == parse_allowlist_file(
audit_info, "s3://test-allowlist/allowlist.yaml"
with open("tests/providers/aws/lib/mutelist/fixtures/mutelist.yaml") as f:
assert yaml.safe_load(f)["Mute List"] == parse_mutelist_file(
audit_info, "s3://test-mutelist/mutelist.yaml"
)
# Test DynamoDB allowlist
# Test DynamoDB mutelist
@mock_dynamodb
def test_dynamo_allowlist(self):
audit_info = set_mocked_aws_audit_info()
def test_dynamo_mutelist(self):
audit_info = self.set_mocked_audit_info()
# Create table and put item
dynamodb_resource = resource("dynamodb", region_name=AWS_REGION_US_EAST_1)
table_name = "test-allowlist"
table_name = "test-mutelist"
params = {
"TableName": table_name,
"KeySchema": [
@@ -75,7 +108,7 @@ class Test_Allowlist:
assert (
"keyword"
in parse_allowlist_file(
in parse_mutelist_file(
audit_info,
"arn:aws:dynamodb:"
+ AWS_REGION_US_EAST_1
@@ -87,11 +120,11 @@ class Test_Allowlist:
)
@mock_dynamodb
def test_dynamo_allowlist_with_tags(self):
audit_info = set_mocked_aws_audit_info()
def test_dynamo_mutelist_with_tags(self):
audit_info = self.set_mocked_audit_info()
# Create table and put item
dynamodb_resource = resource("dynamodb", region_name=AWS_REGION_US_EAST_1)
table_name = "test-allowlist"
table_name = "test-mutelist"
params = {
"TableName": table_name,
"KeySchema": [
@@ -120,7 +153,7 @@ class Test_Allowlist:
assert (
"environment=dev"
in parse_allowlist_file(
in parse_mutelist_file(
audit_info,
"arn:aws:dynamodb:"
+ AWS_REGION_US_EAST_1
@@ -131,11 +164,11 @@ class Test_Allowlist:
)["Accounts"]["*"]["Checks"]["*"]["Tags"]
)
# Allowlist tests
# Mutelist tests
def test_allowlist_findings(self):
# Allowlist example
allowlist = {
def test_mutelist_findings(self):
# Mutelist example
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -160,14 +193,12 @@ class Test_Allowlist:
check_findings.append(finding_1)
allowlisted_findings = allowlist_findings(
allowlist, AWS_ACCOUNT_NUMBER, check_findings
)
assert len(allowlisted_findings) == 1
assert allowlisted_findings[0].status == "WARNING"
muted_findings = mutelist_findings(mutelist, AWS_ACCOUNT_NUMBER, check_findings)
assert len(muted_findings) == 1
assert muted_findings[0].status == "MUTED"
def test_is_allowlisted_with_everything_excepted(self):
allowlist = {
def test_is_muted_with_everything_excepted(self):
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -187,8 +218,8 @@ class Test_Allowlist:
}
}
assert not is_allowlisted(
allowlist,
assert not is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"athena_1",
AWS_REGION_US_EAST_1,
@@ -196,8 +227,8 @@ class Test_Allowlist:
"",
)
def test_is_allowlisted_with_default_allowlist(self):
allowlist = {
def test_is_muted_with_default_mutelist(self):
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -217,8 +248,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"athena_1",
AWS_REGION_US_EAST_1,
@@ -226,9 +257,9 @@ class Test_Allowlist:
"",
)
def test_is_allowlisted(self):
# Allowlist example
allowlist = {
def test_is_muted(self):
# Mutelist example
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -241,8 +272,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -250,8 +281,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -259,8 +290,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -268,8 +299,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -278,14 +309,14 @@ class Test_Allowlist:
)
assert not (
is_allowlisted(
allowlist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
is_muted(
mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
)
)
def test_is_allowlisted_wildcard(self):
# Allowlist example
allowlist = {
def test_is_muted_wildcard(self):
# Mutelist example
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -298,8 +329,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -307,8 +338,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -316,8 +347,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -326,14 +357,14 @@ class Test_Allowlist:
)
assert not (
is_allowlisted(
allowlist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
is_muted(
mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
)
)
def test_is_allowlisted_asterisk(self):
# Allowlist example
allowlist = {
def test_is_muted_asterisk(self):
# Mutelist example
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -346,8 +377,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -355,8 +386,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -364,8 +395,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -374,14 +405,14 @@ class Test_Allowlist:
)
assert not (
is_allowlisted(
allowlist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
is_muted(
mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
)
)
def test_is_allowlisted_all_and_single_account(self):
# Allowlist example
allowlist = {
def test_is_muted_all_and_single_account(self):
# Mutelist example
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -402,8 +433,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test_2",
AWS_REGION_US_EAST_1,
@@ -411,8 +442,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -420,8 +451,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -429,8 +460,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -439,13 +470,13 @@ class Test_Allowlist:
)
assert not (
is_allowlisted(
allowlist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
is_muted(
mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
)
)
def test_is_allowlisted_single_account(self):
allowlist = {
def test_is_muted_single_account(self):
mutelist = {
"Accounts": {
AWS_ACCOUNT_NUMBER: {
"Checks": {
@@ -458,8 +489,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -468,39 +499,39 @@ class Test_Allowlist:
)
assert not (
is_allowlisted(
allowlist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
is_muted(
mutelist, AWS_ACCOUNT_NUMBER, "check_test", "us-east-2", "test", ""
)
)
def test_is_allowlisted_in_region(self):
allowlisted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
def test_is_muted_in_region(self):
muted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
finding_region = AWS_REGION_US_EAST_1
assert is_allowlisted_in_region(allowlisted_regions, finding_region)
assert is_muted_in_region(muted_regions, finding_region)
def test_is_allowlisted_in_region_wildcard(self):
allowlisted_regions = ["*"]
def test_is_muted_in_region_wildcard(self):
muted_regions = ["*"]
finding_region = AWS_REGION_US_EAST_1
assert is_allowlisted_in_region(allowlisted_regions, finding_region)
assert is_muted_in_region(muted_regions, finding_region)
def test_is_not_allowlisted_in_region(self):
allowlisted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
def test_is_not_muted_in_region(self):
muted_regions = [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1]
finding_region = "eu-west-2"
assert not is_allowlisted_in_region(allowlisted_regions, finding_region)
assert not is_muted_in_region(muted_regions, finding_region)
def test_is_allowlisted_in_check(self):
allowlisted_checks = {
def test_is_muted_in_check(self):
muted_checks = {
"check_test": {
"Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1],
"Resources": ["*"],
}
}
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -508,8 +539,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -517,8 +548,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -527,8 +558,8 @@ class Test_Allowlist:
)
assert not (
is_allowlisted_in_check(
allowlisted_checks,
is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"check_test",
"us-east-2",
@@ -537,17 +568,17 @@ class Test_Allowlist:
)
)
def test_is_allowlisted_in_check_regex(self):
# Allowlist example
allowlisted_checks = {
def test_is_muted_in_check_regex(self):
# Mutelist example
muted_checks = {
"s3_*": {
"Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1],
"Resources": ["*"],
}
}
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"s3_bucket_public_access",
AWS_REGION_US_EAST_1,
@@ -555,8 +586,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"s3_bucket_no_mfa_delete",
AWS_REGION_US_EAST_1,
@@ -564,8 +595,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"s3_bucket_policy_public_write_access",
AWS_REGION_US_EAST_1,
@@ -574,8 +605,8 @@ class Test_Allowlist:
)
assert not (
is_allowlisted_in_check(
allowlisted_checks,
is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"iam_user_hardware_mfa_enabled",
AWS_REGION_US_EAST_1,
@@ -584,16 +615,16 @@ class Test_Allowlist:
)
)
def test_is_allowlisted_lambda_generic_check(self):
allowlisted_checks = {
def test_is_muted_lambda_generic_check(self):
muted_checks = {
"lambda_*": {
"Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1],
"Resources": ["*"],
}
}
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_invoke_api_operations_cloudtrail_logging_enabled",
AWS_REGION_US_EAST_1,
@@ -601,8 +632,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_no_secrets_in_code",
AWS_REGION_US_EAST_1,
@@ -610,8 +641,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_no_secrets_in_variables",
AWS_REGION_US_EAST_1,
@@ -619,8 +650,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_not_publicly_accessible",
AWS_REGION_US_EAST_1,
@@ -628,8 +659,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_url_cors_policy",
AWS_REGION_US_EAST_1,
@@ -637,8 +668,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_url_public",
AWS_REGION_US_EAST_1,
@@ -646,8 +677,8 @@ class Test_Allowlist:
"",
)
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_using_supported_runtimes",
AWS_REGION_US_EAST_1,
@@ -655,16 +686,16 @@ class Test_Allowlist:
"",
)
def test_is_allowlisted_lambda_concrete_check(self):
allowlisted_checks = {
def test_is_muted_lambda_concrete_check(self):
muted_checks = {
"lambda_function_no_secrets_in_variables": {
"Regions": [AWS_REGION_US_EAST_1, AWS_REGION_EU_WEST_1],
"Resources": ["*"],
}
}
assert is_allowlisted_in_check(
allowlisted_checks,
assert is_muted_in_check(
muted_checks,
AWS_ACCOUNT_NUMBER,
"awslambda_function_no_secrets_in_variables",
AWS_REGION_US_EAST_1,
@@ -672,9 +703,9 @@ class Test_Allowlist:
"",
)
def test_is_allowlisted_tags(self):
# Allowlist example
allowlist = {
def test_is_muted_tags(self):
# Mutelist example
mutelist = {
"Accounts": {
"*": {
"Checks": {
@@ -688,8 +719,8 @@ class Test_Allowlist:
}
}
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -697,8 +728,8 @@ class Test_Allowlist:
"environment=dev",
)
assert is_allowlisted(
allowlist,
assert is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
AWS_REGION_US_EAST_1,
@@ -707,8 +738,8 @@ class Test_Allowlist:
)
assert not (
is_allowlisted(
allowlist,
is_muted(
mutelist,
AWS_ACCOUNT_NUMBER,
"check_test",
"us-east-2",
@@ -717,49 +748,49 @@ class Test_Allowlist:
)
)
def test_is_allowlisted_in_tags(self):
allowlist_tags = ["environment=dev", "project=prowler"]
def test_is_muted_in_tags(self):
mutelist_tags = ["environment=dev", "project=prowler"]
assert is_allowlisted_in_tags(allowlist_tags, "environment=dev")
assert is_muted_in_tags(mutelist_tags, "environment=dev")
assert is_allowlisted_in_tags(
allowlist_tags,
assert is_muted_in_tags(
mutelist_tags,
"environment=dev | project=prowler",
)
assert not (
is_allowlisted_in_tags(
allowlist_tags,
is_muted_in_tags(
mutelist_tags,
"environment=pro",
)
)
def test_is_allowlisted_in_tags_regex(self):
allowlist_tags = ["environment=(dev|test)", ".*=prowler"]
def test_is_muted_in_tags_regex(self):
mutelist_tags = ["environment=(dev|test)", ".*=prowler"]
assert is_allowlisted_in_tags(
allowlist_tags,
assert is_muted_in_tags(
mutelist_tags,
"environment=test | proj=prowler",
)
assert is_allowlisted_in_tags(
allowlist_tags,
assert is_muted_in_tags(
mutelist_tags,
"env=prod | project=prowler",
)
assert not is_allowlisted_in_tags(
allowlist_tags,
assert not is_muted_in_tags(
mutelist_tags,
"environment=prod | project=myproj",
)
def test_is_allowlisted_in_tags_with_no_tags_in_finding(self):
allowlist_tags = ["environment=(dev|test)", ".*=prowler"]
def test_is_muted_in_tags_with_no_tags_in_finding(self):
mutelist_tags = ["environment=(dev|test)", ".*=prowler"]
finding_tags = ""
assert not is_allowlisted_in_tags(allowlist_tags, finding_tags)
assert not is_muted_in_tags(mutelist_tags, finding_tags)
def test_is_excepted(self):
# Allowlist example
# Mutelist example
exceptions = {
"Accounts": [AWS_ACCOUNT_NUMBER],
"Regions": ["eu-central-1", "eu-south-3"],
@@ -837,10 +868,10 @@ class Test_Allowlist:
"environment=pro",
)
def test_is_allowlisted_in_resource(self):
allowlist_resources = ["prowler", "^test", "prowler-pro"]
def test_is_muted_in_resource(self):
mutelist_resources = ["prowler", "^test", "prowler-pro"]
assert is_allowlisted_in_resource(allowlist_resources, "prowler")
assert is_allowlisted_in_resource(allowlist_resources, "prowler-test")
assert is_allowlisted_in_resource(allowlist_resources, "test-prowler")
assert not is_allowlisted_in_resource(allowlist_resources, "random")
assert is_muted_in_resource(mutelist_resources, "prowler")
assert is_muted_in_resource(mutelist_resources, "prowler-test")
assert is_muted_in_resource(mutelist_resources, "test-prowler")
assert not is_muted_in_resource(mutelist_resources, "random")
@@ -66,11 +66,11 @@ class Test_accessanalyzer_enabled:
assert result[0].region == AWS_REGION_1
assert result[0].resource_tags == []
def test_one_analyzer_not_available_allowlisted(self):
def test_one_analyzer_not_available_muted(self):
# Include analyzers to check
accessanalyzer_client = mock.MagicMock
accessanalyzer_client.region = AWS_REGION_2
accessanalyzer_client.audit_config = {"allowlist_non_default_regions": True}
accessanalyzer_client.audit_config = {"mute_non_default_regions": True}
accessanalyzer_client.analyzers = [
Analyzer(
arn=AWS_ACCOUNT_ARN,
@@ -93,7 +93,7 @@ class Test_accessanalyzer_enabled:
result = check.execute()
assert len(result) == 1
assert result[0].status == "WARNING"
assert result[0].status == "MUTED"
assert (
result[0].status_extended
== f"IAM Access Analyzer in account {AWS_ACCOUNT_NUMBER} is not enabled."
@@ -132,7 +132,7 @@ class Test_config_recorder_all_regions_enabled:
assert recorder.region == AWS_REGION_US_EAST_1
@mock_config
def test_config_one_recorder_disabled_allowlisted(self):
def test_config_one_recorder_disabled_muted(self):
# Create Config Mocked Resources
config_client = client("config", region_name=AWS_REGION_US_EAST_1)
# Create Config Recorder
@@ -165,7 +165,7 @@ class Test_config_recorder_all_regions_enabled:
# Search for the recorder just created
for recorder in result:
if recorder.region == AWS_REGION_US_EAST_1:
assert recorder.status == "WARNING"
assert recorder.status == "MUTED"
assert (
recorder.status_extended
== f"AWS Config recorder {AWS_ACCOUNT_NUMBER} is disabled."
@@ -120,9 +120,9 @@ class Test_drs_job_exist:
assert result[0].region == AWS_REGION
assert result[0].resource_tags == []
def test_drs_disabled_allowlisted(self):
def test_drs_disabled_muted(self):
drs_client = mock.MagicMock
drs_client.audit_config = {"allowlist_non_default_regions": True}
drs_client.audit_config = {"mute_non_default_regions": True}
drs_client.audited_account = AWS_ACCOUNT_NUMBER
drs_client.audited_account_arn = f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
drs_client.region = "eu-west-2"
@@ -147,7 +147,7 @@ class Test_drs_job_exist:
result = check.execute()
assert len(result) == 1
assert result[0].status == "WARNING"
assert result[0].status == "MUTED"
assert result[0].status_extended == "DRS is not enabled for this region."
assert result[0].resource_id == AWS_ACCOUNT_NUMBER
assert result[0].resource_arn == f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"
@@ -137,9 +137,9 @@ class Test_:
assert result[0].resource_arn == DETECTOR_ARN
assert result[0].region == AWS_REGION_EU_WEST_1
def test_guardduty_not_configured_allowlisted(self):
def test_guardduty_not_configured_muted(self):
guardduty_client = mock.MagicMock
guardduty_client.audit_config = {"allowlist_non_default_regions": True}
guardduty_client.audit_config = {"mute_non_default_regions": True}
guardduty_client.region = "eu-south-2"
guardduty_client.detectors = []
guardduty_client.detectors.append(
@@ -160,7 +160,7 @@ class Test_:
check = guardduty_is_enabled()
result = check.execute()
assert len(result) == 1
assert result[0].status == "WARNING"
assert result[0].status == "MUTED"
assert (
result[0].status_extended
== f"GuardDuty detector {DETECTOR_ID} not configured."
@@ -147,9 +147,9 @@ class Test_securityhub_enabled:
)
assert result[0].region == AWS_REGION_EU_WEST_1
def test_securityhub_hub_active_without_integrations_or_standards_allowlisted(self):
def test_securityhub_hub_active_without_integrations_or_standards_muted(self):
securityhub_client = mock.MagicMock
securityhub_client.audit_config = {"allowlist_non_default_regions": True}
securityhub_client.audit_config = {"mute_non_default_regions": True}
securityhub_client.region = AWS_REGION_EU_WEST_1
securityhub_client.securityhubs = [
SecurityHubHub(
@@ -173,7 +173,7 @@ class Test_securityhub_enabled:
check = securityhub_enabled()
result = check.execute()
assert result[0].status == "WARNING"
assert result[0].status == "MUTED"
assert (
result[0].status_extended
== "Security Hub is enabled but without any standard or integration."
+15 -15
View File
@@ -99,10 +99,10 @@ class Test_Common_Output_Options:
arguments.send_sh_only_fails = True
audit_info = self.set_mocked_aws_audit_info()
allowlist_file = ""
mutelist_file = ""
bulk_checks_metadata = {}
output_options = set_provider_output_options(
provider, arguments, audit_info, allowlist_file, bulk_checks_metadata
provider, arguments, audit_info, mutelist_file, bulk_checks_metadata
)
assert isinstance(output_options, Aws_Output_Options)
assert output_options.security_hub_enabled
@@ -110,7 +110,7 @@ class Test_Common_Output_Options:
assert output_options.is_quiet
assert output_options.output_modes == ["html", "csv", "json", "json-asff"]
assert output_options.output_directory == arguments.output_directory
assert output_options.allowlist_file == ""
assert output_options.mutelist_file == ""
assert output_options.bulk_checks_metadata == {}
assert output_options.verbose
assert output_options.output_filename == arguments.output_filename
@@ -132,16 +132,16 @@ class Test_Common_Output_Options:
arguments.unix_timestamp = False
audit_info = self.set_mocked_gcp_audit_info()
allowlist_file = ""
mutelist_file = ""
bulk_checks_metadata = {}
output_options = set_provider_output_options(
provider, arguments, audit_info, allowlist_file, bulk_checks_metadata
provider, arguments, audit_info, mutelist_file, bulk_checks_metadata
)
assert isinstance(output_options, Gcp_Output_Options)
assert output_options.is_quiet
assert output_options.output_modes == ["html", "csv", "json"]
assert output_options.output_directory == arguments.output_directory
assert output_options.allowlist_file == ""
assert output_options.mutelist_file == ""
assert output_options.bulk_checks_metadata == {}
assert output_options.verbose
assert output_options.output_filename == arguments.output_filename
@@ -167,10 +167,10 @@ class Test_Common_Output_Options:
# Mock AWS Audit Info
audit_info = self.set_mocked_aws_audit_info()
allowlist_file = ""
mutelist_file = ""
bulk_checks_metadata = {}
output_options = set_provider_output_options(
provider, arguments, audit_info, allowlist_file, bulk_checks_metadata
provider, arguments, audit_info, mutelist_file, bulk_checks_metadata
)
assert isinstance(output_options, Aws_Output_Options)
assert output_options.security_hub_enabled
@@ -178,7 +178,7 @@ class Test_Common_Output_Options:
assert output_options.is_quiet
assert output_options.output_modes == ["html", "csv", "json", "json-asff"]
assert output_options.output_directory == arguments.output_directory
assert output_options.allowlist_file == ""
assert output_options.mutelist_file == ""
assert output_options.bulk_checks_metadata == {}
assert output_options.verbose
assert (
@@ -205,10 +205,10 @@ class Test_Common_Output_Options:
audit_info = self.set_mocked_azure_audit_info()
audit_info.identity.domain = "test-domain"
allowlist_file = ""
mutelist_file = ""
bulk_checks_metadata = {}
output_options = set_provider_output_options(
provider, arguments, audit_info, allowlist_file, bulk_checks_metadata
provider, arguments, audit_info, mutelist_file, bulk_checks_metadata
)
assert isinstance(output_options, Azure_Output_Options)
assert output_options.is_quiet
@@ -218,7 +218,7 @@ class Test_Common_Output_Options:
"json",
]
assert output_options.output_directory == arguments.output_directory
assert output_options.allowlist_file == ""
assert output_options.mutelist_file == ""
assert output_options.bulk_checks_metadata == {}
assert output_options.verbose
assert (
@@ -246,10 +246,10 @@ class Test_Common_Output_Options:
tenants = ["tenant-1", "tenant-2"]
audit_info.identity.tenant_ids = tenants
allowlist_file = ""
mutelist_file = ""
bulk_checks_metadata = {}
output_options = set_provider_output_options(
provider, arguments, audit_info, allowlist_file, bulk_checks_metadata
provider, arguments, audit_info, mutelist_file, bulk_checks_metadata
)
assert isinstance(output_options, Azure_Output_Options)
assert output_options.is_quiet
@@ -259,7 +259,7 @@ class Test_Common_Output_Options:
"json",
]
assert output_options.output_directory == arguments.output_directory
assert output_options.allowlist_file == ""
assert output_options.mutelist_file == ""
assert output_options.bulk_checks_metadata == {}
assert output_options.verbose
assert (