feat(efs): add new check efs_access_point_enforce_root_directory (#5277)

Co-authored-by: Sergio Garcia <38561120+sergargar@users.noreply.github.com>

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 <access-point-id> --root-directory Path=<non-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": ""
+}

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

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] = []

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']}"
+            )

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 == "/"