feat(fsx): Add new service FSx (#5412)

Co-authored-by: Sergio <sergio@prowler.com>

prowler/providers/aws/services/fsx/fsx_client.py

@@ -0,0 +1,4 @@
+from prowler.providers.aws.services.fsx.fsx_service import FSx
+from prowler.providers.common.provider import Provider
+
+fsx_client = FSx(Provider.get_global_provider())

prowler/providers/aws/services/fsx/fsx_service.py

@@ -0,0 +1,70 @@
+from typing import Optional
+
+from pydantic import BaseModel
+
+from prowler.lib.logger import logger
+from prowler.lib.scan_filters.scan_filters import is_resource_filtered
+from prowler.providers.aws.lib.service.service import AWSService
+
+
+class FSx(AWSService):
+    def __init__(self, provider):
+        # Call AWSService's __init__
+        super().__init__(__class__.__name__, provider)
+        self.file_systems = {}
+        self.__threading_call__(self._describe_file_systems)
+
+    def _describe_file_systems(self, regional_client):
+        logger.info("FSx - Describing file systems...")
+        try:
+            describe_file_system_paginator = regional_client.get_paginator(
+                "describe_file_systems"
+            )
+            for page in describe_file_system_paginator.paginate():
+                for file_system in page["FileSystems"]:
+                    file_system_arn = f"arn:aws:fsx:{regional_client.region}:{self.audited_account}:file-system/{file_system['FileSystemId']}"
+                    if not self.audit_resources or (
+                        is_resource_filtered(file_system_arn, self.audit_resources)
+                    ):
+                        type = file_system["FileSystemType"]
+                        copy_tags_to_backups_aux = None
+                        copy_tags_to_volumes_aux = None
+                        if type == "LUSTRE":
+                            copy_tags_to_backups_aux = file_system.get(
+                                "LustreConfiguration", {}
+                            ).get("CopyTagsToBackups", False)
+                        elif type == "WINDOWS":
+                            copy_tags_to_backups_aux = file_system.get(
+                                "WindowsConfiguration", {}
+                            ).get("CopyTagsToBackups", False)
+                        elif type == "OPENZFS":
+                            copy_tags_to_backups_aux = file_system.get(
+                                "OpenZFSConfiguration", {}
+                            ).get("CopyTagsToBackups", False)
+                            copy_tags_to_volumes_aux = file_system.get(
+                                "OpenZFSConfiguration", {}
+                            ).get("CopyTagsToVolumes", False)
+
+                        self.file_systems[file_system_arn] = FileSystem(
+                            id=file_system["FileSystemId"],
+                            arn=file_system_arn,
+                            type=file_system["FileSystemType"],
+                            copy_tags_to_backups=copy_tags_to_backups_aux,
+                            copy_tags_to_volumes=copy_tags_to_volumes_aux,
+                            region=regional_client.region,
+                            tags=file_system.get("Tags", []),
+                        )
+        except Exception as error:
+            logger.error(
+                f"{regional_client.region} -- {error.__class__.__name__}[{error.__traceback__.tb_lineno}]: {error}"
+            )
+
+
+class FileSystem(BaseModel):
+    id: str
+    arn: str
+    region: str
+    type: str
+    copy_tags_to_backups: Optional[bool]
+    copy_tags_to_volumes: Optional[bool]
+    tags: Optional[list] = []

tests/providers/aws/services/fsx/fsx_service_test.py

@@ -0,0 +1,69 @@
+from boto3 import client
+from mock import patch
+from moto import mock_aws
+
+from prowler.providers.aws.services.fsx.fsx_service import FSx
+from tests.providers.aws.utils import (
+    AWS_ACCOUNT_NUMBER,
+    AWS_REGION_US_EAST_1,
+    set_mocked_aws_provider,
+)
+
+
+def mock_generate_regional_clients(provider, service):
+    regional_client = provider._session.current_session.client(
+        service, region_name=AWS_REGION_US_EAST_1
+    )
+    regional_client.region = AWS_REGION_US_EAST_1
+    return {AWS_REGION_US_EAST_1: regional_client}
+
+
+@patch(
+    "prowler.providers.aws.aws_provider.AwsProvider.generate_regional_clients",
+    new=mock_generate_regional_clients,
+)
+class Test_FSx_Service:
+    # Test FSx Service
+    def test_service(self):
+        aws_provider = set_mocked_aws_provider()
+        fsx = FSx(aws_provider)
+        assert fsx.service == "fsx"
+
+    # Test FSx Client
+    def test_client(self):
+        aws_provider = set_mocked_aws_provider()
+        fsx = FSx(aws_provider)
+        assert fsx.client.__class__.__name__ == "FSx"
+
+    # Test FSx Session
+    def test__get_session__(self):
+        aws_provider = set_mocked_aws_provider()
+        fsx = FSx(aws_provider)
+        assert fsx.session.__class__.__name__ == "Session"
+
+    # Test FSx Session
+    def test_audited_account(self):
+        aws_provider = set_mocked_aws_provider()
+        fsx = FSx(aws_provider)
+        assert fsx.audited_account == AWS_ACCOUNT_NUMBER
+
+    # Test FSx Describe File Systems
+    @mock_aws
+    def test_describe_file_systems(self):
+        fsx_client = client("fsx", region_name=AWS_REGION_US_EAST_1)
+        file_system = fsx_client.create_file_system(
+            FileSystemType="LUSTRE",
+            StorageCapacity=1200,
+            LustreConfiguration={"CopyTagsToBackups": True},
+            Tags=[{"Key": "Name", "Value": "Test"}],
+            SubnetIds=["subnet-12345678"],
+        )
+        arn = f"arn:aws:fsx:{AWS_REGION_US_EAST_1}:{AWS_ACCOUNT_NUMBER}:file-system/{file_system['FileSystem']['FileSystemId']}"
+        aws_provider = set_mocked_aws_provider([AWS_REGION_US_EAST_1])
+        fsx = FSx(aws_provider)
+        assert len(fsx.file_systems) == 1
+        assert fsx.file_systems[arn].id == file_system["FileSystem"]["FileSystemId"]
+        assert fsx.file_systems[arn].type == "LUSTRE"
+        assert fsx.file_systems[arn].copy_tags_to_backups
+        assert fsx.file_systems[arn].region == AWS_REGION_US_EAST_1
+        assert fsx.file_systems[arn].tags == [{"Key": "Name", "Value": "Test"}]