ben.carrasco/prowler
1
0 Fork 0
mirror of https://github.com/prowler-cloud/prowler.git synced 2026-10-12 22:44:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore(aws): migrate fixers to new structure until ec2_ebs

Browse Source
This commit is contained in:
Daniel Barranquero committed 2025-06-11 14:21:43 +02:00
1 parent 09b33d05a3
commit 07c5ae547f
23 files changed
+960 -631
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

prowler/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer.py
+60 -33
View File
@@ -1,42 +1,69 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.accessanalyzer.accessanalyzer_client import (
accessanalyzer_client,
)
def fixer(region):
class AccessAnalyzerEnabledFixer(AWSFixer):
"""
Enable Access Analyzer in a region. Requires the access-analyzer:CreateAnalyzer permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "access-analyzer:CreateAnalyzer",
"Resource": "*"
}
]
}
Args:
region (str): AWS region
Returns:
bool: True if Access Analyzer is enabled, False otherwise
Fixer for enabling Access Analyzer in a region.
"""
try:
regional_client = accessanalyzer_client.regional_clients[region]
regional_client.create_analyzer(
analyzerName=accessanalyzer_client.fixer_config.get(
"accessanalyzer_enabled", {}
).get("AnalyzerName", "DefaultAnalyzer"),
type=accessanalyzer_client.fixer_config.get(
"accessanalyzer_enabled", {}
).get("AnalyzerType", "ACCOUNT_UNUSED_ACCESS"),
def __init__(self):
super().__init__(
description="Enable Access Analyzer in a region",
cost_impact=False,
cost_description=None,
service="accessanalyzer",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "access-analyzer:CreateAnalyzer",
"Resource": "*",
}
],
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Enable Access Analyzer in a region.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region (if finding is not provided)
Returns:
bool: True if Access Analyzer is enabled, False otherwise
"""
try:
if finding:
region = finding.region
else:
region = kwargs.get("region")
if not region:
raise ValueError("Region is required")
# Show the fixing message
super().fix(region=region)
regional_client = accessanalyzer_client.regional_clients[region]
regional_client.create_analyzer(
analyzerName=accessanalyzer_client.fixer_config.get(
"accessanalyzer_enabled", {}
).get("AnalyzerName", "DefaultAnalyzer"),
type=accessanalyzer_client.fixer_config.get(
"accessanalyzer_enabled", {}
).get("AnalyzerType", "ACCOUNT_UNUSED_ACCESS"),
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer.py
+73 -51
View File
@@ -1,61 +1,83 @@
import json
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.awslambda.awslambda_client import awslambda_client
def fixer(resource_id: str, region: str) -> bool:
class LambdaFunctionNotPubliclyAccessibleFixer(AWSFixer):
"""
Remove the Lambda function's resource-based policy to prevent public access and add a new permission for the account.
Specifically, this fixer deletes all permission statements associated with the Lambda function's policy and then adds a new permission.
Requires the lambda:RemovePermission and lambda:AddPermission permissions.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "lambda:RemovePermission",
"Resource": "*"
Fixer for removing public access from Lambda function policies.
"""
def __init__(self):
super().__init__(
description="Remove public access from Lambda function's resource-based policy and add a new permission for the account.",
cost_impact=False,
cost_description=None,
service="awslambda",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "lambda:RemovePermission",
"Resource": "*",
},
{
"Effect": "Allow",
"Action": "lambda:AddPermission",
"Resource": "*",
},
],
},
{
"Effect": "Allow",
"Action": "lambda:AddPermission",
"Resource": "*"
}
]
}
Args:
resource_id (str): The Lambda function name or ARN.
region (str): AWS region where the Lambda function exists.
Returns:
bool: True if the operation is successful (policy removed and permission added), False otherwise.
"""
try:
account_id = awslambda_client.audited_account
regional_client = awslambda_client.regional_clients[region]
policy_response = regional_client.get_policy(FunctionName=resource_id)
policy = json.loads(policy_response.get("Policy"))
for statement in policy.get("Statement", []):
statement_id = statement.get("Sid")
if statement_id:
regional_client.remove_permission(
FunctionName=resource_id, StatementId=statement_id
)
regional_client.add_permission(
FunctionName=resource_id,
StatementId="ProwlerFixerStatement",
Principal=account_id,
Action="lambda:InvokeFunction",
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Remove the Lambda function's resource-based policy to prevent public access and add a new permission for the account.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region, resource_id (if finding is not provided)
Returns:
bool: True if the operation is successful, False otherwise.
"""
try:
if finding:
region = finding.region
resource_id = finding.resource_id
else:
region = kwargs.get("region")
resource_id = kwargs.get("resource_id")
if not region or not resource_id:
raise ValueError("Region and resource_id are required")
super().fix(region=region, resource_id=resource_id)
account_id = awslambda_client.audited_account
regional_client = awslambda_client.regional_clients[region]
policy_response = regional_client.get_policy(FunctionName=resource_id)
policy = json.loads(policy_response.get("Policy"))
for statement in policy.get("Statement", []):
statement_id = statement.get("Sid")
if statement_id:
regional_client.remove_permission(
FunctionName=resource_id, StatementId=statement_id
)
regional_client.add_permission(
FunctionName=resource_id,
StatementId="ProwlerFixerStatement",
Principal=account_id,
Action="lambda:InvokeFunction",
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py
+64 -41
View File
@@ -1,52 +1,75 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
cloudtrail_client,
)
from prowler.providers.aws.services.s3.s3_client import s3_client
def fixer(resource_id: str, region: str) -> bool:
class CloudtrailLogsS3BucketIsNotPubliclyAccessibleFixer(AWSFixer):
"""
Modify the CloudTrail's associated S3 bucket's public access settings to ensure the bucket is not publicly accessible.
Specifically, this fixer configures the S3 bucket's public access block settings to block all public access.
Requires the s3:PutBucketPublicAccessBlock permissions.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "s3:PutBucketPublicAccessBlock",
"Resource": "*"
}
]
}
Args:
resource_id (str): The CloudTrail name.
region (str): AWS region where the CloudTrail and S3 bucket exist.
Returns:
bool: True if the operation is successful (policy and ACL updated), False otherwise.
Fixer for ensuring CloudTrail's associated S3 bucket is not publicly accessible.
"""
try:
regional_client = s3_client.regional_clients[region]
for trail in cloudtrail_client.trails.values():
if trail.name == resource_id:
trail_bucket = trail.s3_bucket
regional_client.put_public_access_block(
Bucket=trail_bucket,
PublicAccessBlockConfiguration={
"BlockPublicAcls": True,
"IgnorePublicAcls": True,
"BlockPublicPolicy": True,
"RestrictPublicBuckets": True,
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
def __init__(self):
super().__init__(
description="Modify the CloudTrail's associated S3 bucket's public access settings to ensure the bucket is not publicly accessible.",
cost_impact=False,
cost_description=None,
service="cloudtrail",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "s3:PutBucketPublicAccessBlock",
"Resource": "*",
}
],
},
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Modify the CloudTrail's associated S3 bucket's public access settings to ensure the bucket is not publicly accessible.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region, resource_id (if finding is not provided)
Returns:
bool: True if the operation is successful, False otherwise.
"""
try:
if finding:
region = finding.region
resource_id = finding.resource_id
else:
region = kwargs.get("region")
resource_id = kwargs.get("resource_id")
if not region or not resource_id:
raise ValueError("Region and resource_id are required")
super().fix(region=region, resource_id=resource_id)
regional_client = s3_client.regional_clients[region]
for trail in cloudtrail_client.trails.values():
if trail.name == resource_id:
trail_bucket = trail.s3_bucket
regional_client.put_public_access_block(
Bucket=trail_bucket,
PublicAccessBlockConfiguration={
"BlockPublicAcls": True,
"IgnorePublicAcls": True,
"BlockPublicPolicy": True,
"RestrictPublicBuckets": True,
},
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer.py
+75 -50
View File
@@ -1,59 +1,84 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.cloudtrail.cloudtrail_client import (
cloudtrail_client,
)
def fixer(region):
class CloudtrailMultiRegionEnabledFixer(AWSFixer):
"""
NOTE: Define the S3 bucket name in the fixer_config.yaml file.
Enable CloudTrail in a region. Requires the cloudtrail:CreateTrail permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "cloudtrail:CreateTrail",
"Resource": "*"
Fixer for enabling CloudTrail as a multi-region trail in a region.
"""
def __init__(self):
super().__init__(
description="Enable CloudTrail as a multi-region trail in a region.",
cost_impact=False,
cost_description=None,
service="cloudtrail",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "cloudtrail:CreateTrail",
"Resource": "*",
}
],
},
)
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Enable CloudTrail as a multi-region trail in a region.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region (if finding is not provided)
Returns:
bool: True if CloudTrail is enabled, False otherwise
"""
try:
if finding:
region = finding.region
else:
region = kwargs.get("region")
if not region:
raise ValueError("Region is required")
super().fix(region=region)
cloudtrail_fixer_config = cloudtrail_client.fixer_config.get(
"cloudtrail_multi_region_enabled", {}
)
regional_client = cloudtrail_client.regional_clients[region]
args = {
"Name": cloudtrail_fixer_config.get("TrailName", "DefaultTrail"),
"S3BucketName": cloudtrail_fixer_config.get("S3BucketName"),
"IsMultiRegionTrail": cloudtrail_fixer_config.get(
"IsMultiRegionTrail", True
),
"EnableLogFileValidation": cloudtrail_fixer_config.get(
"EnableLogFileValidation", True
),
}
]
}
Args:
region (str): AWS region
Returns:
bool: True if CloudTrail is enabled, False otherwise
"""
try:
cloudtrail_fixer_config = cloudtrail_client.fixer_config.get(
"cloudtrail_multi_region_enabled", {}
)
regional_client = cloudtrail_client.regional_clients[region]
args = {
"Name": cloudtrail_fixer_config.get("TrailName", "DefaultTrail"),
"S3BucketName": cloudtrail_fixer_config.get("S3BucketName"),
"IsMultiRegionTrail": cloudtrail_fixer_config.get(
"IsMultiRegionTrail", True
),
"EnableLogFileValidation": cloudtrail_fixer_config.get(
"EnableLogFileValidation", True
),
}
if cloudtrail_fixer_config.get("CloudWatchLogsLogGroupArn"):
args["CloudWatchLogsLogGroupArn"] = cloudtrail_fixer_config.get(
"CloudWatchLogsLogGroupArn"
if cloudtrail_fixer_config.get("CloudWatchLogsLogGroupArn"):
args["CloudWatchLogsLogGroupArn"] = cloudtrail_fixer_config.get(
"CloudWatchLogsLogGroupArn"
)
if cloudtrail_fixer_config.get("CloudWatchLogsRoleArn"):
args["CloudWatchLogsRoleArn"] = cloudtrail_fixer_config.get(
"CloudWatchLogsRoleArn"
)
if cloudtrail_fixer_config.get("KmsKeyId"):
args["KmsKeyId"] = cloudtrail_fixer_config.get("KmsKeyId")
regional_client.create_trail(**args)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
if cloudtrail_fixer_config.get("CloudWatchLogsRoleArn"):
args["CloudWatchLogsRoleArn"] = cloudtrail_fixer_config.get(
"CloudWatchLogsRoleArn"
)
if cloudtrail_fixer_config.get("KmsKeyId"):
args["KmsKeyId"] = cloudtrail_fixer_config.get("KmsKeyId")
regional_client.create_trail(**args)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
return False
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_fixer.py
+84 -61
View File
@@ -1,71 +1,94 @@
import json
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.iam.iam_client import iam_client
def fixer(resource_arn: str) -> bool:
class CloudtrailThreatDetectionEnumerationFixer(AWSFixer):
"""
Fixer for restricting access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
"""
Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
Requires the following permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:PutUserPolicy",
"iam:PutRolePolicy",
def __init__(self):
super().__init__(
description="Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.",
cost_impact=False,
cost_description=None,
service="cloudtrail",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:PutUserPolicy",
"iam:PutRolePolicy",
],
"Resource": "*",
}
],
"Resource": "*"
}
]
}
Args:
resource_arn (str): The ARN of the compromised AWS entity (IAM User or Role).
Returns:
bool: True if the fix was applied successfully, False otherwise.
"""
try:
if ":user/" in resource_arn:
entity_type = "user"
entity_name = resource_arn.split("/")[-1]
elif ":role/" in resource_arn:
entity_type = "role"
entity_name = resource_arn.split("/")[-1]
else:
return False
deny_policy = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
policy_name = "DenyAllAccess"
if entity_type == "user":
iam_client.client.put_user_policy(
UserName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to user {entity_name}")
elif entity_type == "role":
iam_client.client.put_role_policy(
RoleName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to role {entity_name}")
return True
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
},
)
return False
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix (should have resource_arn)
**kwargs: resource_arn (if finding is not provided)
Returns:
bool: True if the fix was applied successfully, False otherwise.
"""
try:
if finding:
resource_arn = getattr(finding, "resource_arn", None)
else:
resource_arn = kwargs.get("resource_arn")
if not resource_arn:
raise ValueError("resource_arn is required")
super().fix(resource_arn=resource_arn)
if ":user/" in resource_arn:
entity_type = "user"
entity_name = resource_arn.split("/")[-1]
elif ":role/" in resource_arn:
entity_type = "role"
entity_name = resource_arn.split("/")[-1]
else:
return False
deny_policy = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
policy_name = "DenyAllAccess"
if entity_type == "user":
iam_client.client.put_user_policy(
UserName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to user {entity_name}")
elif entity_type == "role":
iam_client.client.put_role_policy(
RoleName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to role {entity_name}")
return True
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_fixer.py
+84 -61
View File
@@ -1,71 +1,94 @@
import json
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.iam.iam_client import iam_client
def fixer(resource_arn: str) -> bool:
class CloudtrailThreatDetectionLlmJackingFixer(AWSFixer):
"""
Fixer for restricting access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
"""
Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
Requires the following permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:PutUserPolicy",
"iam:PutRolePolicy",
def __init__(self):
super().__init__(
description="Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.",
cost_impact=False,
cost_description=None,
service="cloudtrail",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:PutUserPolicy",
"iam:PutRolePolicy",
],
"Resource": "*",
}
],
"Resource": "*"
}
]
}
Args:
resource_arn (str): The ARN of the compromised AWS entity (IAM User or Role).
Returns:
bool: True if the fix was applied successfully, False otherwise.
"""
try:
if ":user/" in resource_arn:
entity_type = "user"
entity_name = resource_arn.split("/")[-1]
elif ":role/" in resource_arn:
entity_type = "role"
entity_name = resource_arn.split("/")[-1]
else:
return False
deny_policy = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
policy_name = "DenyAllAccess"
if entity_type == "user":
iam_client.client.put_user_policy(
UserName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to user {entity_name}")
elif entity_type == "role":
iam_client.client.put_role_policy(
RoleName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to role {entity_name}")
return True
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
},
)
return False
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix (should have resource_arn)
**kwargs: resource_arn (if finding is not provided)
Returns:
bool: True if the fix was applied successfully, False otherwise.
"""
try:
if finding:
resource_arn = getattr(finding, "resource_arn", None)
else:
resource_arn = kwargs.get("resource_arn")
if not resource_arn:
raise ValueError("resource_arn is required")
super().fix(resource_arn=resource_arn)
if ":user/" in resource_arn:
entity_type = "user"
entity_name = resource_arn.split("/")[-1]
elif ":role/" in resource_arn:
entity_type = "role"
entity_name = resource_arn.split("/")[-1]
else:
return False
deny_policy = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
policy_name = "DenyAllAccess"
if entity_type == "user":
iam_client.client.put_user_policy(
UserName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to user {entity_name}")
elif entity_type == "role":
iam_client.client.put_role_policy(
RoleName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to role {entity_name}")
return True
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_fixer.py
+84 -61
View File
@@ -1,71 +1,94 @@
import json
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.iam.iam_client import iam_client
def fixer(resource_arn: str) -> bool:
class CloudtrailThreatDetectionPrivilegeEscalationFixer(AWSFixer):
"""
Fixer for restricting access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
"""
Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
Requires the following permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:PutUserPolicy",
"iam:PutRolePolicy",
def __init__(self):
super().__init__(
description="Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.",
cost_impact=False,
cost_description=None,
service="cloudtrail",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": [
"iam:PutUserPolicy",
"iam:PutRolePolicy",
],
"Resource": "*",
}
],
"Resource": "*"
}
]
}
Args:
resource_arn (str): The ARN of the compromised AWS entity (IAM User or Role).
Returns:
bool: True if the fix was applied successfully, False otherwise.
"""
try:
if ":user/" in resource_arn:
entity_type = "user"
entity_name = resource_arn.split("/")[-1]
elif ":role/" in resource_arn:
entity_type = "role"
entity_name = resource_arn.split("/")[-1]
else:
return False
deny_policy = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
policy_name = "DenyAllAccess"
if entity_type == "user":
iam_client.client.put_user_policy(
UserName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to user {entity_name}")
elif entity_type == "role":
iam_client.client.put_role_policy(
RoleName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to role {entity_name}")
return True
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
},
)
return False
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Restricts access to a compromised AWS entity by attaching a deny-all inline policy to the user or role.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix (should have resource_arn)
**kwargs: resource_arn (if finding is not provided)
Returns:
bool: True if the fix was applied successfully, False otherwise.
"""
try:
if finding:
resource_arn = getattr(finding, "resource_arn", None)
else:
resource_arn = kwargs.get("resource_arn")
if not resource_arn:
raise ValueError("resource_arn is required")
super().fix(resource_arn=resource_arn)
if ":user/" in resource_arn:
entity_type = "user"
entity_name = resource_arn.split("/")[-1]
elif ":role/" in resource_arn:
entity_type = "role"
entity_name = resource_arn.split("/")[-1]
else:
return False
deny_policy = {
"Version": "2012-10-17",
"Statement": [{"Effect": "Deny", "Action": "*", "Resource": "*"}],
}
policy_name = "DenyAllAccess"
if entity_type == "user":
iam_client.client.put_user_policy(
UserName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to user {entity_name}")
elif entity_type == "role":
iam_client.client.put_role_policy(
RoleName=entity_name,
PolicyName=policy_name,
PolicyDocument=json.dumps(deny_policy),
)
logger.info(f"Applied Deny policy to role {entity_name}")
return True
except Exception as error:
logger.error(
f"{error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer.py
+71 -49
View File
@@ -1,59 +1,81 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.codeartifact.codeartifact_client import (
codeartifact_client,
)
def fixer(resource_id: str, region: str) -> bool:
class CodeartifactPackagesExternalPublicPublishingDisabledFixer(AWSFixer):
"""
Modify the CodeArtifact package's configuration to restrict public access.
Specifically, this fixer changes the package's configuration to block public access by
setting restrictions on the "publish" and "upstream" actions.
Requires the codeartifact:PutPackageOriginConfiguration permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "codeartifact:PutPackageOriginConfiguration",
"Resource": "*"
}
]
}
Args:
resource_id (str): The CodeArtifact package name in the format "domain_name/package_name".
region (str): AWS region where the CodeArtifact package exists.
Returns:
bool: True if the operation is successful (configuration updated), False otherwise.
Fixer for modifying the CodeArtifact package's configuration to restrict public access.
"""
try:
domain_name, package_name = resource_id.split("/")
regional_client = codeartifact_client.regional_clients[region]
for repository in codeartifact_client.repositories.values():
if repository.domain_name == domain_name:
for package in repository.packages:
if package.name == package_name:
publish_value = (
package.origin_configuration.restrictions.publish.value
)
regional_client.put_package_origin_configuration(
domain=domain_name,
repository=repository.name,
format=package.format,
package=package_name,
restrictions={
"publish": publish_value,
"upstream": "BLOCK",
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
def __init__(self):
super().__init__(
description="Modify the CodeArtifact package's configuration to restrict public access.",
cost_impact=False,
cost_description=None,
service="codeartifact",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "codeartifact:PutPackageOriginConfiguration",
"Resource": "*",
}
],
},
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Modify the CodeArtifact package's configuration to restrict public access.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region, resource_id (if finding is not provided)
Returns:
bool: True if the operation is successful (configuration updated), False otherwise.
"""
try:
if finding:
region = finding.region
resource_id = finding.resource_id
else:
region = kwargs.get("region")
resource_id = kwargs.get("resource_id")
if not region or not resource_id:
raise ValueError("Region and resource_id are required")
super().fix(region=region, resource_id=resource_id)
domain_name, package_name = resource_id.split("/")
regional_client = codeartifact_client.regional_clients[region]
for repository in codeartifact_client.repositories.values():
if repository.domain_name == domain_name:
for package in repository.packages:
if package.name == package_name:
publish_value = (
package.origin_configuration.restrictions.publish.value
)
regional_client.put_package_origin_configuration(
domain=domain_name,
repository=repository.name,
format=package.format,
package=package_name,
restrictions={
"publish": publish_value,
"upstream": "BLOCK",
},
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer.py
+58 -33
View File
@@ -1,42 +1,67 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.documentdb.documentdb_client import (
documentdb_client,
)
def fixer(resource_id: str, region: str) -> bool:
class DocumentdbClusterPublicSnapshotFixer(AWSFixer):
"""
Modify the attributes of a DocumentDB cluster snapshot to remove public access.
Specifically, this fixer removes the 'all' value from the 'restore' attribute to
prevent the snapshot from being publicly accessible. Requires the rds:ModifyDBClusterSnapshotAttribute permissions.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "rds:ModifyDBClusterSnapshotAttribute",
"Resource": "*"
}
]
}
Args:
resource_id (str): The DB cluster snapshot identifier.
region (str): AWS region where the snapshot exists.
Returns:
bool: True if the operation is successful (public access is removed), False otherwise.
Fixer for modifying the attributes of a DocumentDB cluster snapshot to remove public access.
"""
try:
regional_client = documentdb_client.regional_clients[region]
regional_client.modify_db_cluster_snapshot_attribute(
DBClusterSnapshotIdentifier=resource_id,
AttributeName="restore",
ValuesToRemove=["all"],
def __init__(self):
super().__init__(
description="Modify the attributes of a DocumentDB cluster snapshot to remove public access.",
cost_impact=False,
cost_description=None,
service="documentdb",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "rds:ModifyDBClusterSnapshotAttribute",
"Resource": "*",
}
],
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Modify the attributes of a DocumentDB cluster snapshot to remove public access.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region, resource_id (if finding is not provided)
Returns:
bool: True if the operation is successful (public access is removed), False otherwise.
"""
try:
if finding:
region = finding.region
resource_id = finding.resource_id
else:
region = kwargs.get("region")
resource_id = kwargs.get("resource_id")
if not region or not resource_id:
raise ValueError("Region and resource_id are required")
super().fix(region=region, resource_id=resource_id)
regional_client = documentdb_client.regional_clients[region]
regional_client.modify_db_cluster_snapshot_attribute(
DBClusterSnapshotIdentifier=resource_id,
AttributeName="restore",
ValuesToRemove=["all"],
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer.py
+57 -33
View File
@@ -1,40 +1,64 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
def fixer(resource_id: str, region: str) -> bool:
class Ec2AmiPublicFixer(AWSFixer):
"""
Modify the attributes of an EC2 AMI to remove public access.
Specifically, this fixer removes the 'all' value from the 'LaunchPermission' attribute
to prevent the AMI from being publicly accessible.
Requires the ec2:ModifyImageAttribute permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:ModifyImageAttribute",
"Resource": "*"
}
]
}
Args:
resource_id (str): The ID of the EC2 AMI to make private.
region (str): AWS region where the AMI exists.
Returns:
bool: True if the operation is successful (the AMI is no longer publicly accessible), False otherwise.
Fixer for modifying the attributes of an EC2 AMI to remove public access.
"""
try:
regional_client = ec2_client.regional_clients[region]
regional_client.modify_image_attribute(
ImageId=resource_id,
LaunchPermission={"Remove": [{"Group": "all"}]},
def __init__(self):
super().__init__(
description="Modify the attributes of an EC2 AMI to remove public access.",
cost_impact=False,
cost_description=None,
service="ec2",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:ModifyImageAttribute",
"Resource": "*",
}
],
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Modify the attributes of an EC2 AMI to remove public access.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region, resource_id (if finding is not provided)
Returns:
bool: True if the operation is successful (the AMI is no longer publicly accessible), False otherwise.
"""
try:
if finding:
region = finding.region
resource_id = finding.resource_id
else:
region = kwargs.get("region")
resource_id = kwargs.get("resource_id")
if not region or not resource_id:
raise ValueError("Region and resource_id are required")
super().fix(region=region, resource_id=resource_id)
regional_client = ec2_client.regional_clients[region]
regional_client.modify_image_attribute(
ImageId=resource_id,
LaunchPermission={"Remove": [{"Group": "all"}]},
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer.py
+53 -27
View File
@@ -1,34 +1,60 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
def fixer(region):
class Ec2EbsDefaultEncryptionFixer(AWSFixer):
"""
Enable EBS encryption by default in a region. NOTE: Custom KMS keys for EBS Default Encryption may be overwritten.
Requires the ec2:EnableEbsEncryptionByDefault permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:EnableEbsEncryptionByDefault",
"Resource": "*"
}
]
}
Args:
region (str): AWS region
Returns:
bool: True if EBS encryption by default is enabled, False otherwise
Fixer to enable EBS encryption by default in a region.
"""
try:
regional_client = ec2_client.regional_clients[region]
return regional_client.enable_ebs_encryption_by_default()[
"EbsEncryptionByDefault"
]
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
def __init__(self):
super().__init__(
description="Enable EBS encryption by default in a region. NOTE: Custom KMS keys for EBS Default Encryption may be overwritten.",
cost_impact=False,
cost_description=None,
service="ec2",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:EnableEbsEncryptionByDefault",
"Resource": "*",
}
],
},
)
return False
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Enable EBS encryption by default in a region.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region (if finding is not provided)
Returns:
bool: True if EBS encryption by default is enabled, False otherwise
"""
try:
if finding:
region = finding.region
else:
region = kwargs.get("region")
if not region:
raise ValueError("Region is required")
super().fix(region=region)
regional_client = ec2_client.regional_clients[region]
return regional_client.enable_ebs_encryption_by_default()[
"EbsEncryptionByDefault"
]
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer.py
+59 -34
View File
@@ -1,41 +1,66 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
def fixer(resource_id: str, region: str) -> bool:
class Ec2EbsPublicSnapshotFixer(AWSFixer):
"""
Modify the attributes of an EBS snapshot to remove public access.
Specifically, this fixer removes the 'all' value from the 'createVolumePermission' attribute to
prevent the snapshot from being publicly accessible. Requires the ec2:ModifySnapshotAttribute permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:ModifySnapshotAttribute",
"Resource": "*"
}
]
}
Args:
resource_id (str): The snapshot identifier.
region (str): AWS region where the snapshot exists.
Returns:
bool: True if the operation is successful (public access is removed), False otherwise.
Fixer for modifying the attributes of an EBS snapshot to remove public access.
"""
try:
regional_client = ec2_client.regional_clients[region]
regional_client.modify_snapshot_attribute(
SnapshotId=resource_id,
Attribute="createVolumePermission",
OperationType="remove",
GroupNames=["all"],
def __init__(self):
super().__init__(
description="Modify the attributes of an EBS snapshot to remove public access.",
cost_impact=False,
cost_description=None,
service="ec2",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:ModifySnapshotAttribute",
"Resource": "*",
}
],
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
else:
return True
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Modify the attributes of an EBS snapshot to remove public access.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region, resource_id (if finding is not provided)
Returns:
bool: True if the operation is successful (public access is removed), False otherwise.
"""
try:
if finding:
region = finding.region
resource_id = finding.resource_id
else:
region = kwargs.get("region")
resource_id = kwargs.get("resource_id")
if not region or not resource_id:
raise ValueError("Region and resource_id are required")
super().fix(region=region, resource_id=resource_id)
regional_client = ec2_client.regional_clients[region]
regional_client.modify_snapshot_attribute(
SnapshotId=resource_id,
Attribute="createVolumePermission",
OperationType="remove",
GroupNames=["all"],
)
return True
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
prowler/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer.py
+59 -31
View File
@@ -1,38 +1,66 @@
from typing import Optional
from prowler.lib.check.models import Check_Report_AWS
from prowler.lib.logger import logger
from prowler.providers.aws.lib.fix.fixer import AWSFixer
from prowler.providers.aws.services.ec2.ec2_client import ec2_client
def fixer(region):
class Ec2EbsSnapshotAccountBlockPublicAccessFixer(AWSFixer):
"""
Enable EBS snapshot block public access in a region.
Requires the ec2:EnableSnapshotBlockPublicAccess permission.
Permissions:
{
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:EnableSnapshotBlockPublicAccess",
"Resource": "*"
}
]
}
Args:
region (str): AWS region
Returns:
bool: True if EBS snapshot block public access is enabled, False otherwise
Fixer to enable EBS snapshot block public access in a region.
"""
try:
regional_client = ec2_client.regional_clients[region]
state = ec2_client.fixer_config.get(
"ec2_ebs_snapshot_account_block_public_access", {}
).get("State", "block-all-sharing")
return (
regional_client.enable_snapshot_block_public_access(State=state)["State"]
== state
def __init__(self):
super().__init__(
description="Enable EBS snapshot block public access in a region.",
cost_impact=False,
cost_description=None,
service="ec2",
iam_policy_required={
"Version": "2012-10-17",
"Statement": [
{
"Effect": "Allow",
"Action": "ec2:EnableSnapshotBlockPublicAccess",
"Resource": "*",
}
],
},
)
except Exception as error:
logger.error(
f"{region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
def fix(self, finding: Optional[Check_Report_AWS] = None, **kwargs) -> bool:
"""
Enable EBS snapshot block public access in a region.
Args:
finding (Optional[Check_Report_AWS]): Finding to fix
**kwargs: region (if finding is not provided)
Returns:
bool: True if EBS snapshot block public access is enabled, False otherwise
"""
try:
if finding:
region = finding.region
else:
region = kwargs.get("region")
if not region:
raise ValueError("Region is required")
super().fix(region=region)
regional_client = ec2_client.regional_clients[region]
state = ec2_client.fixer_config.get(
"ec2_ebs_snapshot_account_block_public_access", {}
).get("State", "block-all-sharing")
return (
regional_client.enable_snapshot_block_public_access(State=state)[
"State"
]
== state
)
except Exception as error:
logger.error(
f"{region if 'region' in locals() else 'unknown'} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
)
return False
tests/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer_test.py
+3 -3
View File
@@ -3,7 +3,7 @@ from unittest import mock
from tests.providers.aws.utils import AWS_ACCOUNT_ARN, AWS_REGION_EU_WEST_1
class Test_accessanalyzer_enabled_fixer:
class TestAccessAnalyzerEnabledFixer:
def test_accessanalyzer_enabled_fixer(self):
regional_client = mock.MagicMock()
accessanalyzer_client = mock.MagicMock()
@@ -26,7 +26,7 @@ class Test_accessanalyzer_enabled_fixer:
):
# Test Check
from prowler.providers.aws.services.accessanalyzer.accessanalyzer_enabled.accessanalyzer_enabled_fixer import (
fixer,
AccessAnalyzerEnabledFixer,
)
assert fixer(AWS_REGION_EU_WEST_1)
assert AccessAnalyzerEnabledFixer().fix(region=AWS_REGION_EU_WEST_1)
tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer_test.py
+13 -13
View File
@@ -11,7 +11,7 @@ from tests.providers.aws.utils import (
)
class Test_awslambda_function_not_publicly_accessible_fixer:
class TestLambdaFunctionNotPubliclyAccessibleFixer:
@mock_aws
def test_function_public(self):
# Create the mock IAM role
@@ -71,12 +71,12 @@ class Test_awslambda_function_not_publicly_accessible_fixer:
new=Lambda(aws_provider),
),
):
# Test Fixer
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible_fixer import (
fixer,
LambdaFunctionNotPubliclyAccessibleFixer,
)
assert fixer(function_name, AWS_REGION_EU_WEST_1)
fixer = LambdaFunctionNotPubliclyAccessibleFixer()
assert fixer.fix(region=AWS_REGION_EU_WEST_1, resource_id=function_name)
@mock_aws
def test_function_public_with_source_account(self):
@@ -139,12 +139,12 @@ class Test_awslambda_function_not_publicly_accessible_fixer:
new=Lambda(aws_provider),
),
):
# Test Fixer
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible_fixer import (
fixer,
LambdaFunctionNotPubliclyAccessibleFixer,
)
assert fixer(function_name, AWS_REGION_EU_WEST_1)
fixer = LambdaFunctionNotPubliclyAccessibleFixer()
assert fixer.fix(region=AWS_REGION_EU_WEST_1, resource_id=function_name)
@mock_aws
def test_function_not_public(self):
@@ -206,12 +206,12 @@ class Test_awslambda_function_not_publicly_accessible_fixer:
new=Lambda(aws_provider),
),
):
# Test Fixer
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible_fixer import (
fixer,
LambdaFunctionNotPubliclyAccessibleFixer,
)
assert fixer(function_name, AWS_REGION_EU_WEST_1)
fixer = LambdaFunctionNotPubliclyAccessibleFixer()
assert fixer.fix(region=AWS_REGION_EU_WEST_1, resource_id=function_name)
@mock_aws
def test_function_public_error(self):
@@ -272,9 +272,9 @@ class Test_awslambda_function_not_publicly_accessible_fixer:
new=Lambda(aws_provider),
),
):
# Test Fixer
from prowler.providers.aws.services.awslambda.awslambda_function_not_publicly_accessible.awslambda_function_not_publicly_accessible_fixer import (
fixer,
LambdaFunctionNotPubliclyAccessibleFixer,
)
assert not fixer("non-exsting", AWS_REGION_EU_WEST_1)
fixer = LambdaFunctionNotPubliclyAccessibleFixer()
assert not fixer.fix(region=AWS_REGION_EU_WEST_1, resource_id="non-exsting")
tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py
+9 -7
View File
@@ -28,7 +28,7 @@ def mock_make_api_call_error(self, operation_name, kwarg):
return mock_make_api_call(self, operation_name, kwarg)
class Test_cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer:
class TestCloudtrailLogsS3BucketIsNotPubliclyAccessibleFixer:
@mock_aws
def test_trail_bucket_public_acl(self):
aws_provider = set_mocked_aws_provider(
@@ -81,12 +81,12 @@ class Test_cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer:
new=S3(aws_provider),
),
):
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer import (
fixer,
CloudtrailLogsS3BucketIsNotPubliclyAccessibleFixer,
)
assert fixer(trail_name_us, AWS_REGION_US_EAST_1)
fixer = CloudtrailLogsS3BucketIsNotPubliclyAccessibleFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1, resource_id=trail_name_us)
@mock_aws
def test_trail_bucket_public_acl_error(self):
@@ -145,9 +145,11 @@ class Test_cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer:
new=S3(aws_provider),
),
):
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_logs_s3_bucket_is_not_publicly_accessible.cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer import (
fixer,
CloudtrailLogsS3BucketIsNotPubliclyAccessibleFixer,
)
assert not fixer(trail_name_us, AWS_REGION_US_EAST_1)
fixer = CloudtrailLogsS3BucketIsNotPubliclyAccessibleFixer()
assert not fixer.fix(
region=AWS_REGION_US_EAST_1, resource_id=trail_name_us
)
tests/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer_test.py
+4 -4
View File
@@ -6,7 +6,7 @@ from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
class Test_cloudtrail_multi_region_enabled_fixer:
class TestCloudtrailMultiRegionEnabledFixer:
@mock_aws
def test_cloudtrail_multi_region_enabled_fixer(self):
from prowler.providers.aws.services.cloudtrail.cloudtrail_service import (
@@ -36,9 +36,9 @@ class Test_cloudtrail_multi_region_enabled_fixer:
"prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled.cloudtrail_multi_region_enabled_fixer.cloudtrail_client",
new=Cloudtrail(aws_provider),
):
# Test Check
from prowler.providers.aws.services.cloudtrail.cloudtrail_multi_region_enabled.cloudtrail_multi_region_enabled_fixer import (
fixer,
CloudtrailMultiRegionEnabledFixer,
)
assert fixer(AWS_REGION_US_EAST_1)
fixer = CloudtrailMultiRegionEnabledFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1)
tests/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer_test.py
+11 -8
View File
@@ -38,7 +38,7 @@ def mock_make_api_call_codeartifact_error(self, operation_name, kwarg):
return mock_make_api_call(self, operation_name, kwarg)
class Test_codeartifact_packages_external_public_publishing_disabled_fixer:
class TestCodeartifactPackagesExternalPublicPublishingDisabledFixer:
@mock_aws
def test_repository_package_public_publishing_origin_internal(self):
with mock.patch(
@@ -57,12 +57,14 @@ class Test_codeartifact_packages_external_public_publishing_disabled_fixer:
new=CodeArtifact(aws_provider),
),
):
# Test Fixer
from prowler.providers.aws.services.codeartifact.codeartifact_packages_external_public_publishing_disabled.codeartifact_packages_external_public_publishing_disabled_fixer import (
fixer,
CodeartifactPackagesExternalPublicPublishingDisabledFixer,
)
assert fixer("test/test-package", AWS_REGION_EU_WEST_1)
fixer = CodeartifactPackagesExternalPublicPublishingDisabledFixer()
assert fixer.fix(
region=AWS_REGION_EU_WEST_1, resource_id="test/test-package"
)
@mock_aws
def test_repository_package_public_publishing_origin_internal_error(self):
@@ -70,7 +72,6 @@ class Test_codeartifact_packages_external_public_publishing_disabled_fixer:
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_codeartifact_error,
):
aws_provider = set_mocked_aws_provider([AWS_REGION_EU_WEST_1])
with (
@@ -83,9 +84,11 @@ class Test_codeartifact_packages_external_public_publishing_disabled_fixer:
new=CodeArtifact(aws_provider),
),
):
# Test Fixer
from prowler.providers.aws.services.codeartifact.codeartifact_packages_external_public_publishing_disabled.codeartifact_packages_external_public_publishing_disabled_fixer import (
fixer,
CodeartifactPackagesExternalPublicPublishingDisabledFixer,
)
assert not fixer("non-existing-package", AWS_REGION_EU_WEST_1)
fixer = CodeartifactPackagesExternalPublicPublishingDisabledFixer()
assert not fixer.fix(
region=AWS_REGION_EU_WEST_1, resource_id="non-existing-package"
)
tests/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer_test.py
+10 -6
View File
@@ -39,7 +39,7 @@ def mock_make_api_call_public_snapshot_error(self, operation_name, kwarg):
return mock_make_api_call(self, operation_name, kwarg)
class Test_documentdb_cluster_public_snapshot_fixer:
class TestDocumentdbClusterPublicSnapshotFixer:
@mock_aws
def test_documentdb_cluster_public_snapshot_fixer(self):
with mock.patch(
@@ -63,10 +63,13 @@ class Test_documentdb_cluster_public_snapshot_fixer:
),
):
from prowler.providers.aws.services.documentdb.documentdb_cluster_public_snapshot.documentdb_cluster_public_snapshot_fixer import (
fixer,
DocumentdbClusterPublicSnapshotFixer,
)
assert fixer(resource_id="test-snapshot", region=AWS_REGION_EU_WEST_1)
fixer = DocumentdbClusterPublicSnapshotFixer()
assert fixer.fix(
region=AWS_REGION_EU_WEST_1, resource_id="test-snapshot"
)
@mock_aws
def test_documentdb_cluster_public_snapshot_fixer_error(self):
@@ -91,9 +94,10 @@ class Test_documentdb_cluster_public_snapshot_fixer:
),
):
from prowler.providers.aws.services.documentdb.documentdb_cluster_public_snapshot.documentdb_cluster_public_snapshot_fixer import (
fixer,
DocumentdbClusterPublicSnapshotFixer,
)
assert not fixer(
resource_id="test-snapshot", region=AWS_REGION_EU_WEST_1
fixer = DocumentdbClusterPublicSnapshotFixer()
assert not fixer.fix(
region=AWS_REGION_EU_WEST_1, resource_id="test-snapshot"
)
tests/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer_test.py
+12 -7
View File
@@ -12,7 +12,7 @@ from tests.providers.aws.utils import (
EXAMPLE_AMI_ID = "ami-12c6146b"
class Test_ec2_ami_public_fixer:
class TestEc2AmiPublicFixer:
@mock_aws
def test_one_private_ami(self):
ec2 = client("ec2", region_name=AWS_REGION_US_EAST_1)
@@ -42,10 +42,11 @@ class Test_ec2_ami_public_fixer:
),
):
from prowler.providers.aws.services.ec2.ec2_ami_public.ec2_ami_public_fixer import (
fixer,
Ec2AmiPublicFixer,
)
assert fixer(image_id, AWS_REGION_US_EAST_1)
fixer = Ec2AmiPublicFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1, resource_id=image_id)
@mock_aws
def test_one_public_ami(self):
@@ -85,10 +86,11 @@ class Test_ec2_ami_public_fixer:
),
):
from prowler.providers.aws.services.ec2.ec2_ami_public.ec2_ami_public_fixer import (
fixer,
Ec2AmiPublicFixer,
)
assert fixer(image_id, AWS_REGION_US_EAST_1)
fixer = Ec2AmiPublicFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1, resource_id=image_id)
@mock_aws
def test_one_public_ami_error(self):
@@ -128,7 +130,10 @@ class Test_ec2_ami_public_fixer:
),
):
from prowler.providers.aws.services.ec2.ec2_ami_public.ec2_ami_public_fixer import (
fixer,
Ec2AmiPublicFixer,
)
assert not fixer("image_id_non_existing", AWS_REGION_US_EAST_1)
fixer = Ec2AmiPublicFixer()
assert not fixer.fix(
region=AWS_REGION_US_EAST_1, resource_id="image_id_non_existing"
)
tests/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer_test.py
+4 -4
View File
@@ -5,7 +5,7 @@ from moto import mock_aws
from tests.providers.aws.utils import AWS_REGION_US_EAST_1, set_mocked_aws_provider
class Test_ec2_ebs_default_encryption_fixer:
class TestEc2EbsDefaultEncryptionFixer:
@mock_aws
def test_ec2_ebs_encryption_fixer(self):
from prowler.providers.aws.services.ec2.ec2_service import EC2
@@ -23,8 +23,8 @@ class Test_ec2_ebs_default_encryption_fixer:
),
):
from prowler.providers.aws.services.ec2.ec2_ebs_default_encryption.ec2_ebs_default_encryption_fixer import (
fixer,
Ec2EbsDefaultEncryptionFixer,
)
# By default, the account has not public access blocked
assert fixer(region=AWS_REGION_US_EAST_1)
fixer = Ec2EbsDefaultEncryptionFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1)
tests/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer_test.py
+9 -9
View File
@@ -38,14 +38,13 @@ def mock_make_api_call_error(self, operation_name, kwarg):
return mock_make_api_call(self, operation_name, kwarg)
class Test_ec2_ebs_public_snapshot_fixer_test:
class TestEc2EbsPublicSnapshotFixer:
@mock_aws
def test_ebs_public_snapshot(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call",
new=mock_make_api_call_public_snapshot,
):
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
@@ -62,19 +61,18 @@ class Test_ec2_ebs_public_snapshot_fixer_test:
new=EC2(aws_provider),
),
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_ebs_public_snapshot.ec2_ebs_public_snapshot_fixer import (
fixer,
Ec2EbsPublicSnapshotFixer,
)
assert fixer("testsnap", AWS_REGION_US_EAST_1)
fixer = Ec2EbsPublicSnapshotFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1, resource_id="testsnap")
@mock_aws
def test_ebs_public_snapshot_error(self):
with mock.patch(
"botocore.client.BaseClient._make_api_call", new=mock_make_api_call_error
):
from prowler.providers.aws.services.ec2.ec2_service import EC2
aws_provider = set_mocked_aws_provider(
@@ -91,9 +89,11 @@ class Test_ec2_ebs_public_snapshot_fixer_test:
new=EC2(aws_provider),
),
):
# Test Check
from prowler.providers.aws.services.ec2.ec2_ebs_public_snapshot.ec2_ebs_public_snapshot_fixer import (
fixer,
Ec2EbsPublicSnapshotFixer,
)
assert not fixer("testsnap", AWS_REGION_US_EAST_1)
fixer = Ec2EbsPublicSnapshotFixer()
assert not fixer.fix(
region=AWS_REGION_US_EAST_1, resource_id="testsnap"
)
tests/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer_test.py
+4 -5
View File
@@ -26,7 +26,7 @@ def mock_enable_snapshot_block_public_access(State):
return {"State": State}
class Test_ec2_ebs_snapshot_account_block_public_access_fixer:
class TestEc2EbsSnapshotAccountBlockPublicAccessFixer:
@mock_aws
def test_ec2_ebs_snapshot_account_block_public_access_fixer(self):
ec2_service = mock.MagicMock()
@@ -61,10 +61,9 @@ class Test_ec2_ebs_snapshot_account_block_public_access_fixer:
ec2_service,
),
):
from prowler.providers.aws.services.ec2.ec2_ebs_snapshot_account_block_public_access.ec2_ebs_snapshot_account_block_public_access_fixer import (
fixer,
Ec2EbsSnapshotAccountBlockPublicAccessFixer,
)
# By default, the account has not public access blocked
assert fixer(region=AWS_REGION_US_EAST_1)
fixer = Ec2EbsSnapshotAccountBlockPublicAccessFixer()
assert fixer.fix(region=AWS_REGION_US_EAST_1)