From 07c5ae547fb5e745d9f0a2eda87032fbbbb36268 Mon Sep 17 00:00:00 2001 From: Daniel Barranquero Date: Wed, 11 Jun 2025 14:21:43 +0200 Subject: [PATCH] chore(aws): migrate fixers to new structure until ec2_ebs --- .../accessanalyzer_enabled_fixer.py | 93 +++++++---- ..._function_not_publicly_accessible_fixer.py | 124 +++++++++------ ...bucket_is_not_publicly_accessible_fixer.py | 105 ++++++++----- .../cloudtrail_multi_region_enabled_fixer.py | 125 +++++++++------ ...rail_threat_detection_enumeration_fixer.py | 145 ++++++++++-------- ...rail_threat_detection_llm_jacking_fixer.py | 145 ++++++++++-------- ...at_detection_privilege_escalation_fixer.py | 145 ++++++++++-------- ...ternal_public_publishing_disabled_fixer.py | 120 +++++++++------ ...ocumentdb_cluster_public_snapshot_fixer.py | 91 +++++++---- .../ec2_ami_public/ec2_ami_public_fixer.py | 90 +++++++---- .../ec2_ebs_default_encryption_fixer.py | 80 ++++++---- .../ec2_ebs_public_snapshot_fixer.py | 93 +++++++---- ...pshot_account_block_public_access_fixer.py | 90 +++++++---- .../accessanalyzer_enabled_fixer_test.py | 6 +- ...tion_not_publicly_accessible_fixer_test.py | 26 ++-- ...t_is_not_publicly_accessible_fixer_test.py | 16 +- ...udtrail_multi_region_enabled_fixer_test.py | 8 +- ...l_public_publishing_disabled_fixer_test.py | 19 ++- ...ntdb_cluster_public_snapshot_fixer_test.py | 16 +- .../ec2_ami_public_fixer_test.py | 19 ++- .../ec2_ebs_default_encryption_fixer_test.py | 8 +- .../ec2_ebs_public_snapshot_fixer_test.py | 18 +-- ..._account_block_public_access_fixer_test.py | 9 +- 23 files changed, 960 insertions(+), 631 deletions(-) diff --git a/prowler/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer.py b/prowler/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer.py index 120c6d2336..24b020ce6c 100644 --- a/prowler/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer.py +++ b/prowler/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer.py b/prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer.py index 3a74fb17d6..13f12e28c7 100644 --- a/prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer.py +++ b/prowler/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py index f79780093f..fc44434c57 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer.py index 92da4fc48f..d08bb3d0b0 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_fixer.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_fixer.py index 90fb25d6f6..71ee9211e0 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_fixer.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_enumeration/cloudtrail_threat_detection_enumeration_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_fixer.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_fixer.py index 90fb25d6f6..d8106475a1 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_fixer.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_llm_jacking/cloudtrail_threat_detection_llm_jacking_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_fixer.py b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_fixer.py index 90fb25d6f6..dea49c8858 100644 --- a/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_fixer.py +++ b/prowler/providers/aws/services/cloudtrail/cloudtrail_threat_detection_privilege_escalation/cloudtrail_threat_detection_privilege_escalation_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer.py b/prowler/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer.py index 1b140b11f7..de418ee4ea 100644 --- a/prowler/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer.py +++ b/prowler/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer.py b/prowler/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer.py index 341eaf251d..de3affb72e 100644 --- a/prowler/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer.py +++ b/prowler/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer.py b/prowler/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer.py index d26f4733b4..3a2e014e81 100644 --- a/prowler/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer.py +++ b/prowler/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer.py b/prowler/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer.py index f733b4f2da..0f615ad5b4 100644 --- a/prowler/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer.py +++ b/prowler/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer.py b/prowler/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer.py index 1c906b420b..e1e7702401 100644 --- a/prowler/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer.py +++ b/prowler/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer.py @@ -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 diff --git a/prowler/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer.py b/prowler/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer.py index d1a21c51cf..4d0a15f40f 100644 --- a/prowler/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer.py +++ b/prowler/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer.py @@ -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 diff --git a/tests/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer_test.py b/tests/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer_test.py index 39f90095f1..bb1ca50fdd 100644 --- a/tests/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer_test.py +++ b/tests/providers/aws/services/accessanalyzer/accessanalyzer_enabled/accessanalyzer_enabled_fixer_test.py @@ -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) diff --git a/tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer_test.py b/tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer_test.py index 0535856c15..5eefd19a51 100644 --- a/tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer_test.py +++ b/tests/providers/aws/services/awslambda/awslambda_function_not_publicly_accessible/awslambda_function_not_publicly_accessible_fixer_test.py @@ -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") diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py index a3eea32968..332a005c65 100644 --- a/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py +++ b/tests/providers/aws/services/cloudtrail/cloudtrail_logs_s3_bucket_is_not_publicly_accessible/cloudtrail_logs_s3_bucket_is_not_publicly_accessible_fixer_test.py @@ -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 + ) diff --git a/tests/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer_test.py b/tests/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer_test.py index 756ac7bb6e..19a729a804 100644 --- a/tests/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer_test.py +++ b/tests/providers/aws/services/cloudtrail/cloudtrail_multi_region_enabled/cloudtrail_multi_region_enabled_fixer_test.py @@ -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) diff --git a/tests/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer_test.py b/tests/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer_test.py index f776406e9c..d410d871bb 100644 --- a/tests/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer_test.py +++ b/tests/providers/aws/services/codeartifact/codeartifact_packages_external_public_publishing_disabled/codeartifact_packages_external_public_publishing_disabled_fixer_test.py @@ -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" + ) diff --git a/tests/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer_test.py b/tests/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer_test.py index 6ac53f42cc..afaae3d6e3 100644 --- a/tests/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer_test.py +++ b/tests/providers/aws/services/documentdb/documentdb_cluster_public_snapshot/documentdb_cluster_public_snapshot_fixer_test.py @@ -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" ) diff --git a/tests/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer_test.py b/tests/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer_test.py index e87718efb8..ffc25f7a02 100644 --- a/tests/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer_test.py +++ b/tests/providers/aws/services/ec2/ec2_ami_public/ec2_ami_public_fixer_test.py @@ -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" + ) diff --git a/tests/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer_test.py b/tests/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer_test.py index 1fdf72c337..eb31c97a19 100644 --- a/tests/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer_test.py +++ b/tests/providers/aws/services/ec2/ec2_ebs_default_encryption/ec2_ebs_default_encryption_fixer_test.py @@ -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) diff --git a/tests/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer_test.py b/tests/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer_test.py index 77e7e802c3..4647c2c3e2 100644 --- a/tests/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer_test.py +++ b/tests/providers/aws/services/ec2/ec2_ebs_public_snapshot/ec2_ebs_public_snapshot_fixer_test.py @@ -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" + ) diff --git a/tests/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer_test.py b/tests/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer_test.py index be715ac5ed..4d937b1835 100644 --- a/tests/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer_test.py +++ b/tests/providers/aws/services/ec2/ec2_ebs_snapshot_account_block_public_access/ec2_ebs_snapshot_account_block_public_access_fixer_test.py @@ -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)