From aa3425a7dea2d9fb67299a46668165c8cd2da11d Mon Sep 17 00:00:00 2001 From: Mario Rodriguez Lopez <101330800+MarioRgzLpz@users.noreply.github.com> Date: Fri, 4 Oct 2024 19:12:47 +0200 Subject: [PATCH] feat(efs): add new check `efs_access_point_enforce_root_directory` (#5277) Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com> --- .../__init__.py | 0 ...point_enforce_root_directory.metadata.json | 34 +++++ ...efs_access_point_enforce_root_directory.py | 25 ++++ .../providers/aws/services/efs/efs_service.py | 38 +++++ ...ccess_point_enforce_root_directory_test.py | 138 ++++++++++++++++++ .../aws/services/efs/efs_service_test.py | 25 ++++ 6 files changed, 260 insertions(+) create mode 100644 prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/__init__.py create mode 100644 prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.metadata.json create mode 100644 prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.py create mode 100644 tests/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory_test.py diff --git a/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/__init__.py b/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/__init__.py new file mode 100644 index 0000000000..e69de29bb2 diff --git a/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.metadata.json b/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.metadata.json new file mode 100644 index 0000000000..b02871bdd3 --- /dev/null +++ b/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.metadata.json @@ -0,0 +1,34 @@ +{ + "Provider": "aws", + "CheckID": "efs_access_point_enforce_root_directory", + "CheckTitle": "EFS access points should enforce a root directory", + "CheckType": [ + "Software and Configuration Checks/AWS Security Best Practices" + ], + "ServiceName": "efs", + "SubServiceName": "access-point", + "ResourceIdTemplate": "arn:aws:elasticfilesystem:{region}:{account-id}:access-point/{access-point-id}", + "Severity": "medium", + "ResourceType": "AwsEfsAccessPoint", + "Description": "This control checks if Amazon EFS access points are configured to enforce a root directory. The control fails if the value of Path is set to / (the default root directory of the file system).", + "Risk": "Access points without enforced root directories can potentially expose the entire file system's root directory to clients, which may result in unauthorized access.", + "RelatedUrl": "https://docs.aws.amazon.com/config/latest/developerguide/efs-access-point-enforce-root-directory.html", + "Remediation": { + "Code": { + "CLI": "aws efs update-access-point --access-point-id --root-directory Path=", + "NativeIaC": "", + "Other": "https://docs.aws.amazon.com/securityhub/latest/userguide/efs-controls.html#efs-3", + "Terraform": "" + }, + "Recommendation": { + "Text": "Update the EFS access point to enforce a non-root directory. This ensures clients can only access a specified subdirectory.", + "Url": "https://docs.aws.amazon.com/efs/latest/ug/enforce-root-directory-access-point.html" + } + }, + "Categories": [ + "vulnerabilities" + ], + "DependsOn": [], + "RelatedTo": [], + "Notes": "" +} diff --git a/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.py b/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.py new file mode 100644 index 0000000000..8e7641cc9e --- /dev/null +++ b/prowler/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory.py @@ -0,0 +1,25 @@ +from prowler.lib.check.models import Check, Check_Report_AWS +from prowler.providers.aws.services.efs.efs_client import efs_client + + +class efs_access_point_enforce_root_directory(Check): + def execute(self): + findings = [] + for fs in efs_client.filesystems.values(): + if fs.access_points: + 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 access point allowing access to the root directory." + access_points = [] + for access_point in fs.access_points: + if access_point.root_directory_path == "/": + access_points.append(access_point) + if access_points: + report.status = "FAIL" + report.status_extended = f"EFS {fs.id} has access points which allow access to the root directory: {', '.join([ap.id for ap in access_points])}." + findings.append(report) + return findings diff --git a/prowler/providers/aws/services/efs/efs_service.py b/prowler/providers/aws/services/efs/efs_service.py index b3f8140e1e..c205487dad 100644 --- a/prowler/providers/aws/services/efs/efs_service.py +++ b/prowler/providers/aws/services/efs/efs_service.py @@ -20,6 +20,7 @@ class EFS(AWSService): self._describe_file_system_policies, self.filesystems.values() ) self.__threading_call__(self._describe_mount_targets, self.filesystems.values()) + self.__threading_call__(self._describe_access_points, self.filesystems.values()) def _describe_file_systems(self, regional_client): logger.info("EFS - Describing file systems...") @@ -113,6 +114,36 @@ class EFS(AWSService): f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" ) + def _describe_access_points(self, filesystem): + logger.info("EFS - Describing access points...") + try: + client = self.regional_clients[filesystem.region] + describe_access_point_paginator = client.get_paginator( + "describe_access_points" + ) + for page in describe_access_point_paginator.paginate( + FileSystemId=filesystem.id + ): + for access_point in page["AccessPoints"]: + access_point_id = access_point["AccessPointId"] + access_point_arn = access_point["AccessPointArn"] + if not self.audit_resources or ( + is_resource_filtered(access_point_arn, self.audit_resources) + ): + self.filesystems[filesystem.arn].access_points.append( + AccessPoint( + id=access_point_id, + file_system_id=access_point["FileSystemId"], + root_directory_path=access_point["RootDirectory"][ + "Path" + ], + ) + ) + except Exception as error: + logger.error( + f"{client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}" + ) + class MountTarget(BaseModel): id: str @@ -120,6 +151,12 @@ class MountTarget(BaseModel): subnet_id: str +class AccessPoint(BaseModel): + id: str + file_system_id: str + root_directory_path: str + + class FileSystem(BaseModel): id: str arn: str @@ -128,4 +165,5 @@ class FileSystem(BaseModel): backup_policy: Optional[str] = "DISABLED" encrypted: bool mount_targets: list[MountTarget] = [] + access_points: list[AccessPoint] = [] tags: Optional[list] = [] diff --git a/tests/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory_test.py b/tests/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory_test.py new file mode 100644 index 0000000000..f7d1fe2cf9 --- /dev/null +++ b/tests/providers/aws/services/efs/efs_access_point_enforce_root_directory/efs_access_point_enforce_root_directory_test.py @@ -0,0 +1,138 @@ +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_access_point_enforce_root_directory: + @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_access_point_enforce_root_directory.efs_access_point_enforce_root_directory.efs_client", + new=EFS(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_access_point_enforce_root_directory.efs_access_point_enforce_root_directory import ( + efs_access_point_enforce_root_directory, + ) + + check = efs_access_point_enforce_root_directory() + result = check.execute() + assert len(result) == 0 + + @mock_aws + def test_efs_no_access_point(self): + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + 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_access_point_enforce_root_directory.efs_access_point_enforce_root_directory.efs_client", + new=EFS(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_access_point_enforce_root_directory.efs_access_point_enforce_root_directory import ( + efs_access_point_enforce_root_directory, + ) + + check = efs_access_point_enforce_root_directory() + result = check.execute() + assert len(result) == 0 + + @mock_aws + def test_efs_access_point_default_root_directory(self): + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + access_point = efs_client.create_access_point( + FileSystemId=file_system["FileSystemId"], + PosixUser={"Uid": 1000, "Gid": 1000}, + RootDirectory={"Path": "/"}, + ) + + 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_access_point_enforce_root_directory.efs_access_point_enforce_root_directory.efs_client", + new=EFS(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_access_point_enforce_root_directory.efs_access_point_enforce_root_directory import ( + efs_access_point_enforce_root_directory, + ) + + check = efs_access_point_enforce_root_directory() + result = check.execute() + assert len(result) == 1 + assert result[0].status == "FAIL" + assert ( + result[0].status_extended + == f"EFS {file_system['FileSystemId']} has access points which allow access to the root directory: {access_point['AccessPointId']}." + ) + 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_access_point_enforced_root_directory(self): + efs_client = client("efs", region_name=AWS_REGION_US_EAST_1) + file_system = efs_client.create_file_system(CreationToken=CREATION_TOKEN) + + efs_client.create_access_point( + FileSystemId=file_system["FileSystemId"], + PosixUser={"Uid": 1000, "Gid": 1000}, + RootDirectory={"Path": "/notdefault"}, + ) + + 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_access_point_enforce_root_directory.efs_access_point_enforce_root_directory.efs_client", + new=EFS(aws_provider), + ): + from prowler.providers.aws.services.efs.efs_access_point_enforce_root_directory.efs_access_point_enforce_root_directory import ( + efs_access_point_enforce_root_directory, + ) + + check = efs_access_point_enforce_root_directory() + 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 access point allowing access to the root directory." + ) + 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_service_test.py b/tests/providers/aws/services/efs/efs_service_test.py index 4e85a9e04f..bcde323f93 100644 --- a/tests/providers/aws/services/efs/efs_service_test.py +++ b/tests/providers/aws/services/efs/efs_service_test.py @@ -55,6 +55,18 @@ def mock_make_api_call(self, operation_name, kwarg): } ] } + if operation_name == "DescribeAccessPoints": + return { + "AccessPoints": [ + { + "AccessPointId": "fsap-123", + "AccessPointArn": f"arn:aws:elasticfilesystem:{AWS_REGION_EU_WEST_1}:{AWS_ACCOUNT_NUMBER}:access-point/{FILE_SYSTEM_ID}/fsap-123", + "FileSystemId": FILE_SYSTEM_ID, + "RootDirectory": {"Path": "/"}, + "PosixUser": {"Uid": 1000, "Gid": 1000}, + } + ] + } if operation_name == "DescribeFileSystemPolicy": return {"FileSystemId": FILE_SYSTEM_ID, "Policy": json.dumps(FILESYSTEM_POLICY)} if operation_name == "DescribeBackupPolicy": @@ -124,3 +136,16 @@ class Test_EFS: ) assert efs.filesystems[efs_arn].mount_targets[0].id == "fsmt-123" assert efs.filesystems[efs_arn].mount_targets[0].subnet_id == "subnet-123" + + @patch("botocore.client.BaseClient._make_api_call", new=mock_make_api_call) + # Test EFS describe access points + def test_describe_access_points(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].access_points[0].file_system_id == FILE_SYSTEM_ID + ) + assert efs.filesystems[efs_arn].access_points[0].id == "fsap-123" + assert efs.filesystems[efs_arn].access_points[0].root_directory_path == "/"