diff --git a/prowler/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled.py b/prowler/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled.py index baedd5f6b7..ceecaf9216 100644 --- a/prowler/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled.py +++ b/prowler/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled.py @@ -5,7 +5,7 @@ from prowler.providers.aws.services.efs.efs_client import efs_client class efs_encryption_at_rest_enabled(Check): def execute(self): findings = [] - for fs in efs_client.filesystems: + for fs in efs_client.filesystems.values(): report = Check_Report_AWS(self.metadata()) report.region = fs.region report.resource_id = fs.id diff --git a/prowler/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled.py b/prowler/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled.py index ae0e19de86..65ab9d0dd4 100644 --- a/prowler/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled.py +++ b/prowler/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled.py @@ -5,7 +5,7 @@ from prowler.providers.aws.services.efs.efs_client import efs_client class efs_have_backup_enabled(Check): def execute(self): findings = [] - for fs in efs_client.filesystems: + for fs in efs_client.filesystems.values(): report = Check_Report_AWS(self.metadata()) report.region = fs.region report.resource_id = fs.id diff --git a/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/__init__.py b/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible.metadata.json b/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible.metadata.json new file mode 100644 index 0000000000..99dc4ae0d6 --- /dev/null +++ b/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible.metadata.json @@ -0,0 +1,34 @@ +{ + "Provider": "aws", + "CheckID": "efs_mount_target_not_publicly_accessible", + "CheckTitle": "EFS mount targets should not be publicly accessible", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "efs", + "SubServiceName": "", + "ResourceIdTemplate": "arn:aws:elasticfilesystem:{region}:{account-id}:file-system/{filesystem-id}", + "Severity": "medium", + "ResourceType": "AwsEfsFileSystem", + "Description": "This control checks whether an Amazon EFS mount target is associated with a public subnet since it can be accessed from the internet.", + "Risk": "Mount targets in public subnets may expose your EFS to unauthorized access or attacks.", + "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/efs-mount-target-public-accessible.html", + "Remediation": { + "Code": { + "CLI": "aws efs create-mount-target --file-system-id --subnet-id --security-groups ", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/efs-controls.html#efs-6", + "Terraform": "" + }, + "Recommendation": { + "Text": "Recreate the EFS mount target in a private subnet to ensure it is not publicly accessible.", + "Url": "https://docs.aws.amazon.com/efs/latest/ug/accessing-fs.html" + } + }, + "Categories": [ + "internet-exposed" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible.py b/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible.py new file mode 100644 index 0000000000..ede0b25329 --- /dev/null +++ b/prowler/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible.py @@ -0,0 +1,28 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.efs.efs_client import efs_client +from prowler.providers.aws.services.vpc.vpc_client import vpc_client + + +class efs_mount_target_not_publicly_accessible(Check): + def execute(self): + findings = [] + for fs in efs_client.filesystems.values(): + report = Check_Report_AWS(self.metadata()) + report.region = fs.region + report.resource_id = fs.id + report.resource_arn = fs.arn + report.resource_tags = fs.tags + report.status = "PASS" + report.status_extended = ( + f"EFS {fs.id} does not have any public mount targets." + ) + mount_targets = [] + for mt in fs.mount_targets: + if vpc_client.vpc_subnets[mt.subnet_id].public: + mount_targets.append(mt) + if mount_targets: + report.status = "FAIL" + report.status_extended = f"EFS {fs.id} has public mount targets: {', '.join([mt.id for mt in mount_targets])}" + + findings.append(report) + return findings diff --git a/prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py b/prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py index 73d89b938e..2a3f44c578 100644 --- a/prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py +++ b/prowler/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible.py @@ -6,7 +6,7 @@ from prowler.providers.aws.services.iam.lib.policy import is_policy_public class efs_not_publicly_accessible(Check): def execute(self): findings = [] - for fs in efs_client.filesystems: + for fs in efs_client.filesystems.values(): report = Check_Report_AWS(self.metadata()) report.region = fs.region report.resource_id = fs.id diff --git a/prowler/providers/aws/services/efs/efs_service.py b/prowler/providers/aws/services/efs/efs_service.py index 5cf51ce827..b3f8140e1e 100644 --- a/prowler/providers/aws/services/efs/efs_service.py +++ b/prowler/providers/aws/services/efs/efs_service.py @@ -14,9 +14,12 @@ class EFS(AWSService): def __init__(self, provider): # Call AWSService's __init__ super().__init__(__class__.__name__, provider) - self.filesystems = [] + self.filesystems = {} self.__threading_call__(self._describe_file_systems) - self._describe_file_system_policies() + self.__threading_call__( + self._describe_file_system_policies, self.filesystems.values() + ) + self.__threading_call__(self._describe_mount_targets, self.filesystems.values()) def _describe_file_systems(self, regional_client): logger.info("EFS - Describing file systems...") @@ -31,55 +34,98 @@ class EFS(AWSService): if not self.audit_resources or ( is_resource_filtered(efs_arn, self.audit_resources) ): - self.filesystems.append( - FileSystem( - id=efs_id, - arn=efs_arn, - region=regional_client.region, - policy=None, - backup_policy=None, - encrypted=efs["Encrypted"], - tags=efs.get("Tags"), - ) + self.filesystems[efs_arn] = FileSystem( + id=efs_id, + arn=efs_arn, + region=regional_client.region, + encrypted=efs["Encrypted"], + tags=efs.get("Tags"), ) except Exception as error: logger.error( f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) - def _describe_file_system_policies(self): + def _describe_file_system_policies(self, filesystem): logger.info("EFS - Describing file system policies...") try: - for filesystem in self.filesystems: - for region, client in self.regional_clients.items(): - if filesystem.region == region: - try: - filesystem.backup_policy = client.describe_backup_policy( - FileSystemId=filesystem.id - )["BackupPolicy"]["Status"] - except ClientError as e: - if e.response["Error"]["Code"] == "PolicyNotFound": - filesystem.backup_policy = "DISABLED" - try: - fs_policy = client.describe_file_system_policy( - FileSystemId=filesystem.id + client = self.regional_clients[filesystem.region] + try: + filesystem.backup_policy = client.describe_backup_policy( + FileSystemId=filesystem.id + )["BackupPolicy"]["Status"] + except ClientError as error: + if error.response["Error"]["Code"] == "PolicyNotFound": + filesystem.backup_policy = "DISABLED" + logger.warning( + f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + else: + logger.error( + f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + try: + fs_policy = client.describe_file_system_policy( + FileSystemId=filesystem.id + ) + if "Policy" in fs_policy: + filesystem.policy = json.loads(fs_policy["Policy"]) + except ClientError as error: + if error.response["Error"]["Code"] == "PolicyNotFound": + filesystem.policy = {} + logger.warning( + f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + else: + logger.error( + f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + except Exception as error: + logger.error( + f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + + def _describe_mount_targets(self, filesystem): + logger.info("EFS - Describing mount targets...") + try: + client = self.regional_clients[filesystem.region] + describe_mount_target_paginator = client.get_paginator( + "describe_mount_targets" + ) + for page in describe_mount_target_paginator.paginate( + FileSystemId=filesystem.id + ): + for mount_target in page["MountTargets"]: + mount_target_id = mount_target["MountTargetId"] + mount_target_arn = f"arn:{self.audited_partition}:elasticfilesystem:{client.region}:{self.audited_account}:mount-target/{mount_target_id}" + if not self.audit_resources or ( + is_resource_filtered(mount_target_arn, self.audit_resources) + ): + self.filesystems[filesystem.arn].mount_targets.append( + MountTarget( + id=mount_target_id, + file_system_id=mount_target["FileSystemId"], + subnet_id=mount_target["SubnetId"], ) - if "Policy" in fs_policy: - filesystem.policy = json.loads(fs_policy["Policy"]) - except ClientError as e: - if e.response["Error"]["Code"] == "PolicyNotFound": - filesystem.policy = {} + ) except Exception as error: logger.error( f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) +class MountTarget(BaseModel): + id: str + file_system_id: str + subnet_id: str + + class FileSystem(BaseModel): id: str arn: str region: str - policy: Optional[dict] - backup_policy: Optional[str] + policy: Optional[dict] = {} + backup_policy: Optional[str] = "DISABLED" encrypted: bool + mount_targets: list[MountTarget] = [] tags: Optional[list] = [] diff --git a/tests/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled_test.py b/tests/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled_test.py index 9d0036a159..9670fceb7c 100644 --- a/tests/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled_test.py +++ b/tests/providers/aws/services/efs/efs_encryption_at_rest_enabled/efs_encryption_at_rest_enabled_test.py @@ -1,34 +1,37 @@ -from re import search from unittest import mock -from prowler.providers.aws.services.efs.efs_service import FileSystem +from boto3 import client +from moto import mock_aws -# Mock Test Region -AWS_REGION = "eu-west-1" -AWS_ACCOUNT_NUMBER = "123456789012" +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) -file_system_id = "fs-c7a0456e" - -backup_valid_policy_status = "ENABLED" +CREATION_TOKEN = "fs-123" class Test_efs_encryption_at_rest_enabled: + @mock_aws def test_efs_encryption_enabled(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=None, - backup_policy=backup_valid_policy_status, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + filesystem = efs_client.create_file_system( + CreationToken=CREATION_TOKEN, Encrypted=True + ) + + efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{filesystem['FileSystemId']}" + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled import ( efs_encryption_at_rest_enabled, @@ -38,26 +41,33 @@ class Test_efs_encryption_at_rest_enabled: result = check.execute() assert len(result) == 1 assert result[0].status == "PASS" - assert search("has encryption at rest enabled", result[0].status_extended) - assert result[0].resource_id == file_system_id + assert result[0].region == AWS_REGION_US_EAST_1 + assert ( + result[0].status_extended + == f"EFS {filesystem['FileSystemId']} has encryption at rest enabled." + ) + assert result[0].resource_id == filesystem["FileSystemId"] assert result[0].resource_arn == efs_arn + @mock_aws def test_efs_encryption_disabled(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=None, - backup_policy=backup_valid_policy_status, - encrypted=False, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + filesystem = efs_client.create_file_system( + CreationToken=CREATION_TOKEN, Encrypted=False + ) + + efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{filesystem['FileSystemId']}" + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_encryption_at_rest_enabled.efs_encryption_at_rest_enabled import ( efs_encryption_at_rest_enabled, @@ -67,8 +77,10 @@ class Test_efs_encryption_at_rest_enabled: result = check.execute() assert len(result) == 1 assert result[0].status == "FAIL" - assert search( - "does not have encryption at rest enabled", result[0].status_extended + assert result[0].region == AWS_REGION_US_EAST_1 + assert ( + result[0].status_extended + == f"EFS {filesystem['FileSystemId']} does not have encryption at rest enabled." ) - assert result[0].resource_id == file_system_id + assert result[0].resource_id == filesystem["FileSystemId"] assert result[0].resource_arn == efs_arn diff --git a/tests/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled_test.py b/tests/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled_test.py index 4c53c10010..3bdaf808ba 100644 --- a/tests/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled_test.py +++ b/tests/providers/aws/services/efs/efs_have_backup_enabled/efs_have_backup_enabled_test.py @@ -1,36 +1,36 @@ from re import search from unittest import mock -from prowler.providers.aws.services.efs.efs_service import FileSystem +from boto3 import client +from moto import mock_aws -# Mock Test Region -AWS_REGION = "eu-west-1" -AWS_ACCOUNT_NUMBER = "123456789012" +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) -file_system_id = "fs-c7a0456e" - -backup_valid_policy_status = "ENABLED" -backup_valid_invalid_policy_status_1 = "DISABLING" -backup_valid_invalid_policy_status_2 = "DISABLED" +CREATION_TOKEN = "fs-123" class Test_efs_have_backup_enabled: + @mock_aws def test_efs_valid_backup_policy(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=None, - backup_policy=backup_valid_policy_status, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system( + CreationToken=CREATION_TOKEN, Backup=True + ) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled import ( efs_have_backup_enabled, @@ -41,25 +41,29 @@ class Test_efs_have_backup_enabled: assert len(result) == 1 assert result[0].status == "PASS" assert search("has backup enabled", result[0].status_extended) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" + ) + @mock_aws def test_efs_invalid_policy_backup_1(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=None, - backup_policy=backup_valid_invalid_policy_status_1, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system( + CreationToken=CREATION_TOKEN, Backup=False + ) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled import ( efs_have_backup_enabled, @@ -70,25 +74,29 @@ class Test_efs_have_backup_enabled: assert len(result) == 1 assert result[0].status == "FAIL" assert search("does not have backup enabled", result[0].status_extended) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" + ) + @mock_aws def test_efs_invalid_policy_backup_2(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=None, - backup_policy=backup_valid_invalid_policy_status_2, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system( + CreationToken=CREATION_TOKEN, Backup=False + ) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_have_backup_enabled.efs_have_backup_enabled import ( efs_have_backup_enabled, @@ -99,5 +107,8 @@ class Test_efs_have_backup_enabled: assert len(result) == 1 assert result[0].status == "FAIL" assert search("does not have backup enabled", result[0].status_extended) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" + ) diff --git a/tests/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible_test.py b/tests/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible_test.py new file mode 100644 index 0000000000..5893b96a01 --- /dev/null +++ b/tests/providers/aws/services/efs/efs_mount_target_not_publicly_accessible/efs_mount_target_not_publicly_accessible_test.py @@ -0,0 +1,188 @@ +from unittest import mock + +from boto3 import client +from moto import mock_aws + +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) + +CREATION_TOKEN = "fs-123" + + +class Test_efs_mount_target_not_publicly_accessible: + @mock_aws + def test_efs_no_file_system(self): + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client", + new=EFS(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import ( + efs_mount_target_not_publicly_accessible, + ) + + check = efs_mount_target_not_publicly_accessible() + result = check.execute() + assert len(result) == 0 + + @mock_aws + def test_efs_no_mount_target(self): + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client", + new=EFS(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import ( + efs_mount_target_not_publicly_accessible, + ) + + check = efs_mount_target_not_publicly_accessible() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"EFS {file_system['FileSystemId']} does not have any public mount targets." + ) + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" + ) + + @mock_aws + def test_efs_mount_target_public_subnet(self): + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + vpc = ec2_client.create_vpc( + CidrBlock="172.28.7.0/24", InstanceTenancy="default" + ) + igw = ec2_client.create_internet_gateway() + ec2_client.attach_internet_gateway( + InternetGatewayId=igw["InternetGateway"]["InternetGatewayId"], + VpcId=vpc["Vpc"]["VpcId"], + ) + subnet_public = ec2_client.create_subnet( + VpcId=vpc["Vpc"]["VpcId"], + CidrBlock="172.28.7.192/26", + AvailabilityZone=f"{AWS_REGION_US_EAST_1}a", + ) + route_table = ec2_client.create_route_table(VpcId=vpc["Vpc"]["VpcId"]) + ec2_client.create_route( + RouteTableId=route_table["RouteTable"]["RouteTableId"], + DestinationCidrBlock="0.0.0.0/0", # Ruta predeterminada para todo el tráfico + GatewayId=igw["InternetGateway"]["InternetGatewayId"], + ) + ec2_client.associate_route_table( + RouteTableId=route_table["RouteTable"]["RouteTableId"], + SubnetId=subnet_public["Subnet"]["SubnetId"], + ) + mount_target = efs_client.create_mount_target( + FileSystemId=file_system["FileSystemId"], + SubnetId=subnet_public["Subnet"]["SubnetId"], + ) + + from prowler.providers.aws.services.efs.efs_service import EFS + from prowler.providers.aws.services.vpc.vpc_service import VPC + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client", + new=EFS(aws_provider), + ), mock.patch( + "prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.vpc_client", + new=VPC(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import ( + efs_mount_target_not_publicly_accessible, + ) + + check = efs_mount_target_not_publicly_accessible() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"EFS {file_system['FileSystemId']} has public mount targets: {mount_target['MountTargetId']}" + ) + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" + ) + + @mock_aws + def test_efs_mount_target_private_subnet(self): + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + ec2_client = client("ec2", region_name=AWS_REGION_US_EAST_1) + vpc = ec2_client.create_vpc( + CidrBlock="172.28.7.0/24", InstanceTenancy="default" + ) + subnet_private = ec2_client.create_subnet( + VpcId=vpc["Vpc"]["VpcId"], + CidrBlock="172.28.7.192/26", + AvailabilityZone=f"{AWS_REGION_US_EAST_1}a", + ) + efs_client.create_mount_target( + FileSystemId=file_system["FileSystemId"], + SubnetId=subnet_private["Subnet"]["SubnetId"], + ) + + from prowler.providers.aws.services.efs.efs_service import EFS + from prowler.providers.aws.services.vpc.vpc_service import VPC + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + + with mock.patch( + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.efs_client", + new=EFS(aws_provider), + ), mock.patch( + "prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible.vpc_client", + new=VPC(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_mount_target_not_publicly_accessible.efs_mount_target_not_publicly_accessible import ( + efs_mount_target_not_publicly_accessible, + ) + + check = efs_mount_target_not_publicly_accessible() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "PASS" + assert ( + result[0].status_extended + == f"EFS {file_system['FileSystemId']} does not have any public mount targets." + ) + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" + ) diff --git a/tests/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible_test.py b/tests/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible_test.py index 2ea9111cde..14adf7ac22 100644 --- a/tests/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible_test.py +++ b/tests/providers/aws/services/efs/efs_not_publicly_accessible/efs_not_publicly_accessible_test.py @@ -1,96 +1,137 @@ +import json from unittest import mock -from prowler.providers.aws.services.efs.efs_service import FileSystem +import botocore +from boto3 import client +from moto import mock_aws -# Mock Test Region -AWS_REGION = "eu-west-1" -AWS_ACCOUNT_NUMBER = "123456789012" +from tests.providers.aws.utils import ( + AWS_ACCOUNT_NUMBER, + AWS_REGION_US_EAST_1, + set_mocked_aws_provider, +) -file_system_id = "fs-c7a0456e" +CREATION_TOKEN = "fs-123" -filesystem_policy = { - "Id": "1", - "Statement": [ - { - "Effect": "Allow", - "Action": ["elasticfilesystem:ClientMount"], - "Principal": {"AWS": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root"}, - } - ], -} +FILE_SYSTEM_POLICY = json.dumps( + { + "Id": "1", + "Statement": [ + { + "Effect": "Allow", + "Action": ["elasticfilesystem:ClientMount"], + "Principal": {"AWS": "arn:aws:iam::123456789012:root"}, + } + ], + } +) -filesystem_invalid_policy = { - "Id": "1", - "Statement": [ - { - "Effect": "Allow", - "Action": ["elasticfilesystem:ClientMount"], - "Principal": {"AWS": "*"}, - } - ], -} +FILE_SYSTEM_INVALID_POLICY = json.dumps( + { + "Id": "1", + "Statement": [ + { + "Effect": "Allow", + "Action": ["elasticfilesystem:ClientMount"], + "Principal": {"AWS": "*"}, + } + ], + } +) # https://docs.aws.amazon.com/efs/latest/ug/access-control-block-public-access.html#what-is-a-public-policy -filesystem_policy_with_source_arn_condition = { - "Version": "2012-10-17", - "Id": "efs-policy-wizard-15ad9567-2546-4bbb-8168-5541b6fc0e55", - "Statement": [ - { - "Sid": "efs-statement-14a7191c-9401-40e7-a388-6af6cfb7dd9c", - "Effect": "Allow", - "Principal": {"AWS": "*"}, - "Action": [ - "elasticfilesystem:ClientMount", - "elasticfilesystem:ClientWrite", - "elasticfilesystem:ClientRootAccess", - ], - "Condition": { - "ArnEquals": { - "aws:SourceArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" - } - }, - } - ], -} +FILE_SYSTEM_POLICY_WITH_SOURCE_ARN_CONDITION = json.dumps( + { + "Version": "2012-10-17", + "Id": "efs-policy-wizard-15ad9567-2546-4bbb-8168-5541b6fc0e55", + "Statement": [ + { + "Sid": "efs-statement-14a7191c-9401-40e7-a388-6af6cfb7dd9c", + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": [ + "elasticfilesystem:ClientMount", + "elasticfilesystem:ClientWrite", + "elasticfilesystem:ClientRootAccess", + ], + "Condition": { + "ArnEquals": { + "aws:SourceArn": f"arn:aws:iam::{AWS_ACCOUNT_NUMBER}:root" + } + }, + } + ], + } +) # https://docs.aws.amazon.com/efs/latest/ug/access-control-block-public-access.html#what-is-a-public-policy -filesystem_policy_with_mount_target_condition = { - "Version": "2012-10-17", - "Id": "efs-policy-wizard-15ad9567-2546-4bbb-8168-5541b6fc0e55", - "Statement": [ - { - "Sid": "efs-statement-14a7191c-9401-40e7-a388-6af6cfb7dd9c", - "Effect": "Allow", - "Principal": {"AWS": "*"}, - "Action": [ - "elasticfilesystem:ClientMount", - "elasticfilesystem:ClientWrite", - "elasticfilesystem:ClientRootAccess", - ], - "Condition": {"Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"}}, - } - ], -} +FILE_SYSTEM_POLICY_WITH_MOUNT_TARGET_CONDITION = json.dumps( + { + "Version": "2012-10-17", + "Id": "efs-policy-wizard-15ad9567-2546-4bbb-8168-5541b6fc0e55", + "Statement": [ + { + "Sid": "efs-statement-14a7191c-9401-40e7-a388-6af6cfb7dd9c", + "Effect": "Allow", + "Principal": {"AWS": "*"}, + "Action": [ + "elasticfilesystem:ClientMount", + "elasticfilesystem:ClientWrite", + "elasticfilesystem:ClientRootAccess", + ], + "Condition": { + "Bool": {"elasticfilesystem:AccessedViaMountTarget": "true"} + }, + } + ], + } +) + +make_api_call = botocore.client.BaseClient._make_api_call + + +def mock_make_api_call(self, operation_name, kwarg): + if operation_name == "DescribeFileSystemPolicy": + return {"Policy": FILE_SYSTEM_POLICY} + + return make_api_call(self, operation_name, kwarg) + + +def mock_make_api_call_v2(self, operation_name, kwarg): + if operation_name == "DescribeFileSystemPolicy": + return {"Policy": FILE_SYSTEM_POLICY_WITH_MOUNT_TARGET_CONDITION} + return make_api_call(self, operation_name, kwarg) + + +def mock_make_api_call_v3(self, operation_name, kwarg): + if operation_name == "DescribeFileSystemPolicy": + return {"Policy": FILE_SYSTEM_POLICY_WITH_SOURCE_ARN_CONDITION} + return make_api_call(self, operation_name, kwarg) + + +def mock_make_api_call_v4(self, operation_name, kwarg): + if operation_name == "DescribeFileSystemPolicy": + return {"Policy": FILE_SYSTEM_INVALID_POLICY} + return make_api_call(self, operation_name, kwarg) class Test_efs_not_publicly_accessible: + @mock_aws + @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) def test_efs_valid_policy(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=filesystem_policy, - backup_policy=None, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import ( efs_not_publicly_accessible, @@ -102,29 +143,32 @@ class Test_efs_not_publicly_accessible: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"EFS {file_system_id} has a policy which does not allow access to any client within the VPC." + == f"EFS {file_system['FileSystemId']} has a policy which does not allow access to any client within the VPC." + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" ) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn - assert result[0].region == AWS_REGION assert result[0].resource_tags == [] + @mock_aws + @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v2) def test_efs_valid_policy_with_mount_target_condition(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=filesystem_policy_with_mount_target_condition, - backup_policy=None, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import ( efs_not_publicly_accessible, @@ -136,29 +180,32 @@ class Test_efs_not_publicly_accessible: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"EFS {file_system_id} has a policy which does not allow access to any client within the VPC." + == f"EFS {file_system['FileSystemId']} has a policy which does not allow access to any client within the VPC." + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" ) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn - assert result[0].region == AWS_REGION assert result[0].resource_tags == [] + @mock_aws + @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v3) def test_efs_valid_policy_with_source_arn_condition(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=filesystem_policy_with_source_arn_condition, - backup_policy=None, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import ( efs_not_publicly_accessible, @@ -170,30 +217,32 @@ class Test_efs_not_publicly_accessible: assert result[0].status == "PASS" assert ( result[0].status_extended - == f"EFS {file_system_id} has a policy which does not allow access to any client within the VPC." + == f"EFS {file_system['FileSystemId']} has a policy which does not allow access to any client within the VPC." + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" ) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn - assert result[0].region == AWS_REGION assert result[0].resource_tags == [] + @mock_aws + @mock.patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call_v4) def test_efs_invalid_policy(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=filesystem_invalid_policy, - backup_policy=None, - encrypted=True, - ) - ] with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import ( efs_not_publicly_accessible, @@ -205,29 +254,31 @@ class Test_efs_not_publicly_accessible: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"EFS {file_system_id} has a policy which allows access to any client within the VPC." + == f"EFS {file_system['FileSystemId']} has a policy which allows access to any client within the VPC." + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" ) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn - assert result[0].region == AWS_REGION assert result[0].resource_tags == [] + @mock_aws def test_efs_no_policy(self): - efs_client = mock.MagicMock - efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system_id}" - efs_client.filesystems = [ - FileSystem( - id=file_system_id, - arn=efs_arn, - region=AWS_REGION, - policy=None, - backup_policy=None, - encrypted=True, - ) - ] + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + from prowler.providers.aws.services.efs.efs_service import EFS + + aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1]) + with mock.patch( - "prowler.providers.aws.services.efs.efs_service.EFS", - efs_client, + "prowler.providers.common.provider.Provider.get_global_provider", + return_value=aws_provider, + ), mock.patch( + "prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible.efs_client", + new=EFS(aws_provider), ): from prowler.providers.aws.services.efs.efs_not_publicly_accessible.efs_not_publicly_accessible import ( efs_not_publicly_accessible, @@ -239,9 +290,12 @@ class Test_efs_not_publicly_accessible: assert result[0].status == "FAIL" assert ( result[0].status_extended - == f"EFS {file_system_id} doesn't have any policy which means it grants full access to any client within the VPC." + == f"EFS {file_system['FileSystemId']} doesn't have any policy which means it grants full access to any client within the VPC." + ) + assert result[0].region == AWS_REGION_US_EAST_1 + assert result[0].resource_id == file_system["FileSystemId"] + assert ( + result[0].resource_arn + == f"arn:aws:elasticfilesystem:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystemId']}" ) - assert result[0].resource_id == file_system_id - assert result[0].resource_arn == efs_arn - assert result[0].region == AWS_REGION assert result[0].resource_tags == [] diff --git a/tests/providers/aws/services/efs/efs_service_test.py b/tests/providers/aws/services/efs/efs_service_test.py index ac0859e006..4e85a9e04f 100644 --- a/tests/providers/aws/services/efs/efs_service_test.py +++ b/tests/providers/aws/services/efs/efs_service_test.py @@ -2,8 +2,6 @@ import json from unittest.mock import patch import botocore -from boto3 import client -from moto import mock_aws from prowler.providers.aws.services.efs.efs_service import EFS from tests.providers.aws.utils import ( @@ -15,13 +13,11 @@ from tests.providers.aws.utils import ( # Mocking Access Analyzer Calls make_api_call = botocore.client.BaseClient._make_api_call -file_system_id = "fs-c7a0456e" +FILE_SYSTEM_ID = "fs-c7a0456e" -creation_token = "console-d215fa78-1f83-4651-b026-facafd8a7da7" +CREATION_TOKEN = "console-d215fa78-1f83-4651-b026-facafd8a7da7" -backup_policy_status = "ENABLED" - -filesystem_policy = { +FILESYSTEM_POLICY = { "Id": "1", "Statement": [ { @@ -34,10 +30,35 @@ filesystem_policy = { def mock_make_api_call(self, operation_name, kwarg): + if operation_name == "DescribeFileSystems": + return { + "FileSystems": [ + { + "FileSystemId": FILE_SYSTEM_ID, + "Encrypted": True, + "Tags": [{"Key": "test", "Value": "test"}], + "BackupPolicy": {"Status": "ENABLED"}, + "Policy": json.dumps(FILESYSTEM_POLICY), + } + ] + } + if operation_name == "DescribeMountTargets": + return { + "MountTargets": [ + { + "MountTargetId": "fsmt-123", + "FileSystemId": FILE_SYSTEM_ID, + "SubnetId": "subnet-123", + "LifeCycleState": "available", + "OwnerId": AWS_ACCOUNT_NUMBER, + "VpcId": "vpc-123", + } + ] + } if operation_name == "DescribeFileSystemPolicy": - return {"FileSystemId": file_system_id, "Policy": json.dumps(filesystem_policy)} + return {"FileSystemId": FILE_SYSTEM_ID, "Policy": json.dumps(FILESYSTEM_POLICY)} if operation_name == "DescribeBackupPolicy": - return {"BackupPolicy": {"Status": backup_policy_status}} + return {"BackupPolicy": {"Status": "ENABLED"}} return make_api_call(self, operation_name, kwarg) @@ -66,35 +87,40 @@ class Test_EFS: access_analyzer = EFS(set_mocked_aws_provider()) assert access_analyzer.service == "efs" - @mock_aws + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) # Test EFS describe file systems def test_describe_file_systems(self): - efs_client = client("efs", AWS_REGION_EU_WEST_1) - efs = efs_client.create_file_system( - CreationToken=creation_token, - Encrypted=True, - Tags=[ - {"Key": "test", "Value": "test"}, - ], - ) - filesystem = EFS(set_mocked_aws_provider()) - assert len(filesystem.filesystems) == 1 - assert filesystem.filesystems[0].id == efs["FileSystemId"] - assert filesystem.filesystems[0].encrypted == efs["Encrypted"] - assert filesystem.filesystems[0].tags == [ + aws_provider = set_mocked_aws_provider() + efs = EFS(aws_provider) + efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}" + assert len(efs.filesystems) == 1 + assert efs.filesystems[efs_arn].id == FILE_SYSTEM_ID + assert efs.filesystems[efs_arn].encrypted + assert efs.filesystems[efs_arn].tags == [ {"Key": "test", "Value": "test"}, ] - @mock_aws - # Test EFS describe file systems + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + # Test EFS describe file systems policies def test_describe_file_system_policies(self): - efs_client = client("efs", AWS_REGION_EU_WEST_1) - efs = efs_client.create_file_system( - CreationToken=creation_token, Encrypted=True + aws_provider = set_mocked_aws_provider() + efs = EFS(aws_provider) + efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}" + assert len(efs.filesystems) == 1 + assert efs.filesystems[efs_arn].id == FILE_SYSTEM_ID + assert efs.filesystems[efs_arn].encrypted + assert efs.filesystems[efs_arn].backup_policy == "ENABLED" + assert efs.filesystems[efs_arn].policy == FILESYSTEM_POLICY + + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + # Test EFS describe mount targets + def test_describe_mount_targets(self): + aws_provider = set_mocked_aws_provider() + efs = EFS(aws_provider) + assert len(efs.filesystems) == 1 + efs_arn = f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{FILE_SYSTEM_ID}" + assert ( + efs.filesystems[efs_arn].mount_targets[0].file_system_id == FILE_SYSTEM_ID ) - filesystem = EFS(set_mocked_aws_provider()) - assert len(filesystem.filesystems) == 1 - assert filesystem.filesystems[0].id == efs["FileSystemId"] - assert filesystem.filesystems[0].encrypted == efs["Encrypted"] - assert filesystem.filesystems[0].backup_policy == backup_policy_status - assert filesystem.filesystems[0].policy == filesystem_policy + assert efs.filesystems[efs_arn].mount_targets[0].id == "fsmt-123" + assert efs.filesystems[efs_arn].mount_targets[0].subnet_id == "subnet-123"